nelak's profile - activity

Here is some code I use, hopefully you can leverage portions of it to save yourself some time.

 def __ANSI(self, out_file_identifier, out_file_path, subsystem):
        ierr, busarray = self.psspy.abusint(subsystem, 1, 'NUMBER')
        if ierr: return ierr
        busarray = busarray[0]
        # skip rerun
        if not os.path.isfile(os.path.join(out_file_path, '%d_ANSI_rawData_tot.txt' % out_file_identifier)):
            ierr = self.psspy.report_output(2, os.path.join(out_file_path,
                                                            '%d_ANSI_rawData_tot.txt' % out_file_identifier))
            if ierr: return ierr
            vltary = [1.05 for x in busarray if x]
            cptary = [0.033 for x in busarray if x]
            # total current
            ierr = self.psspy.ansi_2([0, 0, 0, 5, 0], [40, 80, 40, 80], len(busarray), busarray, vltary, cptary, '')
            if ierr > 0:
                self.log('ERROR PERFORMING ANSI %s - Error code %s' % (out_file_identifier, ierr), 'error')
            ierr = self.psspy.report_output(2, os.path.join(out_file_path,
                                                            '%d_ANSI_rawData_sym.txt' % out_file_identifier))
            if ierr: return ierr
            vltary = [1.05 for x in busarray if x]
            cptary = [0.033 for x in busarray if x]
            # symmetric current
            ierr = self.psspy.ansi_2([0, 0, 0, 5, 1], [40, 80, 40, 80], len(busarray), busarray, vltary, cptary, '')
            if ierr > 0:
                self.log('ERROR PERFORMING ANSI %s - Error code %s' % (out_file_identifier, ierr), 'error')
        else:
            self.log('skipping %d_ANSI_rawData_tot.txt' % out_file_identifier)



 def RunANSI_study(self):
        self.log('Beginning ANSI portion of study...')
        # trim tp specific to those in study

        TP_specific_areas = list(set(int(x.area) for x in self.ansi_param))

        # create a default subsystem
        self.createSubsystemOfAreas(self.areas, sid=0)

        # find all buses in default subsystem
        study_buses = self.returnBusesInSubsystem(sid=0, in_service_only=False, bus_property='Number')

        # make list of buses manageable size for PSS/E report window
        self.log('  Segmenting %d study buses.' % len(study_buses))
        study_buses = self.splitter(study_buses, chunk=100)

        # loop thru lists of buses, create subsystems, and perform ANSI on them
        self.log('  Created %d segments of size %d.' % (len(study_buses), len(study_buses[0])))
        for i, segment_of_study_buses in enumerate(study_buses, 1):
            self.createSubsystemOfBuses(segment_of_study_buses, 0)
            self.log('    Running default ANSI for segment %d' % i)
            self.__ANSI(i, self.report_dir, 0)
        self.log('ANSI study complete.')

    def processANSIresults(self, workbook, folder):
        """

        :param workbook: Workbook to add results.
        :param folder: Location of psse reports.
        :return: Workbook with results included.
        """
        if not hasattr(self, 'psspy'):
            #self.init_psse()
            self.init_psse()
            #self.PSSEService.init_psse()
            self.openCase(self.model_path)

        ansi_results_sym, ansi_results_tot, ansi_tpspec = self.__compileANSIresults(folder)
        workbook = self.__writeANSIresults(workbook, ansi_results_sym, 'Default-ANSI-Symm Curr Basis')
        workbook = self.__writeANSIresults(workbook, ansi_results_tot, 'Default-ANSI-Total Curr Basis')
        if ansi_tpspec:
            workbook = self.__writeANSIresults(workbook, ansi_tpspec, 'TP-Criteria ANSI')
        return workbook

    def __compileANSIresults(self, folder):
        """ Parses a folder and build ANSI results from PSSE ANSI report txt files."""
        # get list of report filepaths
        list_of_ansi_reports = glob.glob(os.path.join(folder, '*ANSI_rawData*.txt'))
        self.log('Read in %d ANSI report files.' % len(list_of_ansi_reports))
        list_of_ansi_reports_sym = [x for x in list_of_ansi_reports if 'sym' in os.path.basename(x)]
        list_of_ansi_reports_tot = [x for x in list_of_ansi_reports if 'tot' in os.path.basename(x)]
        list_of_ansi_reports_tp = [x for x in list_of_ansi_reports if 'TPSPEC' in os.path.basename(x)]

        # file in memory
        data = []
        for report in list_of_ansi_reports_sym:
            data_file = open(report, 'rU')
            data.extend(data_file.readlines())
        # parse ...

Approximation of how I do this:

Running Short circuit and building lots of text files:

def splitter(self, inputlist, chunk=100):
        """ Breaks inputlist into a list of lists with size chunk."""
        chunks = [inputlist[x:x + chunk] for x in range(0, len(inputlist), chunk)]
        return chunks

def RunASCC_study(self, options):
        ascc_results = []

    # create subsystem of areas in user input
    self.createSubsystemOfAreas(options['Areas'])

    # find all buses in study areas
    study_buses = self.returnBusesInSubsystem(sid=0, in_service_only=False, bus_property='Number')

    # make list of buses manageable size for PSS/E report window
    self.log('Segmenting %d study buses.' % len(study_buses))
    study_buses = self.splitter(study_buses, chunk=100)

    # loop thru lists of buses, create subsystems, and perform ASCC on them
    self.log('Created %d segments of size %d.\nPerforming ASCC portion of study.' % (
             len(study_buses), len(study_buses[0])))
    for i, segment_of_study_buses in enumerate(study_buses, 1):
        self.createSubsystemOfBuses(segment_of_study_buses, sid=0)
        self.log('Running ASCC for segment %d' % i)
        self.__ASCC(i, self.report_directory, 0)

def __ASCC(self, out_file_identifier, out_file_path, subsystem, gen_scfiles=False):
    """
    :param out_file_identifier: increments file names
    :param out_file_path: gives folder to place reports in
    :param subsystem: specifies subsystem id for psspy.ascc api
    :param gen_scfiles: option flag for scfile output of psspy.ascc api
    :return:
    """
    ierr = self.psspy.report_output(2, os.path.join(out_file_path, '%d_ASCC_rawData.txt' % out_file_identifier))
    if ierr: self.log('Error on ASCC segment %s ier %s' % (out_file_identifier, ierr), 'error')
    # class fault analysis conditions
    # ierr = self.psspy.flat_2([1, 1, 1, 2, 1, 2, 1, 3], [1, 1])

    if gen_scfiles:
        if not os.path.exists(os.path.join(out_file_path, 'scfiles')):
            os.makedirs(os.path.join(out_file_path, 'scfiles'))
        scfile_path = os.path.join(out_file_path, 'scfiles', '%d_scfile.sc' % out_file_identifier)
    else:
        scfile_path = ""

    ierr = self.psspy.ascc_3(subsystem, 0, [1, 1, 0, 1, 0, 2, 0, 1, 0, 1, 1, 0, 0, 2, 2, 1, 1], [1],
                                 '', '', scfile_path)
    if ierr: self.log('Error on ASCC segment %s' % out_file_identifier, 'error')

Parsing text files:

def __compileASCCresults(self, folder):
    """
    Parses a folder and build ASCC results from PSSE ASCC report txt files.
    :param folder: Directory of containing psse reports.
    :return: List of result objects.
    """

    # get list of report filepaths
    list_of_ascc_reports = []
    for txtfile in os.listdir(folder):
        if txtfile.endswith("ASCC_rawData.txt"):
            list_of_ascc_reports.append(os.path.join(folder, txtfile))

    # file in memory
    data = []
    for report in list_of_ascc_reports:
        data_file = open(report, 'rU')
        data.extend(data_file.readlines())

    # parse file
    list_of_result_objects = []
    options = []
    while data:
        line = data.pop(0)
        line = line.rstrip().rstrip('\r\n')
        # build options
        if line.upper().startswith('OPTIONS USED'):
            # only need options once because its the same
            if not options:
                options.append(line)
                line = data.pop(0).rstrip().rstrip('\r\n')
                while line:
                    options.append(line.split('-', 1)[1])
                    line = data.pop(0).rstrip().rstrip('\r\n')
            else:
                pass
        # build fault result objects
        temp_list = []
        if line.startswith('AT BUS'):
            while line != "-----------------------------------------------------------------------------------------------------------------------------------------------":
                temp_list.append(line)
                line = data.pop(0).rstrip().rstrip('\r\n')
            list_of_result_objects.append(self.ASCCfaultResult(temp_list, options, self.log))
    self.log('Read in %d ASCC data files.' % len(list_of_ascc_reports))
    return list_of_result_objects

class ASCCfaultResult():
    """ Creates an object from a list of text lines ...