python.TransformUtils Namespace Reference

Functions
	ExecCondAlgsAtPreFork (flags, cfg)
	executeFromFragment (fragment_string, flags, cfg=None)
	processPreExec (runArgs, flags)
	processPostExec (runArgs, flags, cfg)
	processPreInclude (runArgs, flags)
	processPostInclude (runArgs, flags, cfg)
	UseFrontier (flags)
	UseCREST (flags)
	DumpPickle (flags, cfg)
	SortInput (flags, cfg)

Variables
	msg = logging.getLogger(__name__)

Function Documentation

DumpPickle()

python.TransformUtils.DumpPickle	(		flags,
			cfg )

Dump the pickle file for the current configuration

Definition at line 119 of file Tools/PyJobTransforms/python/TransformUtils.py.

def DumpPickle(flags, cfg):
    """Dump the pickle file for the current configuration"""
    with open("Configuration.pkl", "wb") as f:
        cfg.store(f)
 
 

ExecCondAlgsAtPreFork()

python.TransformUtils.ExecCondAlgsAtPreFork	(		flags,
			cfg )

 Execute the entire conditions sequence during the PreFork step.
This allows saving some memory in AthenaMP jobs, which is particularly
useful in the derivation production.

Definition at line 5 of file Control/AthenaServices/python/TransformUtils.py.

def ExecCondAlgsAtPreFork(flags, cfg):
    """ Execute the entire conditions sequence during the PreFork step.
    This allows saving some memory in AthenaMP jobs, which is particularly
    useful in the derivation production.
    """
    log = logging.getLogger('ExecCondAlgsAtPreFork')
    try:
        cfg.getService('AthMpEvtLoopMgr').ExecAtPreFork=['AthCondSeq']
        log.info('AthCondSeq will be executed during PreFork')
    except ConfigurationError: # in this context this might not be an error
        log.info('AthCondSeq will NOT be executed during PreFork')

executeFromFragment()

python.TransformUtils.executeFromFragment	(		fragment_string,
			flags,
			cfg = None )

Execute a function from a pre/post include fragment.

Definition at line 6 of file Tools/PyJobTransforms/python/TransformUtils.py.

def executeFromFragment(fragment_string, flags, cfg=None):
    """Execute a function from a pre/post include fragment."""
    # detect legacy job options in the transition period:
    if fragment_string.endswith('.py'):
        msg.warning(f'Trying to load legacy job options {fragment_string}. This should NOT be used in production!')
        fragment_string = fragment_string[:-3]
        fragment_string = fragment_string.replace('/', '.')
        msg.warning(f'Resolved to {fragment_string}')
 
    parts = fragment_string.split('.')
    if len(parts) < 2:
        raise ValueError('Pre/post include should be of the form Package.Module.Function or Package.Function if defined in __init__.py')
 
    function = parts[-1]
    module = '.'.join(parts[:-1])
 
    from importlib import import_module
    loaded_module = import_module(module)
    function_def = getattr(loaded_module, function)
 
    if not cfg:
        function_def(flags)
        return
 
    from inspect import getfullargspec
    argspec = getfullargspec(function_def)
    print(argspec)
    argCount = len(argspec.args)
    defaultsCount = len(argspec.defaults) if argspec.defaults else 0
    if argCount - defaultsCount == 1:
        cfg.merge(function_def(flags))
    else:
        function_def(flags, cfg)
 
 

processPostExec()

python.TransformUtils.processPostExec	(		runArgs,
			flags,
			cfg )

Process postExec from runtime arguments.

Definition at line 50 of file Tools/PyJobTransforms/python/TransformUtils.py.

def processPostExec(runArgs, flags, cfg):
    """Process postExec from runtime arguments."""
    if not flags.locked():
        raise RuntimeError('Running a postExec before locking ConfigFlags')
 
    if hasattr(runArgs, 'postExec') and runArgs.postExec and runArgs.postExec != 'NONE':
        ConfigFlags = flags  # noqa: F841
        from AthenaConfiguration.ComponentFactory import CompFactory # noqa: F401
        for cmd in runArgs.postExec:
            exec(cmd)
 
 

processPostInclude()

python.TransformUtils.processPostInclude	(		runArgs,
			flags,
			cfg )

Process postInclude from runtime arguments.

Definition at line 69 of file Tools/PyJobTransforms/python/TransformUtils.py.

def processPostInclude(runArgs, flags, cfg):
    """Process postInclude from runtime arguments."""
    if not flags.locked():
        raise RuntimeError('Running a postInclude before locking ConfigFlags')
 
    if hasattr(runArgs, 'postInclude') and runArgs.postInclude and runArgs.postInclude != 'NONE':
        for fragment in runArgs.postInclude:
            executeFromFragment(fragment, flags, cfg)
 
 

processPreExec()

python.TransformUtils.processPreExec	(		runArgs,
			flags )

Process preExec from runtime arguments.

Definition at line 41 of file Tools/PyJobTransforms/python/TransformUtils.py.

def processPreExec(runArgs, flags):
    """Process preExec from runtime arguments."""
    if hasattr(runArgs, 'preExec') and runArgs.preExec and runArgs.preExec != 'NONE':
        ConfigFlags = flags  # noqa: F841
        from AthenaConfiguration.Enums import HIMode,BeamType  # noqa: F401
        for cmd in runArgs.preExec:
            exec(cmd)
 
 

processPreInclude()

python.TransformUtils.processPreInclude	(		runArgs,
			flags )

Process preInclude from runtime arguments.

Definition at line 62 of file Tools/PyJobTransforms/python/TransformUtils.py.

def processPreInclude(runArgs, flags):
    """Process preInclude from runtime arguments."""
    if hasattr(runArgs, 'preInclude') and runArgs.preInclude and runArgs.preInclude != 'NONE':
        for fragment in runArgs.preInclude:
            executeFromFragment(fragment, flags)
 
 

SortInput()

python.TransformUtils.SortInput	(		flags,
			cfg )

Method that should be post-included after a typicaly configured Athena job
It will read original input files, create a list of Events in memory, sort them and produce
an intermediate Event Collection file that Athena will read instead of the original inputs
Event information is read from EventInfoTags (stored by default in all Athena data files)
The default sort key value (Lumi) can be oveerriden, as the sorting order
The intermediate Collection file can be inspected using CollQuery cmdline utility

Definition at line 125 of file Tools/PyJobTransforms/python/TransformUtils.py.

def SortInput(flags, cfg):
    """Method that should be post-included after a typicaly configured Athena job
    It will read original input files, create a list of Events in memory, sort them and produce
    an intermediate Event Collection file that Athena will read instead of the original inputs
    Event information is read from EventInfoTags (stored by default in all Athena data files)
    The default sort key value (Lumi) can be oveerriden, as the sorting order
    The intermediate Collection file can be inspected using CollQuery cmdline utility"""
    import os
    inputs = cfg.getService("EventSelector").InputCollections
 
    # set default sort parameters, read overrides from locals()
    tmpCollFile = locals().get("AthenaInputSortCollName", "sortedEventRefs" + str(os.getpid()) + ".root")
    sortTag     = locals().get("AthenaInputSortTag",      "LumiBlockN")
    sortOrd     = locals().get("AthenaInputSortOrder",    "Ascending")
 
    from CollectionSvc.SortedCollectionCreator import SortedCollectionCreator
    sorter = SortedCollectionCreator(name="SortEvents")
    # Sort Inputs based on one of the EventInfoTag attributes
    # Store sorted event collection in a temporary file
    # This should run as postInclude, so we assume EventSelector.InputCollections is set earlier
 
    # moved execution to a subprocess, because Gaudi messaging created by collections causes
    # AppManager errors
    from PyUtils import PoolFile
    rc = sorter.executeInSubprocess(inputs, outputCollection=tmpCollFile, outputCollectionType=PoolFile.PoolOpts.CollectionType.RootTTreeCollection, sortAttribute=sortTag, sortOrder=sortOrd)
    if rc != 0:
       msg.error(f"Sorting failed with exit code: {rc}")
 
    # Reading Events through References require a populated FileCatalog
    for inpfile in inputs:
        os.system('pool_insertFileToCatalog {}'.format(inpfile))
 
    cfg.getService("PoolSvc").AttemptCatalogPatch = False
 
    # Tell Athena to use the sorted collection instead of the original inputs
    cfg.getService("EventSelector").InputCollections = [tmpCollFile]
    cfg.getService("EventSelector").CollectionType = "RootCollection"
    return cfg

UseCREST()

python.TransformUtils.UseCREST

(

flags

)

PreInclude to switch to using CREST rather than COOL

Definition at line 105 of file Tools/PyJobTransforms/python/TransformUtils.py.

def UseCREST(flags):
    """PreInclude to switch to using CREST rather than COOL
    """
    msg.warning('CREST is now used by default for Run4 and beyond, Please remove this PreInclude from commands!')
    flags.IOVDb.UseCREST = True
    from os import environ
    msg.info('Enabling CREST DB access')
    if environ.get('CREST_SERVER'):
        flags.IOVDb.CrestServer = environ.get('CREST_SERVER')
    else:
        msg.info('CREST_SERVER environment variable not defined - using fall-back.')
    msg.info(f'Using CrestServer: {flags.IOVDb.CrestServer}')
 
 

UseFrontier()

python.TransformUtils.UseFrontier

(

flags

)

Switch database to using FRONTIER, but with a fallback
to DBRelease if FRONTIER_SERVER is undefined (e.g., on HPC)

Move from RecJobTransforms to PyJobTransforms to enable use
in simulation ATN and KV jobs

Definition at line 79 of file Tools/PyJobTransforms/python/TransformUtils.py.

def UseFrontier(flags):
    """Switch database to using FRONTIER, but with a fallback
    to DBRelease if FRONTIER_SERVER is undefined (e.g., on HPC)
 
    Move from RecJobTransforms to PyJobTransforms to enable use
    in simulation ATN and KV jobs
    """
    from os import environ
    from AthenaConfiguration.ComponentAccumulator import ComponentAccumulator
 
    cfg = ComponentAccumulator()
    
    if flags.IOVDb.UseCREST:
        msg.warning('Using PyJobTransforms.UseCREST (setting flags.IOVDb.UseCREST = True) and PyJobTransforms.UseFrontier in the same job is not supported!')
        return cfg
 
    if environ.get('FRONTIER_SERVER'):
        msg.info('Enabling FRONTIER DB access')
        from IOVDbSvc.IOVDbSvcConfig import DBReplicaSvcCfg
        cfg.merge(DBReplicaSvcCfg(flags, vetoDBRelease=True))
    else:
        msg.info('Using default DB access')
 
    return cfg
 
 

Variable Documentation

msg

python.TransformUtils.msg = logging.getLogger(__name__)

Definition at line 4 of file Tools/PyJobTransforms/python/TransformUtils.py.