testReadProjectFile.py - master - afedorov1971/apkdk-tests

testReadProjectFile.py
201 строка7 KB
Alexander Fedorov
generate reports in container
25 окт 2019, 13:23
25 окт 2019, 13:2302a62f7
import os
import zipfile
import collections
import xlsxwriter
import argparse
import xmltodict


def getparsedxml(zipFileName, entryName):
    with zipfile.ZipFile(zipFileName) as zf:
        with zf.open(entryName) as projectContentEntry:
            return xmltodict.parse(projectContentEntry.read())


def get_project_file_entries(zipFileName):
    with zipfile.ZipFile(zipFileName) as zf:
        return zf.namelist()


def make_list(listOrSingle):
    if listOrSingle is None:
        return []
    return [listOrSingle] if isinstance(listOrSingle, collections.OrderedDict) else listOrSingle


def makedict(dict):
    return collections.OrderedDict() if dict is None else dict


def getEntry(dict, key):
    return None if dict is None or not key in dict else dict[key]


def addHeaders(worksheet: object, headers: object) -> object:
    col = 0
    for item in headers:
        worksheet.write(0, col, item)
        col = col + 1


def writeToExcel(filename, objectslist, failureslist, parameterslist):
    workbook = xlsxwriter.Workbook(filename)
    worksheetParameters = workbook.add_worksheet('ControlParams')
    worksheetFailures = workbook.add_worksheet('ControlDiagStates')
    worksheetObjects = workbook.add_worksheet('ControlObjects')

    addHeaders(worksheetParameters, ['Obj_Osn_Id', 'Control_Object_ID', 'Control_Param_ID',
                                     'UniParam_Type_ID', 'Name_Param', 'Description_Param',
                                     'Normal_Max', 'Normal_Min', 'Normal_Ext'])

    row = 1
    for item in parameterslist:
        worksheetParameters.write(row, 0, int(item.uniplaceid))
        worksheetParameters.write(row, 1, int(item.id))
        worksheetParameters.write(row, 2, int(item.controlParamId))
        worksheetParameters.write(row, 3, int(item.uniParamTypeId))
        worksheetParameters.write(row, 4, item.name)
        worksheetParameters.write(row, 5, item.description)
        worksheetParameters.write(row, 6, 0)
        worksheetParameters.write(row, 7, 0)
        worksheetParameters.write(row, 8, 0)
        row += 1

    addHeaders(worksheetFailures, ['Obj_Osn_Id', 'Control_Object_ID', 'Uni_DiagState_ID'])

    row = 1
    for item in failureslist:
        worksheetFailures.write(row, 0, int(item.uniplaceid))
        worksheetFailures.write(row, 1, int(item.id))
        worksheetFailures.write(row, 2, int(item.unidiagstateid))
        row += 1

    addHeaders(worksheetObjects, ['Obj_Osn_Id', 'Control_Object_ID',
                                  'Uni_Type_ID', 'Name_Object', 'Name_Park', 'Description'])

    row = 1
    for item in objectslist:
        worksheetObjects.write(row, 0, int(item.uniplaceid))
        worksheetObjects.write(row, 1, int(item.id))
        worksheetObjects.write(row, 2, int(item.unitypeid))
        worksheetObjects.write(row, 3, item.name)
        worksheetObjects.write(row, 5, item.description)
        row += 1

    workbook.close()


class ObjectResult:
    def __init__(self, id, name, uniplaceid, unitypeid, description: str):
        self.name = name
        self.id = id
        self.unitypeid = unitypeid
        self.uniplaceid = uniplaceid
        self.description = description


class DiagStateResult:
    def __init__(self, id, uniplaceid, unidiagstateid):
        self.id = id
        self.uniplaceid = uniplaceid
        self.unidiagstateid = unidiagstateid


class ControlParamResult:
    def __init__(self, id, uniplaceid, paramId, uniparamtypeid, name, description):
        self.id = id
        self.uniplaceid = uniplaceid
        self.controlParamId = paramId
        self.uniParamTypeId = uniparamtypeid
        self.name = name
        self.description = description


def get_env_or_default(cur_value: str, key: str, default_value: str):
    if not cur_value is None:
        return cur_value

    env_value = os.getenv(key)
    if not env_value is None:
        return env_value
    return default_value

parser = argparse.ArgumentParser(description='Генерация файла отчета для УП по файлу проекта')
parser.add_argument('--project', help='Имя файла проекта')
parser.add_argument('--output', help='Имя файла отчета')

args = parser.parse_args()

projectFileName = get_env_or_default(args.project, "PROJECTFILE", None)
if projectFileName is None:
    print('Не задано имя файла проекта')
    exit(-1)

outputFileName = get_env_or_default(args.output, "OUTPUTFILE", 'output.xlsx')

if outputFileName is None:
    print('Не указано имя файла отчета')
    exit(-1)

entries = get_project_file_entries(projectFileName)
# должен быть единственный файл с расширением prj - файл проекта
projectEntries = [item for item in entries if item.endswith('.prj')]

if len(projectEntries) != 1:
    print('Некорректный архив. Не найден файл проекта.')
    exit(-1)

xmldoc = getparsedxml(projectFileName, projectEntries[0])


failures = make_list(getEntry(xmldoc['Project']['FailureTypes'], 'FailureType'))

measures = make_list(getEntry(xmldoc['Project']['ObjectParameterTypes'], 'ObjectParameter'))

objects = make_list(getEntry(xmldoc['Project']['Objects'], 'Object'))

uniPlaces = make_list(getEntry(xmldoc['Project']['UniProtocol']['UniPlaces'], 'UniPlace'))

object2name = {}

object2description = {}

for item in objects:
    object2name[item['@Id']] = item['@Name']
    object2description[item['@Id']] = item.get('@Description', '')

measure2Name = {}

MeasureInfo = collections.namedtuple('MeasureInfo', 'name shortName')

for item in measures:
    measure2Name[item['@Id']] = MeasureInfo(name=item['@Name'], shortName=getEntry(item, '@ShortName'))

objectsResult = []
diagStatesResult = []
controlParamsResult = []

for item in uniPlaces:
    for uo in make_list(item['UniObjects']['UniObject']):
        objectId = uo['@ObjectId']
        name = object2name[objectId]
        description = object2description[objectId]
        placeId = item['@UniPlaceId']
        objectsResult.append(ObjectResult(objectId, name, placeId, uo['@UniTypeId'], description))

        for ods in make_list(getEntry(uo['UniDiagStates'], 'UniDiagState')):
            diagStatesResult.append(DiagStateResult(objectId, placeId, ods['@UniDiagStateId']))

        for cp in make_list(getEntry(uo['UniParameters'], 'UniParameter')):
            measureInfo = measure2Name[cp['@MeasureTypeId']];
            # Имя измерения состоит из имени объекта и короткого имени
            measureName = '{0}{1}'.format(name, '' if measureInfo.shortName is None else measureInfo.shortName)
            description = measureInfo.name
            controlParamsResult.append(ControlParamResult(objectId, placeId,
                                                          cp['@ControlParamId'],
                                                          cp['@UniParamTypeId'],
                                                          measureName, description))

writeToExcel(outputFileName, objectsResult, diagStatesResult, controlParamsResult)

print('Отчет записан в файл {0}'.format(outputFileName))