transform.py

Go to the documentation of this file.

# Copyright (C) 2002-2024 CERN for the benefit of the ATLAS collaboration
 
 
 
import argparse
import copy
import os
import os.path as path
import re
import sys
import traceback
 
import logging
msg = logging.getLogger(__name__)
 
import PyJobTransforms.trfValidation as trfValidation
import PyJobTransforms.trfExceptions as trfExceptions
import PyJobTransforms.trfMPITools as trfMPITools
 
from PyJobTransforms.trfSignal import setTrfSignalHandlers, resetTrfSignalHandlers
from PyJobTransforms.trfArgs import addStandardTrfArgs, addFileValidationArguments, addValidationArguments
from PyJobTransforms.trfLogger import setRootLoggerLevel, stdLogLevels
from PyJobTransforms.trfArgClasses import trfArgParser, argFile, argHISTFile, argument
from PyJobTransforms.trfExeStepTools import executorStepSuffix, getTotalExecutorSteps
from PyJobTransforms.trfExitCodes import trfExit
from PyJobTransforms.trfUtils import shQuoteStrings, infanticide, pickledDump, JSONDump, cliToKey, convertToStr
from PyJobTransforms.trfUtils import isInteractiveEnv, calcCpuTime, calcWallTime
from PyJobTransforms.trfReports import trfJobReport, defaultFileReport
from PyJobTransforms.trfExe import transformExecutor
from PyJobTransforms.trfGraph import executorGraph
 
 
class transform(object):
    
    
    def __init__(self, standardSignalHandlers = True, standardTrfArgs = True, standardValidationArgs=True, 
                 trfName = None, executor = None, exeArgs = None, description = ''):
        '''Transform class initialiser'''
        msg.debug('Welcome to ATLAS job transforms')
        
        
        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
 
        
        self._trfPredata = os.environ.get('TRF_PREDATA')
 
        
        self._name = trfName or path.basename(sys.argv[0]).rsplit('.py', 1)[0]
        
        
        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)
 
        
        
        self._argdict = dict()
        
        
        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()})
        
        
        self._exitCode = None
        self._exitMsg = None
 
        
        self._report = trfJobReport(parentTrf = self)
        
        
        self._processedEvents = None
        
        # Setup standard signal handling if asked
        if standardSignalHandlers:
            setTrfSignalHandlers(self._exitWithReport)
            msg.debug('Standard signal handlers established')
 
            
    @property
    def name(self):
        return self._name
            
    @property
    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
        
    @property
    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
    
    @property
    def argdict(self):
        return self._argdict
    
    @property
    def dataDictionary(self):
        return self._dataDictionary
    
    @property
    def report(self):
        return self._report
    
    @property
    def transformStart(self):
        return self._transformStart
 
    @property
    def transformSetupCpuTime(self):
        transformSetupCpuTime = None
        if self._transformStart and self._inFileValidationStart:
            transformSetupCpuTime = calcCpuTime(self._transformStart, self._inFileValidationStart)
 
        return transformSetupCpuTime
 
    @property
    def transformSetupWallTime(self):
        transformSetupWallTime = None
        if self._transformStart and self._inFileValidationStart:
            transformSetupWallTime = calcWallTime(self._transformStart, self._inFileValidationStart)
 
        return transformSetupWallTime
 
    @property
    def inFileValidationCpuTime(self):
        inFileValidationCpuTime = None
        if self._inFileValidationStart and self._inFileValidationStop:
            inFileValidationCpuTime = calcCpuTime(self._inFileValidationStart, self._inFileValidationStop)
 
        return inFileValidationCpuTime
 
    @property
    def inFileValidationWallTime(self):
        inFileValidationWallTime = None
        if self._inFileValidationStart and self._inFileValidationStop:
            inFileValidationWallTime = calcWallTime(self._inFileValidationStart, self._inFileValidationStop)
 
        return inFileValidationWallTime
 
    @property
    def outFileValidationCpuTime(self):
        outFileValidationCpuTime = None
        if self._outFileValidationStart and self._outFileValidationStop:
            outFileValidationCpuTime = calcCpuTime(self._outFileValidationStart, self._outFileValidationStop)
 
        return outFileValidationCpuTime
 
    @property
    def outFileValidationWallTime(self):
        outFileValidationWallTime = None
        if self._outFileValidationStart and self._outFileValidationStop:
            outFileValidationWallTime = calcWallTime(self._outFileValidationStart, self._outFileValidationStop)
 
        return outFileValidationWallTime
 
    @property
    def outFileValidationStop(self):
        return self._outFileValidationStop
 
    @property
    def trfPredata(self):
        return self._trfPredata
    
    @property
    def executors(self):
        return self._executors
    
    @property
    def processedEvents(self):
        return self._processedEvents
    
    def getProcessedEvents(self):
        nEvts = None
        for executionStep in self._executorPath:
            executor = self._executorDictionary[executionStep['name']]
            if executor.conf.firstExecutor:
                nEvts = executor.eventCount
        return nEvts
 
    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
 
 
    
    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()
        
 
    
    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']))
 
        
    
    def execute(self):
        msg.debug('Entering transform execution phase')
  
        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()
                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)
        
    
    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()
    
    
    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)
    
    
    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
 
    
    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))                                
        
 
    
    @property
    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
 
 
    
    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'))
    
 
    # Description stolen from old trfs...
    
    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)
    
    
    def updateValidationDict(self, newValidationOptions):
        self.validation.update(newValidationOptions)
 
    
    def getValidationDict(self):
        return self.validation
    
    
    def getValidationOption(self, key):
        if key in self.validation:
            return self.validation[key]
        else:
            return None
 
    
    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
 
 
    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))
 
    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))