Core transform class. More...

Inheritance diagram for python.transform.transform:

Collaboration diagram for python.transform.transform:

Public Member Functions
	__init__ (self, standardSignalHandlers=True, standardTrfArgs=True, standardValidationArgs=True, trfName=None, executor=None, exeArgs=None, description='')
	Initialise a job transform.
	name (self)
	exitCode (self)
	exitMsg (self)
	argdict (self)
	dataDictionary (self)
	report (self)
	transformStart (self)
	transformSetupCpuTime (self)
	transformSetupWallTime (self)
	inFileValidationCpuTime (self)
	inFileValidationWallTime (self)
	outFileValidationCpuTime (self)
	outFileValidationWallTime (self)
	outFileValidationStop (self)
	trfPredata (self)
	executors (self)
	processedEvents (self)
	getProcessedEvents (self)
	appendToExecutorSet (self, executors)
	parseCmdLineArgs (self, args)
	Parse command line arguments for a transform.
	setGlobalLogLevel (self)
	Check transform argument dictionary and set the correct root logger option.
	execute (self)
	Execute transform.
	setupSplitting (self)
	Setup executor splitting.
	lastExecuted (self)
	Return the last executor which actually executed.
	generateReport (self, reportType=None, fast=False, fileReport=defaultFileReport)
	Transform report generator.
	updateValidationDict (self, newValidationOptions)
	Setter for transform's validation dictionary.
	getValidationDict (self)
	Getter function for transform validation dictionary.
	getValidationOption (self, key)
	Getter for a specific validation option.
	getFiles (self, io=None)
	Return a list of fileArgs used by the transform.
	validateInFiles (self)
	validateOutFiles (self)

Public Attributes
	parser
	validation
	name

Protected Member Functions
	_setupGraph (self)
	Setup the executor graph.
	_tracePath (self)
	Trace the path through the executor graph.
	_doSteering (self, steeringDict=None)
	Setup steering, which manipulates the graph before we trace the path for this transform.
	_exitWithReport (self, signum, frame)
	Common signal handler.

Protected Attributes
	_transformStart = os.times()
	Get transform starting timestamp as early as possible.
	_inFileValidationStart = None
	_inFileValidationStop = None
	_outFileValidationStart = None
	_outFileValidationStop = None
	_trfPredata = os.environ.get('TRF_PREDATA')
	Get trf pre-data as early as possible.
	_name = trfName or path.basename(sys.argv[0]).rsplit('.py', 1)[0]
	Transform _name.
	_argdict = dict()
	Argument dictionary for this transform.
	_dataDictionary = dict()
	Dsta dictionary place holder (this maps data types to their argFile instances).
	_executors = set()
dict	_executorDictionary = {}
int	_exitCode = None
	Transform exit code/message holders.
str	_exitMsg = None
	_report = trfJobReport(parentTrf = self)
	Report object for this transform.
	_processedEvents = None
	Transform processed events.
	_exitWithReport
	_executorPath
	_executorGraph
	_inputData = list()
	_outputData = list()

Detailed Description

Core transform class.

Note: Every transform should only have one transform class instantiated

Definition at line 40 of file transform.py.

Constructor & Destructor Documentation

◆ init()

python.transform.transform.__init__	(	self,
		standardSignalHandlers = True,
		standardTrfArgs = True,
		standardValidationArgs = True,
		trfName = None,
		executor = None,
		exeArgs = None,
		description = '' )

Initialise a job transform.

Parameters

standardSignalHandlers	Boolean to set signal handlers. Default `True`.
standardValidationArgs	Boolean to set standard validation options. Default `True`.
trfName	Name of the transform. Default is executable name with .py rstripped.
executor	Executor list

Transform class initialiser

Definition at line 47 of file transform.py.

                 trfName = None, executor = None, exeArgs = None, description = ''):
        '''Transform class initialiser'''
        msg.debug('Welcome to ATLAS job transforms')
        
        ## @brief Get transform starting timestamp as early as possible
        self._transformStart = os.times()
        msg.debug('transformStart time is {0}'.format(self._transformStart))
 
        self._inFileValidationStart = None
        self._inFileValidationStop = None
        self._outFileValidationStart = None
        self._outFileValidationStop = None
 
        ## @brief Get trf pre-data as early as possible
        self._trfPredata = os.environ.get('TRF_PREDATA')
 
        ## Transform _name
        self._name = trfName or path.basename(sys.argv[0]).rsplit('.py', 1)[0]
        
        ## @note Holder for arguments this trf understands
        #  Use @c argparse.SUPPRESS to have non-given arguments unset, rather than None
        #  Support reading arguments from a file using the notation @c @file 
        self.parser = trfArgParser(description='Transform {0}. {1}'.format(self.name, description),
                                   argument_default=argparse.SUPPRESS,
                                   fromfile_prefix_chars='@')
 
        if standardTrfArgs:
            addStandardTrfArgs(self.parser)
 
        if standardValidationArgs:
            addValidationArguments(self.parser)
            addFileValidationArguments(self.parser)
 
        
        ## Argument dictionary for this transform
        self._argdict = dict()
        
        ## Dsta dictionary place holder (this maps data types to their argFile instances)
        self._dataDictionary = dict()
        
        
        # Transform executor list - initalise with an empty set
        self._executors = set()
        self._executorDictionary = {}
        
        # Append the given executors or a default one to the set:
        if executor is not None:
            self.appendToExecutorSet(executor or {transformExecutor()})
        
        ## Transform exit code/message holders
        self._exitCode = None
        self._exitMsg = None
 
        ## Report object for this transform
        self._report = trfJobReport(parentTrf = self)
        
        ## Transform processed events
        self._processedEvents = None
        
        # Setup standard signal handling if asked
        if standardSignalHandlers:
            setTrfSignalHandlers(self._exitWithReport)
            msg.debug('Standard signal handlers established')
 
            

Member Function Documentation

◆ _doSteering()

python.transform.transform._doSteering	(		self,
			steeringDict = None )

protected

Setup steering, which manipulates the graph before we trace the path for this transform.

Parameters

steeringDict Manual steering dictionary (if specified, used instead of the steering from the steering argument - pay attention to the input structure!

Definition at line 617 of file transform.py.

    def _doSteering(self, steeringDict = None):
        if not steeringDict:
            steeringDict = self._argdict['steering'].value
        for substep, steeringValues in steeringDict.items():
            foundSubstep = False
            for executor in self._executors:
                if executor.name == substep or executor.substep == substep:
                    foundSubstep = True
                    msg.debug('Updating {0} with {1}'.format(executor.name, steeringValues))
                    # Steering consists of tuples with (in/out, +/-, datatype) 
                    for steeringValue in steeringValues:
                        if steeringValue[0] == 'in':
                            startSet = executor.inData
                        else:
                            startSet = executor.outData
                        origLen = len(startSet)
                        msg.debug('Data values to be modified are: {0}'.format(startSet))
                        if steeringValue[1] == '+':
                            startSet.add(steeringValue[2])
                            if len(startSet) != origLen + 1:
                                raise trfExceptions.TransformSetupException(trfExit.nameToCode('TRF_GRAPH_STEERING_ERROR'),
                                                                            'Attempting to add data type {0} from {1} {2} fails (original set of data: {3}). Was this datatype already there?'.format(steeringValue[2], executor.name, steeringValue[1], startSet))
                        else:
                            startSet.discard(steeringValue[2])
                            if len(startSet) != origLen - 1:
                                raise trfExceptions.TransformSetupException(trfExit.nameToCode('TRF_GRAPH_STEERING_ERROR'),
                                                                            'Attempting to remove data type {0} from {1} {2} fails (original set of data: {3}). Was this datatype even present?'.format(steeringValue[2], executor.name, steeringValue[1], startSet))
                    msg.debug('Updated data values to: {0}'.format(startSet))
            if not foundSubstep:
                raise trfExceptions.TransformSetupException(trfExit.nameToCode('TRF_GRAPH_STEERING_ERROR'),
                                                            'This transform has no executor/substep {0}'.format(substep))                                
        
 

◆ _exitWithReport()

python.transform.transform._exitWithReport	(	self,
		signum,
		frame )

protected

Common signal handler.

This function is installed in place of the default signal handler and attempts to terminate the transform gracefully. When a signal is caught by the transform, the stdout from the running application process (i.e. athena.py) is allowed to continue uninterrupted and write it's stdout to the log file (to retrieve the traceback) before the associated job report records the fact that a signal has been caught and complete the report accordingly.

Parameters

signum	Signal number. Not used since this is a common handle assigned to predefined signals using the `_installSignalHandlers()`. This param is still required to satisfy the requirements of `signal.signal()`.
frame	Not used. Provided here to satisfy the requirements of `signal.signal()`.

Returns: Does not return. Raises SystemExit exception.

Exceptions

SystemExit()

Definition at line 756 of file transform.py.

    def _exitWithReport(self, signum, frame):
        msg.critical('Transform received signal {0}'.format(signum))
        msg.critical('Stack trace now follows:\n{0!s}'.format(''.join(traceback.format_stack(frame))))
        self._exitCode = 128+signum
        self._exitMsg = 'Transform received signal {0}'.format(signum)
        
        # Reset signal handlers now - we don't want to recurse if the same signal arrives again (e.g. multiple ^C)
        resetTrfSignalHandlers()
 
        msg.critical('Attempting to write reports with known information...')
        self.generateReport(fast=True)
        if ('orphanKiller' in self._argdict):
            infanticide(message=True, listOrphans=True)
        else:
            infanticide(message=True)
 
        sys.exit(self._exitCode)
    

◆ _setupGraph()

python.transform.transform._setupGraph ( self )

protected

Setup the executor graph.

Note: This function might need to be called again when the number of 'substeps' is unknown just based on the input data types - e.g., DigiMReco jobs don't know how many RDOtoESD steps they need to run until after digitisation.

Definition at line 525 of file transform.py.

    def _setupGraph(self):        
        # Get input/output data
        self._inputData = list()
        self._outputData = list()
        
        for key, value in self._argdict.items():
            # Note specifier [A-Za-z0-9_]+? makes this match non-greedy (avoid swallowing the optional 'File' suffix)
            m = re.match(r'(input|output|tmp)([A-Za-z0-9_]+?)(File)?$', key)
            # N.B. Protect against taking argunents which are not type argFile
            if m:
                if isinstance(value, argFile):
                    if m.group(1) == 'input':
                        self._inputData.append(m.group(2))
                    else:
                        self._outputData.append(m.group(2))
                    self._dataDictionary[m.group(2)] = value
                elif isinstance(value, list) and value and isinstance(value[0], argFile):
                    if m.group(1) == 'input':
                        self._inputData.append(m.group(2))
                    else:
                        self._outputData.append(m.group(2))
                    self._dataDictionary[m.group(2)] = value
 
        
        if len(self._inputData) == 0:
            self._inputData.append('inNULL')
        if len(self._outputData) == 0:
            self._outputData.append('outNULL')            
        msg.debug('Transform has this input data: {0}; output data {1}'.format(self._inputData, self._outputData))
        
        # Now see if we have any steering - manipulate the substep inputs and outputs before we
        # setup the graph
        if 'steering' in self._argdict:
            msg.debug('Now applying steering to graph: {0}'.format(self._argdict['steering'].value))
            self._doSteering()
        
        # Setup the graph and topo sort it
        self._executorGraph = executorGraph(self._executors, self._inputData, self._outputData)
        self._executorGraph.doToposort()
    

◆ _tracePath()

python.transform.transform._tracePath ( self )

protected

Trace the path through the executor graph.

Note: This function might need to be called again when the number of 'substeps' is unknown just based on the input data types - e.g., DigiMReco jobs don't know how many RDOtoESD steps they need to run until after digitisation.

Definition at line 602 of file transform.py.

    def _tracePath(self):
        self._executorGraph.findExecutionPath()
        
        self._executorPath = self._executorGraph.execution
        if len(self._executorPath) == 0:
            raise trfExceptions.TransformSetupException(trfExit.nameToCode('TRF_SETUP'), 
                                                        'Execution path finding resulted in no substeps being executed'
                                                        '(Did you correctly specify input data for this transform?)')
        # Tell the first executor that they are the first
        self._executorDictionary[self._executorPath[0]['name']].conf.firstExecutor = True
 

◆ appendToExecutorSet()

python.transform.transform.appendToExecutorSet	(		self,
			executors )

Definition at line 221 of file transform.py.

    def appendToExecutorSet(self, executors):
        # Normalise to something iterable
        if isinstance(executors, transformExecutor):
            executors = [executors,]
        elif not isinstance(executors, (list, tuple, set)):
            raise trfExceptions.TransformInternalException(trfExit.nameToCode('TRF_INTERNAL'), 
                                                           'Transform was initialised with an executor which was not a simple executor or an executor set')
    
        # TRY TO DEPRECATE SETTING trf IN THE EXECUTOR - USE CONF!
        # Executor book keeping: set parent link back to me for all executors
        # Also setup a dictionary, indexed by executor name and check that name is unique
        
        for executor in executors:
            executor.trf = self
            if executor.name in self._executorDictionary:
                raise trfExceptions.TransformInternalException(trfExit.nameToCode('TRF_INTERNAL'), 
                                                               'Transform has been initialised with two executors with the same name ({0})'
                                                               ' - executor names must be unique'.format(executor.name))
            self._executors.add(executor)
            self._executorDictionary[executor.name] = executor
 
 

◆ argdict()

python.transform.transform.argdict ( self )

Definition at line 134 of file transform.py.

    def argdict(self):
        return self._argdict
    

◆ dataDictionary()

python.transform.transform.dataDictionary ( self )

Definition at line 138 of file transform.py.

    def dataDictionary(self):
        return self._dataDictionary
    

◆ execute()

python.transform.transform.execute ( self )

Execute transform.

This function calls the actual transform execution class and sets self.exitCode, self.exitMsg and self.processedEvents transform data members.

Returns: None.

Definition at line 384 of file transform.py.

    def execute(self):
        msg.debug('Entering transform execution phase')
 
        #Warn if a CA-based tranform has deprecated command line args
        #Print warning once per deprecated arg
        deprecationWarningPrinted = False
        for exe in self._executors:
            if isinstance(exe, athenaExecutor):
                if (not deprecationWarningPrinted) and exe.skeletonCA:
                    toBeRemoved = ["autoConfiguration","trigStream","topOptions","valid"]
                    for deprecatedArg in toBeRemoved:
                        if deprecatedArg in self._argdict:
                            msg.warning("!!!Detected use of "+deprecatedArg+" in command line arguments for CA-based transform!!!")
                            msg.warning(deprecatedArg+" is DEPRECATED and due for removal. Applying it will not do anything, please remove it from your Transform definition.")
                            deprecationWarningPrinted = True
                            msg.debug(deprecatedArg+" detected in executor "+exe.name)
                                      
        try:
            # Intercept a few special options here      
            if 'dumpargs' in self._argdict:
                self.parser.dumpArgs()
                sys.exit(0)
                
            # Graph stuff!
            msg.info('Resolving execution graph')
            self._setupGraph()
 
            if 'showSteps' in self._argdict:
                for exe in self._executors:
                    print("Executor Step: {0} (alias {1})".format(exe.name, exe.substep))
                    if msg.level <= logging.DEBUG:
                        print(" {0} -> {1}".format(exe.inData, exe.outData))
                sys.exit(0)
                        
            if 'showGraph' in self._argdict:
                print(self._executorGraph)
                sys.exit(0)
                
            # Graph stuff!
            msg.info('Starting to trace execution path')            
            self._tracePath()
            msg.info('Execution path found with {0} step(s): {1}'.format(len(self._executorPath),
                                                                         ' '.join([exe['name'] for exe in self._executorPath])))
 
            if 'showPath' in self._argdict:
                msg.debug('Execution path list is: {0}'.format(self._executorPath))
                # Now print it nice
                print('Executor path is:')
                for node in self._executorPath:
                    print('  {0}: {1} -> {2}'.format(node['name'], list(node['input']), list(node['output'])))
                sys.exit(0)
 
            msg.debug('Execution path is {0}'.format(self._executorPath))
                
            # Prepare files for execution (separate method?)
            for dataType in [ data for data in self._executorGraph.data if 'NULL' not in data ]:
                if dataType in self._dataDictionary:
                    msg.debug('Data type {0} maps to existing argument {1}'.format(dataType, self._dataDictionary[dataType]))
                else:
                    fileName = 'tmp.' + dataType
                    # How to pick the correct argFile class?
                    for (prefix, suffix) in (('tmp', ''), ('output', 'File'), ('input', 'File')):
                        stdArgName = prefix + dataType + suffix
                        if stdArgName in self.parser._argClass:
                            msg.debug('Matched data type {0} to argument {1}'.format(dataType, stdArgName))
                            self._dataDictionary[dataType] = self.parser._argClass[stdArgName](fileName)
                            self._dataDictionary[dataType].io = 'temporary'
                            break
                    if dataType not in self._dataDictionary:
                        if 'HIST' in fileName:
                            self._dataDictionary[dataType] = argHISTFile(fileName, io='temporary', type=dataType.lower())
                            
                        else:
                            self._dataDictionary[dataType] = argFile(fileName, io='temporary', type=dataType.lower())
                            msg.debug('Did not find any argument matching data type {0} - setting to plain argFile: {1}'.format(dataType, self._dataDictionary[dataType]))
                    self._dataDictionary[dataType].name = fileName
 
            # Do splitting if required
            self.setupSplitting()
 
            # Error if more than one executor in MPI mode
            if 'mpi' in self._argdict:
                if len(self._executorPath) > 1:
                    msg.error("MPI mode is not supported for jobs with more than one execution step!")
                    msg.error(f"We have {len(self._executorPath)}: {self._executorPath}")
                    sys.exit(1)
 
            # Now we can set the final executor configuration properly, with the final dataDictionary
            for executor in self._executors:
                executor.conf.setFromTransform(self)
 
            self.validateInFiles()
            
            for executionStep in self._executorPath:
                msg.debug('Now preparing to execute {0}'.format(executionStep))
                executor = self._executorDictionary[executionStep['name']]
                executor.preExecute(input = executionStep['input'], output = executionStep['output'])
                try:
                    executor.execute()
                    executor.postExecute()
                except Exception as e:
                    msg.error('Exception encountered during execution of {0}:{1}'.format(executor.name,e))
                finally:
                    # Swap out the output files for the version with [] lists expanded
                    if 'mpi' in self._argdict:
                       new_data_dict = {**self._dataDictionary, **trfMPITools.mpiConfig["outputs"]}
                       self._dataDictionary = new_data_dict
                       executor.conf._dataDictionary = new_data_dict
                    executor.validate()
             
            self._processedEvents = self.getProcessedEvents()
            self.validateOutFiles()
            
            msg.debug('Transform executor succeeded')
            self._exitCode = 0
            self._exitMsg = trfExit.codeToName(self._exitCode)
            
        except trfExceptions.TransformNeedCheckException as e:
            msg.warning('Transform executor signaled NEEDCHECK condition: {0}'.format(e.errMsg))
            self._exitCode = e.errCode
            self._exitMsg = e.errMsg
            self.generateReport(fast=False)
 
        except trfExceptions.TransformException as e:
            msg.critical('Transform executor raised %s: %s' % (e.__class__.__name__, e.errMsg))
            self._exitCode = e.errCode
            self._exitMsg = e.errMsg
            # Try and write a job report...
            self.generateReport(fast=True)
 
        finally:
            # Clean up any orphaned processes and exit here if things went bad
            infanticide(message=True)
            if self._exitCode:
                msg.warning('Transform now exiting early with exit code {0} ({1})'.format(self._exitCode, self._exitMsg))
                sys.exit(self._exitCode)
        

◆ executors()

python.transform.transform.executors ( self )

Definition at line 206 of file transform.py.

    def executors(self):
        return self._executors
    

◆ exitCode()

python.transform.transform.exitCode ( self )

Definition at line 118 of file transform.py.

    def exitCode(self):
        if self._exitCode is None:
            msg.warning('Transform exit code getter: _exitCode is unset, returning "TRF_UNKNOWN"')
            return trfExit.nameToCode('TRF_UNKNOWN')
        else:
            return self._exitCode
        

◆ exitMsg()

python.transform.transform.exitMsg ( self )

Definition at line 126 of file transform.py.

    def exitMsg(self):
        if self._exitMsg is None:
            msg.warning('Transform exit message getter: _exitMsg is unset, returning empty string')
            return ''
        else:
            return self._exitMsg
    

◆ generateReport()

python.transform.transform.generateReport	(	self,
		reportType = None,
		fast = False,
		fileReport = defaultFileReport )

Transform report generator.

Parameters

fast	If True ensure that no external calls are made for file metadata (this is used to generate reports in a hurry after a crash or a forced exit)
fileReport	Dictionary giving the type of report to make for each type of file. This dictionary has to have all io types as keys and valid values are: `None` - skip this io type; `'full'` - Provide all details; `'name'` - only dataset and filename will be reported on.
reportType	Iterable with report types to generate, otherwise a sensible default is used (~everything, plus the Tier0 report at Tier0)

Definition at line 674 of file transform.py.

    def generateReport(self, reportType=None, fast=False, fileReport = defaultFileReport):
        msg.debug('Transform report generator')
        if 'mpi' in self.argdict and not trfMPITools.mpiShouldValidate():
            msg.debug("Not in rank 0 -- not generating reports")
            return
 
        if 'reportType' in self._argdict:
            if reportType is not None:
                msg.info('Transform requested report types {0} overridden by command line to {1}'.format(reportType, self._argdict['reportType'].value))
            reportType = self._argdict['reportType'].value
 
        if reportType is None:
            reportType = ['json', ]
            # Only generate the Tier0 report at Tier0 ;-)
            # (It causes spurious warnings for some grid jobs with background files (e.g., digitisation)
            if 'TZHOME' in os.environ:
                reportType.append('gpickle')
 
            if not isInteractiveEnv():
                reportType.append('text')
                msg.debug('Detected Non-Interactive environment. Enabled text report')
 
        if 'reportName' in self._argdict:
            baseName = classicName = self._argdict['reportName'].value
        else:
            baseName = 'jobReport'
            classicName = 'metadata'
 
        try:
            # Text: Writes environment variables and machine report in text format.
            if reportType is None or 'text' in reportType:
                envName = baseName if 'reportName' in self._argdict else 'env'  # Use fallback name 'env.txt' if it's not specified.
                self._report.writeTxtReport(filename='{0}.txt'.format(envName), fast=fast, fileReport=fileReport)
            # JSON
            if reportType is None or 'json' in reportType:
                self._report.writeJSONReport(filename='{0}.json'.format(baseName), fast=fast, fileReport=fileReport)
            # Classic XML
            if reportType is None or 'classic' in reportType:
                self._report.writeClassicXMLReport(filename='{0}.xml'.format(classicName), fast=fast)
            # Classic gPickle
            if reportType is None or 'gpickle' in reportType:
                self._report.writeGPickleReport(filename='{0}.gpickle'.format(baseName), fast=fast)
            # Pickled version of the JSON report for pilot
            if reportType is None or 'pilotPickle' in reportType:
                self._report.writePilotPickleReport(filename='{0}Extract.pickle'.format(baseName), fast=fast, fileReport=fileReport)
 
        except trfExceptions.TransformTimeoutException as reportException:
            msg.error('Received timeout when writing report ({0})'.format(reportException))
            msg.error('Report writing is aborted - sorry. Transform will exit with TRF_METADATA_CALL_FAIL status.')
            if ('orphanKiller' in self._argdict):
                infanticide(message=True, listOrphans=True)
            else:
                infanticide(message=True)
            sys.exit(trfExit.nameToCode('TRF_METADATA_CALL_FAIL'))
 
        except trfExceptions.TransformException as reportException:
            # This is a bad one!
            msg.critical('Attempt to write job report failed with exception {0!s}: {1!s}'.format(reportException.__class__.__name__, reportException))
            msg.critical('Stack trace now follows:\n{0}'.format(traceback.format_exc()))
            msg.critical('Job reports are likely to be missing or incomplete - sorry')
            msg.critical('Please report this as a transforms bug!')
            msg.critical('Before calling the report generator the transform status was: {0}; exit code {1}'.format(self._exitMsg, self._exitCode))
            msg.critical('Now exiting with a transform internal error code')
            if ('orphanKiller' in self._argdict):
                infanticide(message=True, listOrphans=True)
            else:
                infanticide(message=True)
            sys.exit(trfExit.nameToCode('TRF_INTERNAL'))
    
 

◆ getFiles()

python.transform.transform.getFiles	(		self,
			io = None )

Return a list of fileArgs used by the transform.

Parameters

c io Filter files by io attribute

Returns: List of argFile instances

Definition at line 800 of file transform.py.

    def getFiles(self, io = None):
        res = []
        msg.debug('Looking for file arguments matching: io={0}'.format(io))
        for argName, arg in self._argdict.items():
            if isinstance(arg, argFile):
                msg.debug('Argument {0} is argFile type ({1!s})'.format(argName, arg))
                if io is not None and arg.io != io:
                    continue
                msg.debug('Argument {0} matches criteria'.format(argName))
                res.append(arg)
        return res
 
 

◆ getProcessedEvents()

python.transform.transform.getProcessedEvents ( self )

Definition at line 213 of file transform.py.

    def getProcessedEvents(self):
        nEvts = None
        for executionStep in self._executorPath:
            executor = self._executorDictionary[executionStep['name']]
            if executor.conf.firstExecutor:
                nEvts = executor.eventCount
        return nEvts
 

◆ getValidationDict()

python.transform.transform.getValidationDict ( self )

Getter function for transform validation dictionary.

Returns: Validiation dictionary

Definition at line 785 of file transform.py.

    def getValidationDict(self):
        return self.validation
    

◆ getValidationOption()

python.transform.transform.getValidationOption	(		self,
			key )

Getter for a specific validation option.

Parameters

key	Validation dictionary key

Returns: Valdiation key value or None if this key is absent

Definition at line 791 of file transform.py.

    def getValidationOption(self, key):
        if key in self.validation:
            return self.validation[key]
        else:
            return None
 

◆ inFileValidationCpuTime()

python.transform.transform.inFileValidationCpuTime ( self )

Definition at line 166 of file transform.py.

    def inFileValidationCpuTime(self):
        inFileValidationCpuTime = None
        if self._inFileValidationStart and self._inFileValidationStop:
            inFileValidationCpuTime = calcCpuTime(self._inFileValidationStart, self._inFileValidationStop)
 
        return inFileValidationCpuTime
 

◆ inFileValidationWallTime()

python.transform.transform.inFileValidationWallTime ( self )

Definition at line 174 of file transform.py.

    def inFileValidationWallTime(self):
        inFileValidationWallTime = None
        if self._inFileValidationStart and self._inFileValidationStop:
            inFileValidationWallTime = calcWallTime(self._inFileValidationStart, self._inFileValidationStop)
 
        return inFileValidationWallTime
 

◆ lastExecuted()

python.transform.transform.lastExecuted ( self )

Return the last executor which actually executed.

Returns: Last executor which has _hasExecuted == True, or the very first executor if we didn't even start yet

Definition at line 653 of file transform.py.

    def lastExecuted(self):
        # Just make sure we have the path traced
        if not hasattr(self, '_executorPath') or len(self._executorPath) == 0:
            return None
            
        lastExecutor = self._executorDictionary[self._executorPath[0]['name']]
        for executorStep in self._executorPath[1:]:
            if self._executorDictionary[executorStep['name']].hasExecuted:
                lastExecutor = self._executorDictionary[executorStep['name']]
        return lastExecutor
 
 

◆ name()

python.transform.transform.name ( self )

Definition at line 114 of file transform.py.

    def name(self):
        return self._name
            

◆ outFileValidationCpuTime()

python.transform.transform.outFileValidationCpuTime ( self )

Definition at line 182 of file transform.py.

    def outFileValidationCpuTime(self):
        outFileValidationCpuTime = None
        if self._outFileValidationStart and self._outFileValidationStop:
            outFileValidationCpuTime = calcCpuTime(self._outFileValidationStart, self._outFileValidationStop)
 
        return outFileValidationCpuTime
 

◆ outFileValidationStop()

python.transform.transform.outFileValidationStop ( self )

Definition at line 198 of file transform.py.

    def outFileValidationStop(self):
        return self._outFileValidationStop
 

◆ outFileValidationWallTime()

python.transform.transform.outFileValidationWallTime ( self )

Definition at line 190 of file transform.py.

    def outFileValidationWallTime(self):
        outFileValidationWallTime = None
        if self._outFileValidationStart and self._outFileValidationStop:
            outFileValidationWallTime = calcWallTime(self._outFileValidationStart, self._outFileValidationStop)
 
        return outFileValidationWallTime
 

◆ parseCmdLineArgs()

python.transform.transform.parseCmdLineArgs	(		self,
			args )

Parse command line arguments for a transform.

Definition at line 244 of file transform.py.

    def parseCmdLineArgs(self, args):
        msg.info('Transform command line was: %s', ' '.join(shQuoteStrings(sys.argv)))
        
        try:
            # Use the argparse infrastructure to get the actual command line arguments
            self._argdict=vars(self.parser.parse_args(args))
 
            # Need to know if any input or output files were set - if so then we suppress the
            # corresponding parameters from AMI
            inputFiles = outputFiles = False
            for k, v in self._argdict.items():
                if k.startswith('input') and isinstance(v, argFile):
                    inputFiles = True
                elif k.startswith('output') and isinstance(v, argFile):
                    outputFiles = True
            msg.debug("CLI Input files: {0}; Output files {1}".format(inputFiles, outputFiles))
 
            # Now look for special arguments, which expand out to other parameters
            # Note that the pickled argdict beats AMIConfig because dict.update() will overwrite
            # (However, we defend the real command line against updates from either source)
            extraParameters = {}
            # AMI configuration?
            if 'AMIConfig' in self._argdict:
                msg.debug('Given AMI tag configuration {0}'.format(self._argdict['AMIConfig']))
                from PyJobTransforms.trfAMI import TagInfo
                tag=TagInfo(self._argdict['AMIConfig'].value)
                updateDict = {}
                for k, v in dict(tag.trfs[0]).items():
                    # Convert to correct internal key form
                    k = cliToKey(k)
                    if inputFiles and k.startswith('input'):
                        msg.debug('Suppressing argument {0} from AMI'
                                  ' because input files have been specified on the command line'.format(k))
                        continue
                    if outputFiles and k.startswith('output'):
                        msg.debug('Suppressing argument {0} from AMI'
                                  ' because output files have been specified on the command line'.format(k))
                        continue
                    updateDict[k] = v
                extraParameters.update(updateDict)                
 
            # JSON arguments?
            if 'argJSON' in self._argdict:
                try:
                    import json
                    msg.debug('Given JSON encoded arguments in {0}'.format(self._argdict['argJSON']))
                    argfile = open(self._argdict['argJSON'], 'r')
                    jsonParams = json.load(argfile)
                    msg.debug('Read: {0}'.format(jsonParams))
                    extraParameters.update(convertToStr(jsonParams))
                    argfile.close()
                except Exception as e:
                    raise trfExceptions.TransformArgException(trfExit.nameToCode('TRF_ARG_ERROR'), 'Error when deserialising JSON file {0} ({1})'.format(self._argdict['argJSON'], e))
            
            # Event Service
            if 'eventService' in self._argdict and self._argdict['eventService'].value:
                updateDict = {}
                updateDict['athenaMPMergeTargetSize'] = '*:0'
                updateDict['checkEventCount'] = False
                updateDict['outputFileValidation'] = False
                extraParameters.update(updateDict)
                
            # Process anything we found
            # List of command line arguments
            argsList = [ i.split("=", 1)[0].lstrip('-') for i in args if i.startswith('-')]
            for k,v in extraParameters.items():
                msg.debug('Found this extra argument: {0} with value: {1} ({2})'.format(k, v, type(v)))
                if k not in self.parser._argClass and k not in self.parser._argAlias:
                    raise trfExceptions.TransformArgException(trfExit.nameToCode('TRF_ARG_ERROR'), 'Argument "{0}" not known (try "--help")'.format(k))
                # Check if it is an alias
                if k in self.parser._argAlias:
                    msg.debug('Resolving alias from {0} to {1}'.format(k, self.parser._argAlias[k]))
                    k = self.parser._argAlias[k]
                # Check if argument has already been set on the command line
                if k in argsList:
                    msg.debug('Ignored {0}={1} as extra parameter because this argument was given on the command line.'.format(k, v))
                    continue
                # For callable classes we instantiate properly, otherwise we set the value for simple arguments
                if '__call__' in dir(self.parser._argClass[k]):
                    self._argdict[k] = self.parser._argClass[k](v)
                else:
                    self._argdict[k] = v
                msg.debug('Argument {0} set to {1}'.format(k, self._argdict[k]))
 
            # Set the key name as an argument property - useful to be able to look bask at where this
            # argument came from
            for k, v in self._argdict.items():
                if isinstance(v, argument):
                    v.name = k
                elif isinstance(v, list):
                    for it in v:
                        if isinstance(it, argument):
                            it.name = k
 
            # Now we parsed all arguments, if a pickle/json dump is requested do it here and exit
            if 'dumpPickle' in self._argdict:
                msg.info('Now dumping pickled version of command line to {0}'.format(self._argdict['dumpPickle']))
                pickledDump(self._argdict)
                sys.exit(0)
 
            # Now we parsed all arguments, if a pickle/json dump is requested do it here and exit
            if 'dumpJSON' in self._argdict:
                msg.info('Now dumping JSON version of command line to {0}'.format(self._argdict['dumpJSON']))
                JSONDump(self._argdict)
                sys.exit(0)
 
        except trfExceptions.TransformArgException as e:
            msg.critical('Argument parsing failure: {0!s}'.format(e))
            self._exitCode = e.errCode
            self._exitMsg = e.errMsg
            self._report.fast = True
            self.generateReport()
            sys.exit(self._exitCode)
            
        except trfExceptions.TransformAMIException as e:
            msg.critical('AMI failure: {0!s}'.format(e))
            self._exitCode = e.errCode
            self._exitMsg = e.errMsg
            sys.exit(self._exitCode)
 
        self.setGlobalLogLevel()
        
 

◆ processedEvents()

python.transform.transform.processedEvents ( self )

Definition at line 210 of file transform.py.

    def processedEvents(self):
        return self._processedEvents
    

◆ report()

python.transform.transform.report ( self )

Definition at line 142 of file transform.py.

    def report(self):
        return self._report
    

◆ setGlobalLogLevel()

python.transform.transform.setGlobalLogLevel ( self )

Check transform argument dictionary and set the correct root logger option.

Definition at line 368 of file transform.py.

    def setGlobalLogLevel(self):
        if 'verbose' in self._argdict:
            setRootLoggerLevel(stdLogLevels['DEBUG'])
        elif 'loglevel' in self._argdict:
            if self._argdict['loglevel'] in stdLogLevels:
                msg.info("Loglevel option found - setting root logger level to %s", 
                         logging.getLevelName(stdLogLevels[self._argdict['loglevel']]))
                setRootLoggerLevel(stdLogLevels[self._argdict['loglevel']])
            else:
                msg.warning('Unrecognised loglevel ({0}) given - ignored'.format(self._argdict['loglevel']))
 
        

◆ setupSplitting()

python.transform.transform.setupSplitting ( self )

Setup executor splitting.

Definition at line 567 of file transform.py.

    def setupSplitting(self):
        if 'splitConfig' not in self._argdict:
            return
 
        split = []
        for executionStep in self._executorPath:
            baseStepName = executionStep['name']
            if baseStepName in split:
                continue
 
            baseExecutor = self._executorDictionary[baseStepName]
            splitting = getTotalExecutorSteps(baseExecutor, argdict=self._argdict)
            if splitting <= 1:
                continue
 
            msg.info('Splitting {0} into {1} substeps'.format(executionStep, splitting))
            index = self._executorPath.index(executionStep)
            baseStep = self._executorPath.pop(index)
            for i in range(splitting):
                name = baseStepName + executorStepSuffix + str(i)
                step = copy.deepcopy(baseStep)
                step['name'] = name
                self._executorPath.insert(index + i, step)
                executor = copy.deepcopy(baseExecutor)
                executor.name = name
                executor.conf.executorStep = i
                executor.conf.totalExecutorSteps = splitting
                self._executors.add(executor)
                self._executorDictionary[name] = executor
                split.append(name)
    

◆ transformSetupCpuTime()

python.transform.transform.transformSetupCpuTime ( self )

Definition at line 150 of file transform.py.

    def transformSetupCpuTime(self):
        transformSetupCpuTime = None
        if self._transformStart and self._inFileValidationStart:
            transformSetupCpuTime = calcCpuTime(self._transformStart, self._inFileValidationStart)
 
        return transformSetupCpuTime
 

◆ transformSetupWallTime()

python.transform.transform.transformSetupWallTime ( self )

Definition at line 158 of file transform.py.

    def transformSetupWallTime(self):
        transformSetupWallTime = None
        if self._transformStart and self._inFileValidationStart:
            transformSetupWallTime = calcWallTime(self._transformStart, self._inFileValidationStart)
 
        return transformSetupWallTime
 

◆ transformStart()

python.transform.transform.transformStart ( self )

Definition at line 146 of file transform.py.

    def transformStart(self):
        return self._transformStart
 

◆ trfPredata()

python.transform.transform.trfPredata ( self )

Definition at line 202 of file transform.py.

    def trfPredata(self):
        return self._trfPredata
    

◆ updateValidationDict()

python.transform.transform.updateValidationDict	(		self,
			newValidationOptions )

Setter for transform's validation dictionary.

This function updates the validation dictionary for the transform, updating values which are passed in the newValidationOptions argument.

Parameters

newValidationOptions Dictionary (or tuples) to update validation dictionary with

Returns: None

Definition at line 780 of file transform.py.

    def updateValidationDict(self, newValidationOptions):
        self.validation.update(newValidationOptions)
 

◆ validateInFiles()

python.transform.transform.validateInFiles ( self )

Definition at line 813 of file transform.py.

    def validateInFiles(self):
        if self._inFileValidationStart is None:
            self._inFileValidationStart = os.times()
            msg.debug('inFileValidationStart time is {0}'.format(self._inFileValidationStart))
 
        if (('skipFileValidation' in self._argdict and self._argdict['skipFileValidation'] is True) or
            ('skipInputFileValidation' in self._argdict and self._argdict['skipInputFileValidation'] is True) or
            ('fileValidation' in self._argdict and self._argdict['fileValidation'].value is False) or
            ('inputFileValidation' in self._argdict and self._argdict['inputFileValidation'].value is False)
            ):
            msg.info('Standard input file validation turned off for transform %s.', self.name)
        else:
            msg.info('Validating input files')
            if 'parallelFileValidation' in self._argdict:
                trfValidation.performStandardFileValidation(dictionary=self._dataDictionary, io='input', parallelMode=self._argdict['parallelFileValidation'].value )
            else:
                trfValidation.performStandardFileValidation(dictionary=self._dataDictionary, io='input')
 
        self._inFileValidationStop = os.times()
        msg.debug('inFileValidationStop time is {0}'.format(self._inFileValidationStop))
 

◆ validateOutFiles()

python.transform.transform.validateOutFiles ( self )

Definition at line 834 of file transform.py.

    def validateOutFiles(self):
        if self._outFileValidationStart is None:
            self._outFileValidationStart = os.times()
            msg.debug('outFileValidationStart time is {0}'.format(self._outFileValidationStart))
 
        if (('skipFileValidation' in self._argdict and self._argdict['skipFileValidation'] is True) or
            ('skipOutputFileValidation' in self._argdict and self._argdict['skipOutputFileValidation'] is True) or
            ('fileValidation' in self._argdict and self._argdict['fileValidation'].value is False) or
            ('outputFileValidation' in self._argdict and self._argdict['outputFileValidation'].value is False)
            ):
            msg.info('Standard output file validation turned off for transform %s.', self.name)
        elif 'mpi' in self.argdict and not trfMPITools.mpiShouldValidate():
            msg.info("MPI mode and not in rank 0 ∴ not validating partial outputs")
        else:
            msg.info('Validating output files')
            parparallelMode = False
            # Make MT file validation default
            parmultithreadedMode = True
            if 'parallelFileValidation' in self._argdict:
                parparallelMode = self._argdict['parallelFileValidation'].value
            if 'multithreadedFileValidation' in self._argdict:
                parmultithreadedMode = self._argdict['multithreadedFileValidation'].value
            trfValidation.performStandardFileValidation(dictionary=self._dataDictionary, io='output', parallelMode=parparallelMode, multithreadedMode=parmultithreadedMode)
 
        self._outFileValidationStop = os.times()
        msg.debug('outFileValidationStop time is {0}'.format(self._outFileValidationStop))

Member Data Documentation

◆ _argdict

python.transform.transform._argdict = dict()

protected

Argument dictionary for this transform.

Definition at line 83 of file transform.py.

◆ _dataDictionary

python.transform.transform._dataDictionary = dict()

protected

Dsta dictionary place holder (this maps data types to their argFile instances).

Definition at line 86 of file transform.py.

◆ _executorDictionary

dict python.transform.transform._executorDictionary = {}

protected

Definition at line 91 of file transform.py.

◆ _executorGraph

python.transform.transform._executorGraph

protected

Definition at line 419 of file transform.py.

◆ _executorPath

python.transform.transform._executorPath

protected

Definition at line 215 of file transform.py.

◆ _executors

python.transform.transform._executors = set()

protected

Definition at line 90 of file transform.py.

◆ _exitCode

python.transform.transform._exitCode = None

protected

Transform exit code/message holders.

Definition at line 98 of file transform.py.

◆ _exitMsg

python.transform.transform._exitMsg = None

protected

Definition at line 99 of file transform.py.

◆ _exitWithReport

python.transform.transform._exitWithReport

protected

Definition at line 109 of file transform.py.

◆ _inFileValidationStart

python.transform.transform._inFileValidationStart = None

protected

Definition at line 56 of file transform.py.

◆ _inFileValidationStop

python.transform.transform._inFileValidationStop = None

protected

Definition at line 57 of file transform.py.

◆ _inputData

python.transform.transform._inputData = list()

protected

Note: If we have no real data then add the pseudo datatype NULL, which allows us to manage transforms which can run without data

Definition at line 527 of file transform.py.

◆ _name

python.transform.transform._name = trfName or path.basename(sys.argv[0]).rsplit('.py', 1)[0]

protected

Transform _name.

Definition at line 65 of file transform.py.

◆ _outFileValidationStart

python.transform.transform._outFileValidationStart = None

protected

Definition at line 58 of file transform.py.

◆ _outFileValidationStop

python.transform.transform._outFileValidationStop = None

protected

Definition at line 59 of file transform.py.

◆ _outputData

python.transform.transform._outputData = list()

protected

Definition at line 528 of file transform.py.

◆ _processedEvents

python.transform.transform._processedEvents = None

protected

Transform processed events.

Definition at line 105 of file transform.py.

◆ _report

python.transform.transform._report = trfJobReport(parentTrf = self)

protected

Report object for this transform.

Definition at line 102 of file transform.py.

◆ _transformStart

python.transform.transform._transformStart = os.times()

protected

Get transform starting timestamp as early as possible.

Definition at line 53 of file transform.py.

◆ _trfPredata

python.transform.transform._trfPredata = os.environ.get('TRF_PREDATA')

protected

Get trf pre-data as early as possible.

Definition at line 62 of file transform.py.

◆ name

python.transform.transform.name

Definition at line 823 of file transform.py.

◆ parser

python.transform.transform.parser

Initial value:

=  trfArgParser(description='Transform {0}. {1}'.format(self.name, description),
                                   argument_default=argparse.SUPPRESS,
                                   fromfile_prefix_chars='@')

Definition at line 70 of file transform.py.

◆ validation

python.transform.transform.validation

Definition at line 792 of file transform.py.

The documentation for this class was generated from the following file:

transform.py

Public Member Functions

Public Attributes

Protected Member Functions

Protected Attributes

Detailed Description

Constructor & Destructor Documentation

◆ __init__()

Member Function Documentation

◆ _doSteering()

◆ _exitWithReport()

◆ _setupGraph()

◆ _tracePath()

◆ appendToExecutorSet()

◆ argdict()

◆ dataDictionary()

◆ execute()

◆ executors()

◆ exitCode()

◆ exitMsg()

◆ generateReport()

◆ getFiles()

◆ getProcessedEvents()

◆ getValidationDict()

◆ getValidationOption()

◆ inFileValidationCpuTime()

◆ inFileValidationWallTime()

◆ lastExecuted()

◆ name()

◆ outFileValidationCpuTime()

◆ outFileValidationStop()

◆ outFileValidationWallTime()

◆ parseCmdLineArgs()

◆ processedEvents()

◆ report()

◆ setGlobalLogLevel()

◆ setupSplitting()

◆ transformSetupCpuTime()

◆ transformSetupWallTime()

◆ transformStart()

◆ trfPredata()

◆ updateValidationDict()

◆ validateInFiles()

◆ validateOutFiles()

Member Data Documentation

◆ _argdict

◆ _dataDictionary

◆ _executorDictionary

◆ _executorGraph

◆ _executorPath

◆ _executors

◆ _exitCode

◆ _exitMsg

◆ _exitWithReport

◆ _inFileValidationStart

◆ _inFileValidationStop

◆ _inputData

◆ _name

◆ _outFileValidationStart

◆ _outFileValidationStop

◆ _outputData

◆ _processedEvents

◆ _report

◆ _transformStart

◆ _trfPredata

◆ name

◆ parser

◆ validation

◆ init()