python.AutoConfigFlags Namespace Reference

Classes
class	DynamicallyLoadMetadata

Functions
def	GetFileMD (filenames, allowEmpty=True, maxLevel='peeker')
def	_initializeGeometryParameters (geoTag, sqliteDB, sqliteDBFullPath)
def	DetDescrInfo (geoTag, sqliteDB, sqliteDBFullPath)
def	getDefaultDetectors (geoTag, sqliteDB, sqliteDBFullPath, includeForward=False)
def	getRunToTimestampDict ()
def	getInitialTimeStampsFromRunNumbers (runNumbers)
def	getGeneratorsInfo (flags)
def	getSpecialConfigurationMetadata (flags)

Variables
	msg = logging.getLogger('AutoConfigFlags')
	_fileMetaData = dict()

Function Documentation

_initializeGeometryParameters()

def python.AutoConfigFlags._initializeGeometryParameters (   geoTag,   sqliteDB,   sqliteDBFullPath  )

private

Read geometry database for all detectors

Definition at line 86 of file AutoConfigFlags.py.

def _initializeGeometryParameters(geoTag,sqliteDB,sqliteDBFullPath):
    """Read geometry database for all detectors"""
 
    from AtlasGeoModel import CommonGeoDB
    from PixelGeoModel import PixelGeoDB
    from LArGeoAlgsNV import LArGeoDB
    from MuonGeoModel import MuonGeoDB
 
    if not sqliteDB:
        # Read parameters from Oracle/Frontier
        from AtlasGeoModel.AtlasGeoDBInterface import AtlasGeoDBInterface
        dbGeomCursor = AtlasGeoDBInterface(geoTag)
        dbGeomCursor.ConnectAndBrowseGeoDB()
 
        params = { 'Common' : CommonGeoDB.InitializeGeometryParameters(dbGeomCursor),
                   'Pixel' : PixelGeoDB.InitializeGeometryParameters(dbGeomCursor),
                   'LAr' : LArGeoDB.InitializeGeometryParameters(dbGeomCursor),
                   'Muon' : MuonGeoDB.InitializeGeometryParameters(dbGeomCursor),
                   'Luminosity' : CommonGeoDB.InitializeLuminosityDetectorParameters(dbGeomCursor),
               }
 
        msg.debug('Config parameters retrieved from Geometry DB (Frontier/Oracle):')
        for key in params.keys():
            msg.debug(f'{key} -> {params[key]}')
    else:
        # Read parameters from SQLite
        from AtlasGeoModel.AtlasGeoDBInterface import AtlasGeoDBInterface_SQLite
        sqliteReader = AtlasGeoDBInterface_SQLite(geoTag,sqliteDBFullPath)
        sqliteReader.ConnectToDB()
 
        params = { 'Common' : CommonGeoDB.InitializeGeometryParameters_SQLite(sqliteReader),
                   'Pixel' : PixelGeoDB.InitializeGeometryParameters_SQLite(sqliteReader),
                   'LAr' : LArGeoDB.InitializeGeometryParameters_SQLite(sqliteReader),
                   'Muon' : MuonGeoDB.InitializeGeometryParameters_SQLite(sqliteReader),
                   'Luminosity' : CommonGeoDB.InitializeLuminosityDetectorParameters_SQLite(sqliteReader),
               }
 
        msg.debug('Config parameters retrieved from Geometry DB (SQLite):')
        for key in params.keys():
            msg.debug(f'{key} -> {params[key]}')
 
    return params
 
 
@lru_cache(maxsize=4)  # maxsize=1 should be enough for most jobs

DetDescrInfo()

def python.AutoConfigFlags.DetDescrInfo	(		geoTag,
			sqliteDB,
			sqliteDBFullPath
	)

Query geometry DB for detector description. Returns dictionary with
detector description. Queries DB for each tag only once.

geoTag: geometry tag (e.g. ATLAS-R2-2016-01-00-01)

Definition at line 131 of file AutoConfigFlags.py.

def DetDescrInfo(geoTag, sqliteDB, sqliteDBFullPath):
    """Query geometry DB for detector description. Returns dictionary with
    detector description. Queries DB for each tag only once.
 
    geoTag: geometry tag (e.g. ATLAS-R2-2016-01-00-01)
    """
    if not geoTag:
        raise ValueError("No geometry tag specified")
 
    detDescrInfo = _initializeGeometryParameters(geoTag,sqliteDB,sqliteDBFullPath)
    detDescrInfo["geomTag"] = geoTag
    return detDescrInfo
 
 
@lru_cache(maxsize=4)  # maxsize=1 should be enough for most jobs

getDefaultDetectors()

def python.AutoConfigFlags.getDefaultDetectors	(		geoTag,
			sqliteDB,
			sqliteDBFullPath,
			includeForward = False
	)

Query geometry DB for detector description.
Returns a set of detectors used in a geometry tag.

geoTag: geometry tag (e.g. ATLAS-R2-2016-01-00-01)

Definition at line 146 of file AutoConfigFlags.py.

def getDefaultDetectors(geoTag, sqliteDB, sqliteDBFullPath, includeForward=False):
    """Query geometry DB for detector description.
    Returns a set of detectors used in a geometry tag.
 
    geoTag: geometry tag (e.g. ATLAS-R2-2016-01-00-01)
    """
    detectors = DetDescrInfo(geoTag,sqliteDB,sqliteDBFullPath)['Common']['Detectors']
    
    manualConfig = not DetDescrInfo(geoTag,sqliteDB,sqliteDBFullPath)['Common']["DetectorsConfigured"]
    
    
    if manualConfig: 
        detectors.add('Bpipe')
        if DetDescrInfo(geoTag,sqliteDB,sqliteDBFullPath)['Common']['Run'] not in ['RUN1', 'RUN2', 'RUN3']: # RUN4 and beyond
            detectors.add('ITkPixel')
            detectors.add('ITkStrip')
            if DetDescrInfo(geoTag,sqliteDB,sqliteDBFullPath)['Luminosity']['BCMPrime']:
                detectors.add('BCMPrime')  # since https://gitlab.cern.ch/Atlas-Inner-Tracking/ITKLayouts/-/merge_requests/347 introduced BCMPrime geometry
            if DetDescrInfo(geoTag,sqliteDB,sqliteDBFullPath)['Luminosity']['PLR']:
                detectors.add('PLR')
        else:
            detectors.add('Pixel')
            detectors.add('SCT')
            detectors.add('TRT')
            detectors.add('BCM')
 
        if DetDescrInfo(geoTag,sqliteDB,sqliteDBFullPath)['Common']['Run'] not in ['RUN1', 'RUN2', 'RUN3']: # RUN4 and beyond
            detectors.add('HGTD')
 
        detectors.add('LAr')
        detectors.add('Tile')
 
    
    if DetDescrInfo(geoTag,sqliteDB,sqliteDBFullPath)['Common']['Run'] in ['RUN1', 'RUN2', 'RUN3']:
        detectors.add('MBTS')
 
    if DetDescrInfo(geoTag,sqliteDB,sqliteDBFullPath)['Muon']['HasMDT']:
        detectors.add('MDT')
    if DetDescrInfo(geoTag,sqliteDB,sqliteDBFullPath)['Muon']['HasRPC']:
         detectors.add('RPC')
    if DetDescrInfo(geoTag,sqliteDB,sqliteDBFullPath)['Muon']['HasTGC']:
        detectors.add('TGC')
    if DetDescrInfo(geoTag,sqliteDB,sqliteDBFullPath)['Muon']['HasCSC']:
        detectors.add('CSC')
    if DetDescrInfo(geoTag,sqliteDB,sqliteDBFullPath)['Muon']['HasSTGC']:
        detectors.add('sTGC')
    if DetDescrInfo(geoTag,sqliteDB,sqliteDBFullPath)['Muon']['HasMM']:
        detectors.add('MM')
 
    if includeForward:
        detectors.add('Lucid')
        if DetDescrInfo(geoTag,sqliteDB,sqliteDBFullPath)['Common']['Run'] not in ['RUN1']:
            detectors.add('AFP')
        detectors.add('ZDC')
        detectors.add('ALFA')
        detectors.add('FwdRegion')
 
    return detectors
 
 
# Based on RunDMCFlags.py

GetFileMD()

def python.AutoConfigFlags.GetFileMD	(		filenames,
			allowEmpty = True,
			maxLevel = 'peeker'
	)

Definition at line 65 of file AutoConfigFlags.py.

def GetFileMD(filenames, allowEmpty=True, maxLevel='peeker'):
    if not filenames or '_ATHENA_GENERIC_INPUTFILE_NAME_' in filenames:
        if allowEmpty:
            msg.info("Running an input-less job. Will have empty metadata.")
            return {}
        raise RuntimeError("Metadata can not be read in an input-less job.")
    if isinstance(filenames, str):
        filenames = [filenames]
    for filename in filenames:
        if filename not in _fileMetaData:
            msg.info("Obtaining metadata of auto-configuration by peeking into '%s'", filename)
            _fileMetaData[filename] = DynamicallyLoadMetadata(filename, maxLevel)
        if _fileMetaData[filename].maxAccessLevel != maxLevel:
            _fileMetaData[filename].maxAccessLevel = maxLevel
        if _fileMetaData[filename]['nentries'] not in [None, 0]: 
            return _fileMetaData[filename]
        else:
            msg.info("The file: %s has no entries, going to the next one for harvesting the metadata", filename)
    msg.info("No file with events found, returning anyways metadata associated to the first file %s", filenames[0])
    return _fileMetaData[filenames[0]]
 

getGeneratorsInfo()

def python.AutoConfigFlags.getGeneratorsInfo

(

flags

)

Read in GeneratorsInfo from the input file

Definition at line 223 of file AutoConfigFlags.py.

def getGeneratorsInfo(flags):
    """Read in GeneratorsInfo from the input file
    """
    from AthenaConfiguration.Enums import ProductionStep
    inputFiles = flags.Input.Files
    if flags.Common.ProductionStep in [ProductionStep.Overlay, ProductionStep.FastChain] and flags.Input.SecondaryFiles and not flags.Overlay.ByteStream:
        # Do something special for MC Overlay
        inputFiles = flags.Input.SecondaryFiles
    generatorsString = ""
    from AthenaConfiguration.AutoConfigFlags import GetFileMD
    if inputFiles:
        generatorsString = GetFileMD(inputFiles).get("generators", "")
    from GeneratorConfig.Versioning import generatorsGetFromMetadata
    return generatorsGetFromMetadata( generatorsString )
 
 

getInitialTimeStampsFromRunNumbers()

def python.AutoConfigFlags.getInitialTimeStampsFromRunNumbers

(

runNumbers

)

This is used to hold a dictionary of the form
{152166:1269948352889940910, ...} to allow the
timestamp to be determined from the run.

Definition at line 213 of file AutoConfigFlags.py.

def getInitialTimeStampsFromRunNumbers(runNumbers):
    """This is used to hold a dictionary of the form
    {152166:1269948352889940910, ...} to allow the
    timestamp to be determined from the run.
    """
    run2timestampDict =  getRunToTimestampDict()
    timeStamps = [run2timestampDict.get(runNumber,1) for runNumber in runNumbers] # Add protection here?
    return timeStamps
 
 

getRunToTimestampDict()

def python.AutoConfigFlags.getRunToTimestampDict

(

)

Definition at line 207 of file AutoConfigFlags.py.

def getRunToTimestampDict():
    # this wrapper is intended to avoid an initial import
    from .RunToTimestampData import RunToTimestampDict
    return RunToTimestampDict
 
 

getSpecialConfigurationMetadata()

def python.AutoConfigFlags.getSpecialConfigurationMetadata

(

flags

)

Read in special simulation job option fragments based on metadata
passed by the evgen stage

Definition at line 239 of file AutoConfigFlags.py.

def getSpecialConfigurationMetadata(flags):
    """Read in special simulation job option fragments based on metadata
    passed by the evgen stage
    """
    specialConfigDict = dict()
    legacyPreIncludeToCAPostInclude = { 'SimulationJobOptions/preInclude.AMSB.py' : 'Charginos.CharginosConfig.AMSB_Cfg',
                                        'SimulationJobOptions/preInclude.Monopole.py' :  'Monopole.MonopoleConfig.MonopoleCfg',
                                        'SimulationJobOptions/preInclude.Quirks.py' : 'Quirks.QuirksConfig.QuirksCfg',
                                        'SimulationJobOptions/preInclude.SleptonsLLP.py' : 'Sleptons.SleptonsConfig.SleptonsLLPCfg',
                                        'SimulationJobOptions/preInclude.GMSB.py' : 'Sleptons.SleptonsConfig.GMSB_Cfg',
                                        'SimulationJobOptions/preInclude.Qball.py' : 'Monopole.MonopoleConfig.QballCfg',
                                        'SimulationJobOptions/preInclude.RHadronsPythia8.py' : 'RHadrons.RHadronsConfig.RHadronsCfg',
                                        'SimulationJobOptions/preInclude.fcp.py' : 'Monopole.MonopoleConfig.fcpCfg' }
    legacyPreIncludeToCAPreInclude = { 'SimulationJobOptions/preInclude.AMSB.py' : None,
                                       'SimulationJobOptions/preInclude.Monopole.py' :  'Monopole.MonopoleConfig.MonopolePreInclude',
                                       'SimulationJobOptions/preInclude.Quirks.py' : None,
                                       'SimulationJobOptions/preInclude.SleptonsLLP.py' : None,
                                       'SimulationJobOptions/preInclude.GMSB.py' : None,
                                       'SimulationJobOptions/preInclude.Qball.py' : 'Monopole.MonopoleConfig.QballPreInclude',
                                       'SimulationJobOptions/preInclude.RHadronsPythia8.py' : 'RHadrons.RHadronsConfig.RHadronsPreInclude',
                                       'SimulationJobOptions/preInclude.fcp.py' : 'Monopole.MonopoleConfig.fcpPreInclude' }
    specialConfigString = ''
    from AthenaConfiguration.Enums import ProductionStep
    inputFiles = flags.Input.Files
    secondaryInputFiles = flags.Input.SecondaryFiles
    if flags.Common.ProductionStep in [ProductionStep.Overlay, ProductionStep.FastChain] and not flags.Overlay.DataOverlay and flags.Input.SecondaryFiles:
        # Do something special for MC Overlay
        inputFiles = flags.Input.SecondaryFiles
        secondaryInputFiles = flags.Input.Files
    from AthenaConfiguration.AutoConfigFlags import GetFileMD
    if len(inputFiles)>0:
        specialConfigString = GetFileMD(inputFiles).get('specialConfiguration', '')
    if (not len(specialConfigString) or specialConfigString == 'NONE') and len(secondaryInputFiles)>0:
        # If there is no specialConfiguration metadata in the primary
        # input try the secondary inputs (MC Overlay case)
        specialConfigString = GetFileMD(secondaryInputFiles).get('specialConfiguration', '')
    if len(specialConfigString)>0:
        
        spcitems = specialConfigString.split(";")
        for spcitem in spcitems:
            #print spcitem
            
            if not spcitem or spcitem.upper() == "NONE":
                continue
            
            if "=" not in spcitem:
                spcitem = "preInclude=" + spcitem
            
            k, v = spcitem.split("=")
            if k == "preInclude" and v.endswith('.py'): # Translate old preIncludes into CA-based versions.
                if v == 'SimulationJobOptions/preInclude.RhadronsPythia8.py':
                    v = 'SimulationJobOptions/preInclude.RHadronsPythia8.py' # ATLASSIM-6687 Fixup for older EVNT files
                v1 = legacyPreIncludeToCAPreInclude[v]
                if v1 is not None:
                    specialConfigDict[k] = v1
                v2 = legacyPreIncludeToCAPostInclude[v]
                if v2 is not None:
                    specialConfigDict['postInclude'] = v2
            else:
                specialConfigDict[k] = v
    return specialConfigDict
 

Variable Documentation

_fileMetaData

python.AutoConfigFlags._fileMetaData = dict()

private

Definition at line 10 of file AutoConfigFlags.py.

msg

python.AutoConfigFlags.msg = logging.getLogger('AutoConfigFlags')

Definition at line 7 of file AutoConfigFlags.py.