python.TransformUtils Namespace Reference

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

Variables
	msg

Function Documentation

DumpPickle()

def python.TransformUtils.DumpPickle	(		flags,
			cfg
	)

Dump the pickle file for the current configuration

Definition at line 100 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()

def 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()

def 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()

def 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()

def 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()

def 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()

def 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()

def 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 106 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()) )
    sortTag     = locals().get("AthenaInputSortTag",      "LumiBlockN")
    sortOrd     = locals().get("AthenaInputSortOrder",    "Ascending")
 
    from CollectionUtilities.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
    sorter.execute(inputs, outputCollection=tmpCollFile, sortAttribute=sortTag, sortOrder=sortOrd)
 
    # Reading Events through References require a populated FileCatalog
    for inpfile in inputs:
        os.system('pool_insertFileToCatalog {}'.format(inpfile))
 
    # Tell Athena to use the sorted collection instead of the original inputs
    cfg.getService("EventSelector").InputCollections = [tmpCollFile + ".root"]
    cfg.getService("EventSelector").CollectionType = "ExplicitROOT"
    return cfg

UseFrontier()

def 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
    from AthenaConfiguration.ComponentFactory import CompFactory
 
    cfg = ComponentAccumulator()
    if environ.get('FRONTIER_SERVER'):
        msg.info('Enabling FRONTIER DB access')
        cfg.addService(CompFactory.DBReplicaSvc(COOLSQLiteVetoPattern='DBRelease'))
    else:
        msg.info('Using default DB access')
 
    return cfg
 
 

Variable Documentation

msg

python.TransformUtils.msg

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