python.Utils Namespace Reference

Namespaces
namespace	unixtools

Classes
class	Campaign

Functions
	loadDefaultComps (allcomps)
	exposeHandles (allcomps)
	setupLoggingLevels (flags, ca)
	getRunFromName (name, default='', asInt=False)
	fileListSnippet (files, dsName, taskName, jobDir=None)
	blankIfNone (s)
	getUserName (default='UNKNOWN')
	getHostName ()
	getMCCampaign (files)
	getDataYear (flags)
	getEDMVersionFromBS (filename)
	edmDictToList (edmDict)
	edmListToDict (edmList)
	getEDMListFromWriteHandles (configurables)

Variables
str	__version__ = '1.0.0'
str	__author__ = 'Wim Lavrijsen (WLavrijsen@lbl.gov)'
list	__all__ = [ 'unixtools' ]
dict	campaign_runs
	log = logging.getLogger('TrigEDMConfig')

Detailed Description

This module defines miscellaneous utility functions.

Function Documentation

blankIfNone()

python.Utils.blankIfNone

(

s

)

Definition at line 41 of file InnerDetector/InDetExample/InDetBeamSpotExample/python/Utils.py.

def blankIfNone(s):
    if s:
        return s
    else:
        return ''
 
 

edmDictToList()

python.Utils.edmDictToList

(

edmDict

)

Convert EDM dictionary in the format:
    {'type1': ['key1','key2'], 'type2': ['key3']}
to a flat list in the format:
    ['type1#key1', 'type1#key2', 'type2#key3']

Definition at line 68 of file Trigger/TriggerCommon/TrigEDMConfig/python/Utils.py.

def edmDictToList(edmDict):
    '''
    Convert EDM dictionary in the format:
        {'type1': ['key1','key2'], 'type2': ['key3']}
    to a flat list in the format:
        ['type1#key1', 'type1#key2', 'type2#key3']
    '''
    return [ f"{type}#{name}" for type, names in edmDict.items() for name in names ]
 
 

edmListToDict()

python.Utils.edmListToDict

(

edmList

)

Convert EDM list in the format:
    ['type1#key1', 'type1#key2', 'type2#key3']
to a dictionary in the format:
    {'type1': ['key1','key2'], 'type2': ['key3']}

Definition at line 78 of file Trigger/TriggerCommon/TrigEDMConfig/python/Utils.py.

def edmListToDict(edmList):
    '''
    Convert EDM list in the format:
        ['type1#key1', 'type1#key2', 'type2#key3']
    to a dictionary in the format:
        {'type1': ['key1','key2'], 'type2': ['key3']}
    '''
    edmDict = defaultdict(list)
    for typeName in edmList:
        edmType, edmKey = typeName.split('#')
        edmDict[edmType].append(edmKey)
    return edmDict
 
 

exposeHandles()

python.Utils.exposeHandles

(

allcomps

)

Sets all handle keys explicitly

Definition at line 36 of file Control/AthenaConfiguration/python/Utils.py.

def exposeHandles(allcomps):
    """Sets all handle keys explicitly"""
    def __getDefault(d):
        if isinstance(d, collections.abc.Sequence):
            return [el  for el in d]
        else:
            return d.Path
 
    for comp in allcomps:
        for propName, value in comp._descriptors.items():
            if propName in comp._properties: continue
            if "HandleKey" in value.cpp_type:
                comp._properties[propName] = __getDefault(value.default)
 

fileListSnippet()

python.Utils.fileListSnippet	(		files,
			dsName,
			taskName,
			jobDir = None )

Definition at line 30 of file InnerDetector/InDetExample/InDetBeamSpotExample/python/Utils.py.

def fileListSnippet(files,dsName,taskName,jobDir=None):
    s = ''
    if files:
        for f in files.split():
            if jobDir and not os.path.exists('%s/%s/%s/%s' % (jobDir,dsName,taskName,f)):
                s += '%s (archived)<br>' % f
            else:
                s += '<a href="/jobfiles/%s/%s/%s">%s</a><br>' % (dsName,taskName,f,f)
    return s
 
 

getDataYear()

python.Utils.getDataYear

(

flags

)

Definition at line 71 of file Tools/Campaigns/python/Utils.py.

def getDataYear(flags):
    if flags.Input.isMC:
        campaign = getMCCampaign(flags.Input.Files)
        dataYear = {
            Campaign.MC20a: 2016, # prefer over 2015
            Campaign.MC20d: 2017,
            Campaign.MC20e: 2018,
            Campaign.MC21a: 2022,
            Campaign.MC23a: 2022,
            Campaign.MC23c: 2023,
            Campaign.MC23d: 2023,
            Campaign.MC23e: 2024,
            Campaign.MC23g: 2025,
            Campaign.PhaseII: 2030,
        }[campaign]
    else:
        dataYear = flags.Input.DataYear
    return dataYear

getEDMListFromWriteHandles()

python.Utils.getEDMListFromWriteHandles

(

configurables

)

Build OutputStream ItemList from all WriteHandles in a list of components (configurables),
for example a list of AlgTools. The output is in flat list format:
    ['type1#key1', 'type1#key2', 'type2#key3']

Definition at line 92 of file Trigger/TriggerCommon/TrigEDMConfig/python/Utils.py.

def getEDMListFromWriteHandles(configurables):
    '''
    Build OutputStream ItemList from all WriteHandles in a list of components (configurables),
    for example a list of AlgTools. The output is in flat list format:
        ['type1#key1', 'type1#key2', 'type2#key3']
    '''
 
    def getWriteHandles(comp):
        properties = [getattr(comp,propName) for propName in comp.getDefaultProperties().keys()]
        return [prop for prop in properties if isinstance(prop,DataHandle) and prop.mode()=='W']
 
    def formatItem(containerType, containerKey):
        auxType = containerType.replace('Container','AuxContainer')
        return [f'{containerType}#{containerKey}',
                f'{auxType}#{containerKey}Aux.-']
 
    def containerTypedef(containerType):
        if containerType.startswith('xAOD::') and containerType.endswith('Container'):
            # Already the right typedef
            return containerType
        m = re.match(r'DataVector<xAOD::(\w+)_v[0-9]+,', containerType)
        if m and len(m.groups()) > 0:
            return f'xAOD::{m.group(1)}Container'
        raise RuntimeError(f'Failed to convert type {containerType} into a container typedef')
 
    def itemListFromConfigurable(comp):
        items = []
        for handle in getWriteHandles(comp):
            sgKey = handle.Path.replace('StoreGateSvc+','')
            if not sgKey:
                continue
            items += formatItem(containerTypedef(handle.Type), handle.Path)
        return items
 
    itemList = []
    for comp in configurables:
        itemList += itemListFromConfigurable(comp)
    return itemList

getEDMVersionFromBS()

python.Utils.getEDMVersionFromBS

(

filename

)

Determine Trigger EDM version based on the input ByteStream file.

Run-3 EDM is indicated by HLT ROD version > 1.0. For Run 1 and 2 the
HLT ROD version was 0.0 and the run number is used to disambiguate between them.

Definition at line 11 of file Trigger/TriggerCommon/TrigEDMConfig/python/Utils.py.

def getEDMVersionFromBS(filename):
   """Determine Trigger EDM version based on the input ByteStream file.
 
   Run-3 EDM is indicated by HLT ROD version > 1.0. For Run 1 and 2 the
   HLT ROD version was 0.0 and the run number is used to disambiguate between them.
   """
 
   boundary_run12 = 230000
   boundary_run23 = 368000
 
   import eformat
   from libpyeformat_helper import SubDetector
   bs = eformat.istream(filename)
 
   rodVersionM = None
   rodVersionL = None
   # Find the first HLT ROBFragment in the first event
   for robf in bs[0]:
       if robf.rob_source_id().subdetector_id()==SubDetector.TDAQ_HLT:
           rodVersionM = robf.rod_minor_version() >> 8
           rodVersionL = robf.rod_minor_version() & 0xFF
           log.debug("HLT ROD minor version from input file is %d.%d", rodVersionM, rodVersionL)
           break
 
   # Case 1: failed to read ROD version
   if rodVersionM is None or rodVersionL is None:
       log.warning("Cannot determine HLT ROD version from input file, falling back to run-number-based decision")
   # Case 2: ROD version indicating Run 3
   elif rodVersionM >= 1:
       log.info("Determined EDMVersion to be 3, because running on BS file with HLT ROD version %d.%d",
                 rodVersionM, rodVersionL)
       return 3
 
   # Case 3: ROD version indicating Run 1 or 2 - use run number to disambiguate
   runNumber = bs[0].run_no()
   log.debug("Read run number %s", runNumber)
 
   if not runNumber or runNumber <= 0:
       log.warning("Cannot determine EDM version because run number %s is invalid. ", runNumber)
       return None
   elif runNumber < boundary_run12:
       # Run-1 data
       log.info("Determined EDMVersion to be 1 based on BS file run number (runNumber < %d)",
                 boundary_run12)
       return 1
   elif runNumber < boundary_run23:
       # Run-2 data
       log.info("Determined EDMVersion to be 2 based on BS file run number (%d < runNumber < %d)",
                 boundary_run12, boundary_run23)
       return 2
   else:
       # Run-3 data
       log.info("Determined EDMVersion to be 3 based on BS file run number (runNumber > %d)",
                 boundary_run23)
       return 3
 
 

getHostName()

python.Utils.getHostName

(

)

Definition at line 60 of file InnerDetector/InDetExample/InDetBeamSpotExample/python/Utils.py.

def getHostName():
    import platform
    return platform.node()

getMCCampaign()

python.Utils.getMCCampaign

(

files

)

Definition at line 38 of file Tools/Campaigns/python/Utils.py.

def getMCCampaign(files):
    # Auto-configure from file
    from AthenaConfiguration.AutoConfigFlags import GetFileMD
    metadata = GetFileMD(files)
    mc_campaign = Campaign(metadata.get('mc_campaign', ''))
    project_name = metadata.get('project_name', '')
    run_numbers = metadata.get('runNumbers', [])
 
    if mc_campaign is not Campaign.Unknown:
        return mc_campaign
 
    if run_numbers:
        mc_campaign = campaign_runs.get(run_numbers[0], Campaign.Unknown)
 
    # MC-equivalent projects for data
    if 'data25' in project_name:
        return Campaign.MC23g
    elif 'data24' in project_name:
        return Campaign.MC23e
    elif 'data23' in project_name:
        return Campaign.MC23c
    elif 'data22' in project_name:
        return Campaign.MC21a
    elif 'data18' in project_name:
        return Campaign.MC20e
    elif 'data17' in project_name:
        return Campaign.MC20d
    elif 'data16' in project_name or 'data15' in project_name:
        return Campaign.MC20a
 
    return mc_campaign
 
# Get data year corresponding to MC campaign

getRunFromName()

python.Utils.getRunFromName	(		name,
			default = '',
			asInt = False )

Extract the run number from a file name whose first part is a standard dataset name.
   If the run number cannot be determined, returns the default value, which, by
   default is an empty string.

Definition at line 13 of file InnerDetector/InDetExample/InDetBeamSpotExample/python/Utils.py.

def getRunFromName(name,default='',asInt=False):
    """Extract the run number from a file name whose first part is a standard dataset name.
       If the run number cannot be determined, returns the default value, which, by
       default is an empty string."""
    name = os.path.basename(name)
    try:
        run = re.sub('^0*','',name.split('.')[1])
        if not re.search(r'^\d+$',run):
            # Probably wasn't the run number, so use the default instead
            run = default
        else:
            run = int(run) if asInt else run
    except Exception:
        run = default
    return run
 
 

getUserName()

python.Utils.getUserName

(

default = 'UNKNOWN'

)

Get login name in a platform-independent manner.

Definition at line 48 of file InnerDetector/InDetExample/InDetBeamSpotExample/python/Utils.py.

def getUserName(default='UNKNOWN'):
    """Get login name in a platform-independent manner."""
    user = ''
    try:
        user = os.getlogin()     # this doesn't seem to work with acrontab
    except Exception:
        pass
    if not user:
        user = os.getenv('USER',default)
    return user
 
 

loadDefaultComps()

python.Utils.loadDefaultComps

(

allcomps

)

Attempts to load all default components (those that are not actually configured)

Definition at line 10 of file Control/AthenaConfiguration/python/Utils.py.

def loadDefaultComps(allcomps):
    """Attempts to load all default components (those that are not actually configured)"""
    loadedSome = False
    def __load(comp, prop):
        descr = comp._descriptors[prop]
        if descr.default.getType() == "":
            return
        try:
            childComp = CompFactory.getComp(descr.default.getType())(descr.default.getName())
            comp._properties[prop] = childComp
            nonlocal loadedSome
            loadedSome = True
        except Exception:
            msg=logging.getLogger('loadDefaultComps')
            msg.warning("Default component %s can not be loaded", descr.default.typeAndName )
            pass
 
    for comp in allcomps:
        for propName,value in comp._descriptors.items():
            if propName in comp._properties: continue
            if isinstance(value.default, GaudiHandle):
                __load(comp, propName )
                        
    if loadedSome:
        loadDefaultComps(allcomps)
 

setupLoggingLevels()

python.Utils.setupLoggingLevels	(		flags,
			ca )

Read the Exec.*MessageComponents flags and modify OutputLevel of component(s).

The specification of components uses the Python `fnmatch` library and resembles UNIX paths.
An event algorithm MyAlgo/MyInstance has the following path:
    MasterSeq/AthAllAlgSeq/AthAlgSeq/MyAlgo/MyInstance
A private tool MyTool of name ToolInstance used by that algorithm:
    MasterSeq/AthAllAlgSeq/AthAlgSeq/MyAlgo/MyInstance/MyTool/ToolInstance
A public tool:
    ToolSvc/MyTool/ToolInstance

The path specification can take the following forms:
    '*/ToolInstance'      : all tools that have matching instance name
    '*/MyTool/*'          : all instances of type MyTool
    '*/MyAlgo/MyInstance' : specific algorithm instance
    '*/MyAlgo/*'          : all instances of the specific algorithm class
    '*/AthAlgSeq/*'       : all algorithms of the given sequence
    'ToolSvc/My*/*'       : all public tools with instance name starting with "My"

The modifications to the OutputLevel are applied in the order ERROR to VERBOSE, i.e.
it is possible to set higher verbosities with more specific selections.

Each setting can be either a string or a list of strings. If the component path contains
no '/' it is assumed to be a plain component name. In this case, the OutputLevel is set
using the property MessageSvc.setDebug or equivalent. This works also for converters, which
do not have any properties.

Definition at line 50 of file Control/AthenaConfiguration/python/Utils.py.

def setupLoggingLevels(flags, ca):
    """Read the Exec.*MessageComponents flags and modify OutputLevel of component(s).
    
    The specification of components uses the Python `fnmatch` library and resembles UNIX paths.
    An event algorithm MyAlgo/MyInstance has the following path:
        MasterSeq/AthAllAlgSeq/AthAlgSeq/MyAlgo/MyInstance
    A private tool MyTool of name ToolInstance used by that algorithm:
        MasterSeq/AthAllAlgSeq/AthAlgSeq/MyAlgo/MyInstance/MyTool/ToolInstance
    A public tool:
        ToolSvc/MyTool/ToolInstance
 
    The path specification can take the following forms:
        '*/ToolInstance'      : all tools that have matching instance name
        '*/MyTool/*'          : all instances of type MyTool
        '*/MyAlgo/MyInstance' : specific algorithm instance
        '*/MyAlgo/*'          : all instances of the specific algorithm class
        '*/AthAlgSeq/*'       : all algorithms of the given sequence
        'ToolSvc/My*/*'       : all public tools with instance name starting with "My"
 
    The modifications to the OutputLevel are applied in the order ERROR to VERBOSE, i.e.
    it is possible to set higher verbosities with more specific selections.
 
    Each setting can be either a string or a list of strings. If the component path contains
    no '/' it is assumed to be a plain component name. In this case, the OutputLevel is set
    using the property MessageSvc.setDebug or equivalent. This works also for converters, which
    do not have any properties.
    """
 
    def __tolist(d):
        return ([d] if d else []) if isinstance(d, str) else d
 
    for flag_val, level in ((flags.Exec.ErrorMessageComponents, 'ERROR'),
                            (flags.Exec.WarningMessageComponents, 'WARNING'),
                            (flags.Exec.InfoMessageComponents, 'INFO'),
                            (flags.Exec.DebugMessageComponents, 'DEBUG'),
                            (flags.Exec.VerboseMessageComponents, 'VERBOSE')):
        for c in __tolist(flag_val):
            if "/" in c:
                ca.foreach_component(c).OutputLevel = getattr(Constants, level)
 
            # a plain component name is given
            else:
                msgSvc = ca.getService("MessageSvc")  # should exist by the time this code runs
                getattr(msgSvc, f'set{level.title()}').append(c)  # e.g. setDebug property

Variable Documentation

__all__

list python.Utils.__all__ = [ 'unixtools' ]

private

Definition at line 9 of file Control/AthenaCommon/python/Utils/__init__.py.

__author__

str python.Utils.__author__ = 'Wim Lavrijsen (WLavrijsen@lbl.gov)'

private

Definition at line 7 of file Control/AthenaCommon/python/Utils/__init__.py.

__version__

str python.Utils.__version__ = '1.0.0'

private

Definition at line 6 of file Control/AthenaCommon/python/Utils/__init__.py.

campaign_runs

dict python.Utils.campaign_runs

Initial value:

=  {
    284500: Campaign.MC20a,
    300000: Campaign.MC20d,
    310000: Campaign.MC20e,
    410000: Campaign.MC21a,
}

Definition at line 30 of file Tools/Campaigns/python/Utils.py.

log

python.Utils.log = logging.getLogger('TrigEDMConfig')

Definition at line 8 of file Trigger/TriggerCommon/TrigEDMConfig/python/Utils.py.