Source code for pyuseocl.useengine

# coding=utf-8

"""
Wrapper to the USE engine. Call the 'use' command.
"""

__all__ = [
    'USEEngine',
]


import logging
logging.basicConfig(level=logging.DEBUG)
log = logging.getLogger('test.' + __name__)

import os
import tempfile
import operator
import re

#_RES_DIRECTORY = os.path.join(os.path.abspath(
#    os.path.dirname(__file__)),'res')
#_USE_DIRECTORY = os.path.join(_RES_DIRECTORY,'use-3.0.6')

[docs]class USEEngine(object): # USE_OCL_COMMAND = os.path.join(_USE_DIRECTORY,'bin','use') USE_OCL_COMMAND = 'use' """ Name of the use command. If the default value ('use') does not work, for instance if the use binary is not in the system path, you can change this value either in the source, or programmatically using USEEngine.USE_OCL_COMMAND = r'c:\Path\To\UseCommand\bin\use' """ command = None """ Last command executed by the engine""" commandExitCode = None """ Exit code of last execution""" out = None """ output of last execution (in case of separated out/err)""" err = None """ errors of last execution (in case of separated out/err)""" outAndErr = None """ output/errors of last execution if merged out/err """ @classmethod def __soilHelper(cls, name): return os.path.join( os.path.dirname(os.path.abspath(__file__)), 'res', name) @classmethod def __execute(cls, useFile, soilFile, errWithOut=False, executionDirectory=None): """ Execute use command with the given model and given soil file. The soil file MUST terminate by a 'quit' statement so that the process finish. The process is executed in the specified 'executionDirectory'. If not specified the execution directory is set to the directory of the use file given as a parameter. This directory could be important if the soil files contains references to relative path. This is in particular the case of 'open file.soil' statements. """ def readAndRemove(filename): with open(filename, 'r') as f: _ = f.read() # FIXME os.remove(filename) return _ if errWithOut: #-- one unique output file for output and errors (f, output_filename) = tempfile.mkstemp(suffix='.txt', text=True) os.close(f) errors_filename = None redirection = '>%s 2>&1' % output_filename cls.out = None cls.err = None else: # -- two temporary files for output and errors (f, output_filename) = tempfile.mkstemp(suffix='.use', text=True) os.close(f) # prepare temporary file for errors (f, errors_filename) = tempfile.mkstemp(suffix='.err', text=True) os.close(f) redirection = '>%s 2>%s' % (output_filename, errors_filename) cls.outAndErr = None commandPattern = '%s -nogui -nr %s %s '+ redirection cls.command = (commandPattern % (cls.USE_OCL_COMMAND, useFile, soilFile)) cls.directory = executionDirectory if executionDirectory is not None \ else os.path.dirname(os.path.abspath(useFile)) previousDirectory = os.getcwd() # Execute the command log.info('Execute USE OCL in %s: %s', cls.directory, cls.command) os.chdir(cls.directory) cls.commandExitCode = os.system(cls.command) os.chdir(previousDirectory) log.info(' execution returned %s exit code', cls.commandExitCode) if errWithOut: if cls.commandExitCode != 0: cls.outAndErr = None else: cls.outAndErr = readAndRemove(output_filename) else: cls.out = readAndRemove(output_filename) log.info (' with output of %s lines', len(cls.out.split('\n'))) # log.debug('----- output -----') # log.debug(cls.out) # log.debug('----- end of output ------') cls.err = readAndRemove(errors_filename) if len(cls.err) > 0: log.info( ' WITH ERRORS of %s lines: (first lines below)', len(cls.err.split('\n')) ) LINE_COUNT = 3 for err_line in cls.err.split('\n')[:LINE_COUNT]: if err_line != '': log.debug(' ERROR: %s',err_line) else: log.info( ' without anything in stderr' ) # log.debug('----- errors -----') # log.debug(cls.err) # log.debug('----- end of errors ------') return cls.commandExitCode @classmethod
[docs] def useVersion(cls): """ Try to get the version of use by executing it """ cls.__execute( cls.__soilHelper('emptyModel.use'), cls.__soilHelper('quit.soil')) first_line = cls.out.split('\n')[0] m = re.match( r'(use|USE) version (?P<version>[0-9\.]+),', first_line) if m: return m.group('version') else: msg = "Cannot execute USE OCL or get its version.\n" \ raise EnvironmentError('Cannot execute USE OCL or get its version. Is this program installed?')
@classmethod
[docs] def withUseOCL(cls): try: cls.useVersion() except EnvironmentError: return False else: return True
@classmethod
[docs] def analyzeUSEModel(cls, useFileName): cls.__execute( useFileName, cls.__soilHelper('infoModelAndQuit.soil')) return cls.commandExitCode
@classmethod
[docs] def checkSoilFileWithUSEModel(cls, modelFile, stateFile): driver_sequence = 'open %s \nquit' % stateFile (f, driver_filename) = tempfile.mkstemp(suffix='.soil', text=True) os.close(f) with open(driver_filename, 'w') as f: f.write(driver_sequence) cls.__execute( modelFile, driver_filename, errWithOut=False) return cls.commandExitCode
@classmethod
[docs] def evaluateSoilFilesWithUSEModel(cls, modelFile, stateFiles): def __generateSoilValidationDriver(stateFilePaths): """ Create a soil sequence with the necessary statements to drive the sequence of state validation. That is, it loads and checks each state one after each other. The soil driver sequence generated looks like: reset open file1.soil check reset open file2.soil check ... quit The output with error messages can be found after the execution in the variable outAndErr. :param stateFilePaths: A list of .soil files corresponding to states. :type stateFilePaths: [str] :return: The soil text :rtype: str """ if len(stateFiles) == 0: raise Exception('Error: no state file to evaluate') lines = reduce(operator.add, map( lambda file: ['reset', 'open ' + file, 'check -d'], stateFilePaths)) lines.append('quit') return '\n'.join(lines) #-- generate soil driver if len(stateFiles) == 0: raise Exception('Error: no state file to evaluate') driver_sequence = __generateSoilValidationDriver(stateFiles) (f, driver_filename) = tempfile.mkstemp(suffix='.soil', text=True) os.close(f) with open(driver_filename, 'w') as f: f.write(driver_sequence) cls.__execute( modelFile, driver_filename, errWithOut=True) return cls.commandExitCode