Source code for mojo.pavayo.computeMethods

# -*- coding: utf-8 -*-
"""
computeMethods
==============
Module containing methods for compute.py
"""


import datetime
import os
import sys
import time

from .. import RELEASE_NUMBER
from ..bricabrac import fileIO
from ..latex import PyTex

from .. import pavayo
from . import computeData as constants
from .executables.executableResources import ExecutableResources
from .executables.version import getSoftwareVersionNumberString
from .testcase.testcase_generated import SuccessFailStatement


[docs]def getMOJOPythonPath(): """Builds the path to the MOJO src folder. :return: absolute path to MOJOs source folder :rtype: str """ return "/".join(str(pavayo.__file__).split("/")[:-3])
[docs]def getSoftwaresVersionNumbers(options, executableDict, readVersionsFromExecutables=False): """Reads the software version numbers by calling <SOFTWARE> --version or reading the version from a options file. Returns a dictionary which maps the software to its version. :param options: options parsed from the command line :param executableDict: dictionary of executables :param readVersionsFromExecutables: if true the method reads the version from the options file instead of calling the executables :type options: argParse.ArgumentParser :type executableDict: ExecutableResources :type readVersionsFromExecutables: bool :return: a dictionary containing the software versions mapped to the software name :rtype: dict """ if readVersionsFromExecutables: for key in sorted(executableDict.resourceDict): if executableDict[key].displayVersion: options.configFile.set("softwareVersions", key.lower(), str(executableDict[key].versionSettings.version)) with open(options.configFilePath, "w") as fileHandle: options.configFile.write(fileHandle) else: options.configFile.read(options.configFilePath) return dict(options.configFile.items("softwareVersions"))
def convertWallTimeToWeight(elt, weightDefault): weight = weightDefault if hasattr(elt, 'wallTime') and elt.wallTime is not None: for timeFormat in ['%H:%M:%S', '%M:%S', '%S']: try: x = time.strptime(elt.wallTime.split(',')[0], timeFormat) except ValueError: pass else: weight = datetime.timedelta(hours=x.tm_hour, minutes=x.tm_min, seconds=x.tm_sec).total_seconds() break return weight
[docs]def convertErrorCodeToString(errorCode): """Gets an error code and returns the related string :param errorCode: error code :type errorCode: int :return: related text message :rtype: str """ successState = "successfully" if errorCode: successState = "NOT successfully" return successState
[docs]def processTraceSuitePath(options, executableDict): """Sets the TRACE, PREP and POST path from the path to the trace suite. :param options: options parsed from the command line :param executableDict: dictionary of executables :type options: argParse.ArgumentParser :type executableDict: ExecutableResources """ searchPathDict = {constants.TRACE_SUITE_TRACE_EXEC: os.path.join(options.traceSuitePath, "*/TRACE*"), constants.TRACE_SUITE_PREP_EXEC: os.path.join(options.traceSuitePath, "*/PREP*"), constants.TRACE_SUITE_POST_EXEC: os.path.join(options.traceSuitePath, "*/POST*"), } replaceSuffix = constants.TRACE_SUITE_DEBUG_NAME if options.useDebugExecs else '' for key, searchPath in searchPathDict.items(): executableDict.addMultipleExecutables(searchPath, environmentName=key, displayVersion=True, successStatement=[SuccessFailStatement(valueOf_=constants.SUCCESS_STATUS[key]), ], replaceSuffix=replaceSuffix) traceSuiteExecs = (constants.TRACE_SUITE_TRACE_EXEC, constants.TRACE_SUITE_PREP_EXEC, constants.TRACE_SUITE_POST_EXEC,) if not all(k in executableDict.resourceDict for k in traceSuiteExecs): raise RuntimeError(f"TRACE suite executables are missing. Perhaps you forgot to compile the TRACE suite ('{options.traceSuitePath}') or are using the '--debug' option wrong?") if options.useDebugExecs and not (executableDict[constants.TRACE_SUITE_TRACE_EXEC].path.endswith(constants.TRACE_SUITE_DEBUG_NAME) and executableDict[constants.TRACE_SUITE_PREP_EXEC].path.endswith(constants.TRACE_SUITE_DEBUG_NAME) and executableDict[constants.TRACE_SUITE_POST_EXEC].path.endswith(constants.TRACE_SUITE_DEBUG_NAME)): raise RuntimeError("'--debug' flag specified but no debug executables found for TRACE, PREP or POST!\n")
[docs]def createResourceDictionary(options, listOfExecutables): """Create a dictionary of all executables read from the XML file :param options: options parsed from the command line :param listOfExecutables: list of executables read from the XML file :type options: argParse.ArgumentParser :type listOfExecutables: ResourcesSub :return: dictionary of executables :rtype: ExecutableResources """ def _addExecutableToDict(executableDict, executableName, executablePath, maxProcsPerExecution, commandLineArguments=None, versionCheckCommand=None): """Add executable to dictionary :param executableDict: list of executables :param executableName: name of executable :param executablePath: path to executable :param maxProcsPerExecution: maximum number of processors per execution :param commandLineArguments: additional command line arguments :param versionCheckCommand: command line argument for version check :type executableDict: ExecutableResources :type executableName: str :type executablePath: str :type maxProcsPerExecution: Optional[int] :type commandLineArguments: Optional[list(str)] :type versionCheckCommand: Optional[str] """ successFailStatement = None if executableName in constants.SUCCESS_STATUS: successFailStatement = [SuccessFailStatement(valueOf_=constants.SUCCESS_STATUS[executableName])] executableDict.addExecutable(environmentName=executableName, path=executablePath, numberOfLicenses=None, maxProcsPerExecution=maxProcsPerExecution, displayVersion=True, successStatement=successFailStatement, failStatement=None, ignoreExitCode=False, commandLineArguments=commandLineArguments, versionCheckCommand=versionCheckCommand) # dictionary of basic __replaceExecutables = {constants.RM_TEMPLATE: fileIO.globRm, constants.MV_TEMPLATE: fileIO.globMove, constants.MD_TEMPLATE: fileIO.md, constants.CP_TEMPLATE: fileIO.globCopy, constants.LN_TEMPLATE: fileIO.forceSymlink, constants.PYTHON_TEMPLATE: constants.PYTHON_EXECUTABLE, constants.TECPLOT_TEMPLATE: constants.TECPLOT_EXECUTABLE, constants.TECPLOT_MACRO_TEMPLATE: constants.TECPLOT_EXECUTABLE} __versionKeyTagsExecs = {constants.PYTHON_TEMPLATE: constants.PYTHON_EXECUTABLE} executableDict = ExecutableResources() # set internal defaults for key, var in __replaceExecutables.items(): executableDict.addExecutable(environmentName=key, path=var, maxProcsPerExecution=1) for environmentName, versionKey in __versionKeyTagsExecs.items(): execKey, _ = ExecutableResources.getKeyAndEnvironmentName(environmentName) executableDict[execKey].versionSettings.key = versionKey # add settings stored in the XML file if listOfExecutables is not None: executableDict.updateDictFromXML(listOfExecutables, options.basePathDict) # update executable given as options if options.traceSuitePath: processTraceSuitePath(options, executableDict) if options.tracePath: _addExecutableToDict(executableDict, constants.TRACE_SUITE_TRACE_EXEC, options.tracePath, None) if options.prepPath: _addExecutableToDict(executableDict, constants.TRACE_SUITE_PREP_EXEC, options.prepPath, None) if options.postPath: _addExecutableToDict(executableDict, constants.TRACE_SUITE_POST_EXEC, options.postPath, None) if os.path.exists(options.gmcPlayPath): _addExecutableToDict(executableDict, constants.GMC_PLAY_EXEC, options.gmcPlayPath, 1) if options.pyMeshPath and (os.path.exists(options.pyMeshPath) or options.pyMeshPath != constants.PYMESH_EXEC): try: pyMeshHomePath = os.path.abspath(os.path.join(options.pyMeshPath, constants.PYMESH_PATH_BIN_TO_HOME)) except IndexError: pyMeshHomePath = "<dummy>" _addExecutableToDict(executableDict, constants.PYMESH_EXEC_KEY, options.pyMeshPath, 1, versionCheckCommand="--versionShort", commandLineArguments=["--pm-noColor", "--pm-nThreads", "0", "--pm-home", pyMeshHomePath]) if options.additionalExecutables: for key, var in options.additionalExecutables.items(): executableDict.addExecutable(environmentName=key, path=var, displayVersion=True, explicitlyRequested=True) for execName, argument in options.commandLineArguments: executableDict[execName.upper()].commandLineArguments.append(" --" + argument) if constants.TRACE_SUITE_PATH_ENVIRONMENT_NAME in options.basePathDict: executableDict.mainExecutable = constants.TRACE_SUITE_TRACE_EXEC for key in (constants.TECPLOT_TEMPLATE, constants.TECPLOT_MACRO_TEMPLATE,): executableDict[key].versionSettings.args = "-v" executableDict.getVersion(options=options) resourcesDict = {key: int(executable.numberOfLicenses.valueOf_) for key, executable in executableDict if executable.numberOfLicenses} return executableDict, resourcesDict
[docs]def createGeneralTitlePageContent(executableDict, suiteName, userName, hostName, svnRepoPath=None, svnRevision=None, runTime=None): """Create the content of the title page for the report. :param executableDict: dictionary of executables :param suiteName: Name of the test suite specified in the xml file :param userName: user name :param hostName: host name :param svnRepoPath: path of SVN repository :param svnRevision: SVN revision number :param runTime: duration of current run :type executableDict: ExecutableResources :type suiteName: str :type userName: str :type hostName: str :type svnRepoPath: str :type svnRevision: int :type runTime: str :return: content of title page :rtype: str """ def escape(path): return PyTex.escapeChars(str(path)) mojoPath = os.path.abspath(os.path.join(pavayo.__file__, "../..")) mojoVersion = RELEASE_NUMBER pythonPath = os.environ.get("PYTHONPATH", "-") pythonVersion = "{version.major}.{version.minor}.{version.micro}".format(version=sys.version_info) content = [constants.PAVAYO_REPORT_TITLE_PAGE_HEADER_START_FMT.format(suiteName=escape(suiteName))] if userName and hostName: content.append(constants.PAVAYO_REPORT_TITLE_PAGE_USER_SEQ_FMT.format(userName=escape(userName), hostName=escape(hostName))) content.append(constants.PAVAYO_REPORT_TITLE_PAGE_DATE_SEQ_FMT) content.append(constants.PAVAYO_REPORT_HLINE) try: traceExec = executableDict[constants.TRACE_SUITE_TRACE_EXEC] content.append(constants.PAVAYO_REPORT_TITLE_PAGE_TRACE_VERSION.format(version=traceExec.version)) content.append(constants.PAVAYO_REPORT_HLINE) except KeyError: pass content.append(constants.PAVAYO_REPORT_TITLE_PAGE_HEADER_END_SEQ) content.append(constants.PAVAYO_REPORT_TITLE_PAGE_VERSION_SECTION_START_SEQ) def addSoftwareVersionEntry(name, version, path): content.append(constants.PAVAYO_REPORT_TITLE_PAGE_VERSION_ENTRY_FMT.format(name=name, version=escape(version), path=escape(path))) if svnRepoPath is not None and svnRevision is not None: addSoftwareVersionEntry("Testcases SVN", str(svnRevision), str(svnRepoPath)) addSoftwareVersionEntry("MOJO", mojoVersion, mojoPath) addSoftwareVersionEntry("Python", pythonVersion, sys.executable) addSoftwareVersionEntry("Tecplot", getSoftwareVersionNumberString(constants.TECPLOT_EXECUTABLE, versionArgument = '-v'), fileIO.getAbsoluteFilePath(constants.TECPLOT_EXECUTABLE)) try: import matplotlib addSoftwareVersionEntry("Matplotlib", matplotlib.__version__, os.path.dirname(matplotlib.__file__)) except ImportError as e: addSoftwareVersionEntry("Matplotlib", "N/A", str(e)) try: gmcPlayExec = executableDict[constants.GMC_PLAY_EXEC] addSoftwareVersionEntry("GMCplay", gmcPlayExec.version, gmcPlayExec.path) except KeyError: pass try: pyMeshExec = executableDict[constants.PYMESH_EXEC_KEY] addSoftwareVersionEntry("PyMesh", pyMeshExec.version, pyMeshExec.path) except KeyError: pass for key in sorted(executableDict.resourceDict): value = executableDict[key] if value.displayVersion: addSoftwareVersionEntry(key, str(value.version), value.path) content.append(constants.PAVAYO_REPORT_TITLE_PAGE_VERSION_SECTION_END_SEQ) content.append(constants.PAVAYO_REPORT_TITLE_PAGE_ENV_SECTION_START_SEQ) def addEnvVariableEntry(var, value): value = value.replace(":", ": \\newline") content.append(constants.PAVAYO_REPORT_TITLE_PAGE_ENV_ENTRY_FMT.format(variable=escape(var), value=escape(value))) addEnvVariableEntry("PYTHONPATH", pythonPath) content.append(constants.PAVAYO_REPORT_TITLE_PAGE_ENV_SECTION_END_SEQ) if runTime is None: runTime = "-" content.append(constants.PAVAYO_REPORT_TITLE_PAGE_END_SEQ.format(runTime=runTime)) return "\n".join(content)
[docs]def createValidationTitlePageContent(executableDict, basePathDict, reportSettings=None): """Create the content of the title page for the report. :param executableDict: dictionary of executables :param basePathDict: base paths :param reportSettings: cumulated settings for the report :type executableDict: ExecutableResources :type basePathDict: dict :type reportSettings: object :return: content of title page :rtype: str """ def _getPicturePath(logoPath, basePathDictInt): if logoPath.type_ == "relative": try: searchPath = os.path.abspath(os.path.join(basePathDictInt[logoPath.basePath], logoPath.valueOf_)) except KeyError: searchPath = os.path.join(logoPath.basePath, logoPath.valueOf_) else: searchPath = logoPath.valueOf_ return searchPath content = list() if reportSettings and reportSettings.logo: content.append(constants.VALIDATION_REPORT_TITLE_PAGE_PICTURE_START_SEQ) nLogos = len(reportSettings.logo) if nLogos == 1: xPos = 0.7 width = 0.3 content.append(constants.VALIDATION_REPORT_TITLE_PAGE_PICTURE_SEQ_FMT.format(xPos=xPos, width=width, logoPath=_getPicturePath(reportSettings.logo[0], basePathDict))) elif nLogos > 1: frameWidth = 1 / nLogos width = frameWidth * 0.9 for i, logo in enumerate(reportSettings.logo): xPos = i * frameWidth content.append(constants.VALIDATION_REPORT_TITLE_PAGE_PICTURE_SEQ_FMT.format(xPos=xPos, width=width, logoPath=_getPicturePath(logo, basePathDict))) content.append(constants.VALIDATION_REPORT_TITLE_PAGE_PICTURE_END_SEQ) content.append(constants.VALIDATION_REPORT_TITLE_PAGE_TITLE_START_SEQ) if executableDict.mainExecutable: version = executableDict[executableDict.mainExecutable].version content.append(constants.VALIDATION_REPORT_TITLE_PAGE_TITLE_VERSION_SEQ_FMT.format(version=version)) content.append(constants.VALIDATION_REPORT_TITLE_PAGE_TITLE_MID_SEQ) if reportSettings and reportSettings.address: content.append(constants.VALIDATION_REPORT_TITLE_PAGE_TITLE_ADDRESS_SEQ_FMT.format(address=reportSettings.getAddressInLaTex())) content.append(constants.VALIDATION_REPORT_TITLE_PAGE_TITLE_END_SEQ) if reportSettings and reportSettings.address: content.append(constants.VALIDATION_REPORT_TITLE_PAGE_DATE_FMT.format(city=reportSettings.address.city)) return "\n".join(content)