python.TriggerConfigAccess Namespace Reference

Functions
dict[str, int]	getKeysFromCool (int runNr, int lbNr=0)
Optional[dict[str, Any]]	getDBKeysFromMetadata (flags)
Optional[dict[str, Any]]	_getJSONFromMetadata (flags, key)
L1MenuAccess	getL1MenuAccess (flags=None)
L1PrescalesSetAccess	getL1PrescalesSetAccess (flags=None)
BunchGroupSetAccess	getBunchGroupSetAccess (flags=None)
HLTMenuAccess	getHLTMenuAccess (flags=None)
HLTPrescalesSetAccess	getHLTPrescalesSetAccess (flags=None)
HLTJobOptionsAccess	getHLTJobOptionsAccess (flags=None)
HLTMonitoringAccess	getHLTMonitoringAccess (flags=None)

Variables
	log
	maxsize

Function Documentation

_getJSONFromMetadata()

Optional[dict[str,Any]] python.TriggerConfigAccess._getJSONFromMetadata (   flags,   key  )

private

Definition at line 105 of file TriggerConfigAccess.py.

def _getJSONFromMetadata(flags, key) -> Optional[dict[str,Any]]:
    if flags.Input.Format != Format.POOL:
        raise RuntimeError("Cannot read trigger configuration (%s) from input type %s", key, flags.Input.Format)
    from AthenaConfiguration.AutoConfigFlags import GetFileMD
    metadata = GetFileMD(flags.Input.Files)
    menu_json = metadata.get(key, None)
    if menu_json is None:
        menu_json = metadata.get('DataVector<xAOD::TriggerMenuJson_v1>_%s' % key, None)
    if menu_json is None:
        if key == 'TriggerMenuJson_HLTMonitoring':
            # currently the HLT Monitoring information is missing from many pool files' metadata
            log.info("Trigger metadata with key 'TriggerMenuJson_HLTMonitoring' is not available in this file. This feature is currently being introduced.")
            return None
        else:
            raise RuntimeError("Cannot read trigger configuration (%s) from input file metadata" % key)
    return menu_json
 
 
"""
 
L1 information
 
"""
@AccumulatorCache

getBunchGroupSetAccess()

BunchGroupSetAccess python.TriggerConfigAccess.getBunchGroupSetAccess

(

flags = None

)

Definition at line 169 of file TriggerConfigAccess.py.

def getBunchGroupSetAccess( flags = None ) -> BunchGroupSetAccess:
    tc = getTrigConfigFromFlag( flags )
    if tc["SOURCE"] == "FILE":
        cfg = BunchGroupSetAccess( filename = getBunchGroupSetFileName( flags ) )
    elif tc["SOURCE"] == "COOL":
        """This is the case when reconstructing the data."""
        if len(flags.Input.RunNumbers) == 0:
            raise RuntimeError("No run number available in input metadata")
        keysFromCool = getKeysFromCool( flags.Input.RunNumbers[0] )
        cfg = BunchGroupSetAccess( dbalias = keysFromCool["DB"], bgskey = keysFromCool['BGSK'] )
    elif tc["SOURCE"] == "DB":
        cfg = BunchGroupSetAccess( dbalias = tc["DBCONN"], bgskey = tc["BGSK"] )
    elif tc["SOURCE"] == "INFILE":
        if flags.Input.Format != Format.POOL:
            raise RuntimeError(f"Cannot read trigger configuration (Bunchgroup Set) from input type {flags.Input.Format}")
        raise NotImplementedError("Python access to the trigger configuration (Bunchgroup Set) from in-file metadata not yet implemented")
    else:
        raise RuntimeError("Unknown source of trigger configuration: %s" % tc["SOURCE"])
    return cfg
 
 
"""
 
HLT information
 
"""
@AccumulatorCache

getDBKeysFromMetadata()

Optional[dict[str, Any]] python.TriggerConfigAccess.getDBKeysFromMetadata

(

flags

)

Provides access to the database keys from the in-file metadata

Gets the database keys from the in-file metadata which are stored together with the json representation
If the keys are in the file, then usually SMK, L1PSK and HLTPSK are present.
The bunchgroupset is not stored in the metadata, so 0 is returned for the BGS key for completeness.

@returns: dictionary with the DB keys. Returns 'None' if information is not present.

Definition at line 81 of file TriggerConfigAccess.py.

def getDBKeysFromMetadata(flags) -> Optional[dict[str, Any]]:
    """Provides access to the database keys from the in-file metadata
 
    Gets the database keys from the in-file metadata which are stored together with the json representation
    If the keys are in the file, then usually SMK, L1PSK and HLTPSK are present.
    The bunchgroupset is not stored in the metadata, so 0 is returned for the BGS key for completeness.
 
    @returns: dictionary with the DB keys. Returns 'None' if information is not present.
    """
    metadata = GetFileMD(flags.Input.Files)
    keys = metadata.get("TriggerConfigInfo", None)
    if keys is None:
        return None
    return {
        'SMK': keys['HLT']['key'] if 'HLT' in keys else 0,
        'L1PSK': keys['L1PS']['key'] if 'L1PS' in keys else 0,
        'HLTPSK': keys['HLTPS']['key'] if 'HLTPSK' in keys else 0,
        'BGSK': 0
    }
 
"""
Returns a string-serialised JSON object from the metadata store.
Checks AOD syntax first, then fully-qualified ESD syntax
"""

getHLTJobOptionsAccess()

HLTJobOptionsAccess python.TriggerConfigAccess.getHLTJobOptionsAccess

(

flags = None

)

Definition at line 236 of file TriggerConfigAccess.py.

def getHLTJobOptionsAccess( flags = None ) -> HLTJobOptionsAccess:
    tc = getTrigConfigFromFlag( flags )
    if tc["SOURCE"] == "FILE":
        cfg = HLTJobOptionsAccess( filename = getHLTJobOptionsFileName() )
    elif tc["SOURCE"] == "COOL":
        """This is the case when reconstructing the data."""
        if len(flags.Input.RunNumbers) == 0:
            raise RuntimeError("No run number available in input metadata")
        keysFromCool = getKeysFromCool( flags.Input.RunNumbers[0] )
        cfg = HLTJobOptionsAccess( dbalias = keysFromCool["DB"], smkey = keysFromCool['SMK'] )
    elif tc["SOURCE"] == "DB":
        cfg = HLTJobOptionsAccess( dbalias = tc["DBCONN"], smkey = tc["SMK"] )
    elif tc["SOURCE"] == "INFILE":
        raise NotImplementedError("Python access to the HLT Job Options configuration from in-file metadata is NOT SUPPORTED (this file is huge!)")
    else:
        raise RuntimeError("Unknown source of trigger configuration: %s" % tc["SOURCE"])
    return cfg
 
 
@AccumulatorCache

getHLTMenuAccess()

HLTMenuAccess python.TriggerConfigAccess.getHLTMenuAccess

(

flags = None

)

Definition at line 196 of file TriggerConfigAccess.py.

def getHLTMenuAccess( flags = None ) -> HLTMenuAccess:
    tc = getTrigConfigFromFlag( flags )
    if tc["SOURCE"] == "FILE":
        cfg = HLTMenuAccess( filename = getHLTMenuFileName( flags ) )
    elif tc["SOURCE"] == "COOL":
        """This is the case when reconstructing the data."""
        if len(flags.Input.RunNumbers) == 0:
            raise RuntimeError("No run number available in input metadata")
        keysFromCool = getKeysFromCool( flags.Input.RunNumbers[0] )
        cfg = HLTMenuAccess( dbalias = keysFromCool["DB"], smkey = keysFromCool['SMK'] )
    elif tc["SOURCE"] == "DB":
        cfg = HLTMenuAccess( dbalias = tc["DBCONN"], smkey = tc["SMK"] )
    elif tc["SOURCE"] == "INFILE":
        cfg = HLTMenuAccess(jsonString=_getJSONFromMetadata(flags, key='TriggerMenuJson_HLT'))
    else:
        raise RuntimeError("Unknown source of trigger configuration: %s" % tc["SOURCE"])
    return cfg
 
 
@AccumulatorCache

getHLTMonitoringAccess()

HLTMonitoringAccess python.TriggerConfigAccess.getHLTMonitoringAccess

(

flags = None

)

Definition at line 256 of file TriggerConfigAccess.py.

def getHLTMonitoringAccess( flags = None ) -> HLTMonitoringAccess:
    tc = getTrigConfigFromFlag( flags )
    if tc["SOURCE"] == "FILE":
        cfg = HLTMonitoringAccess( filename = getHLTMonitoringFileName( flags ) )
    elif tc["SOURCE"] == "COOL":
        """This is the case when reconstructing the data."""
        if len(flags.Input.RunNumbers) == 0:
            raise RuntimeError("No run number available in input metadata")
        keysFromCool = getKeysFromCool( flags.Input.RunNumbers[0] )
        cfg = HLTMonitoringAccess( dbalias = keysFromCool["DB"], smkey = keysFromCool['SMK'] )
    elif tc["SOURCE"] == "DB":
        cfg = HLTMonitoringAccess( dbalias = tc["DBCONN"], smkey = tc["SMK"] )
    elif tc["SOURCE"] == "INFILE":
        jsonHLTMon = _getJSONFromMetadata(flags, key='TriggerMenuJson_HLTMonitoring')
        if jsonHLTMon is not None:
            cfg = HLTMonitoringAccess(jsonString=jsonHLTMon)
        else:
            keysFromInfileMD = getDBKeysFromMetadata(flags)
            smkey = keysFromInfileMD['SMK'] if keysFromInfileMD is not None else 0
            if smkey < 3000 and not flags.Input.isMC:
                # Run 1/2 data or keys missing in metadata 
                log.info("Trigger metadata with key 'TriggerMenuJson_HLTMonitoring' is not available for Run 2 data. Returning empty dummy.")
                jsonHLTMon = '{"filetype": "hltmonitoringsummary","name": "EmptyDefault", "signatures": {}}'
                cfg = HLTMonitoringAccess(jsonString=jsonHLTMon)
            else:
                # Run 3 data or MC
                try:
                    log.info("Falling back on reading the HLTMonitoring from the TRIGGERDB_RUN3 for SMK %i.", smkey)
                    cfg = HLTMonitoringAccess( dbalias = "TRIGGERDB_RUN3", smkey = smkey )
                except KeyError:
                    # SMK for this run has no HLT monitoring (earlier 2022 data) => providing dummy configuration
                    log.info("Trigger HLTMonitoring is not available for SMK %i. Returning empty dummy.", smkey)
                    jsonHLTMon = '{"filetype": "hltmonitoringsummary","name": "EmptyDefault", "signatures": {}}'
                    cfg = HLTMonitoringAccess(jsonString=jsonHLTMon)
 
    else:
        raise RuntimeError("Unknown source of trigger configuration: %s" % tc["SOURCE"])
    return cfg

getHLTPrescalesSetAccess()

HLTPrescalesSetAccess python.TriggerConfigAccess.getHLTPrescalesSetAccess

(

flags = None

)

Definition at line 216 of file TriggerConfigAccess.py.

def getHLTPrescalesSetAccess( flags = None ) -> HLTPrescalesSetAccess:
    tc = getTrigConfigFromFlag( flags )
    if tc["SOURCE"] == "FILE":
        cfg = HLTPrescalesSetAccess( filename = getHLTPrescalesSetFileName( flags ) )
    elif tc["SOURCE"] == "COOL":
        """This is the case when reconstructing the data."""
        if len(flags.Input.RunNumbers) == 0:
            raise RuntimeError("No run number available in input metadata")
        keysFromCool = getKeysFromCool( flags.Input.RunNumbers[0] )
        cfg = HLTPrescalesSetAccess( dbalias = keysFromCool["DB"], hltpskey = keysFromCool['HLTPSK'] )
    elif tc["SOURCE"] == "DB":
        cfg = HLTPrescalesSetAccess( dbalias = tc["DBCONN"], hltpskey = tc["HLTPSK"] )
    elif tc["SOURCE"] == "INFILE":
        cfg = HLTPrescalesSetAccess(jsonString=_getJSONFromMetadata(flags, key='TriggerMenuJson_HLTPS'))
    else:
        raise RuntimeError("Unknown source of trigger configuration: %s" % tc["SOURCE"])
    return cfg
 
 
@AccumulatorCache

getKeysFromCool()

dict[str,int] python.TriggerConfigAccess.getKeysFromCool	(	int	runNr,
		int	lbNr = 0
	)

Return dictionary of trigger keys for given run and lumiblock number

Definition at line 52 of file TriggerConfigAccess.py.

def getKeysFromCool(runNr: int, lbNr: int = 0) -> dict[str,int]:
    """Return dictionary of trigger keys for given run and lumiblock number
    """
    from TrigConfStorage.TriggerCoolUtil import TriggerCoolUtil
    condb = "CONDBR2" if runNr > 236108 else "COMP200"
    db = TriggerCoolUtil.GetConnection(condb)
    run_range = [[runNr,runNr]]
    d = { k: TriggerCoolUtil.getHLTConfigKeys(db, run_range)[runNr][k] for k in ['SMK', 'DB'] }
    for ( key, lbfirst, lblast) in TriggerCoolUtil.getBunchGroupKey(db, run_range)[runNr]['BGKey']:
        if lbNr>=lbfirst and (lbNr<=lblast or lblast==-1):
            d['BGSK'] = key
            break
    for ( key, lbfirst, lblast) in TriggerCoolUtil.getL1ConfigKeys(db, run_range)[runNr]['LVL1PSK']:
        if lbNr>=lbfirst and (lbNr<=lblast or lblast==-1):
            d['L1PSK'] = key
            break
    for ( key, lbfirst, lblast) in TriggerCoolUtil.getHLTPrescaleKeys(db, run_range)[runNr]['HLTPSK2']:
        if lbNr>=lbfirst and (lbNr<=lblast or lblast==-1):
            d['HLTPSK'] = key
            break
 
    # dbalias mapping
    dbaliasMapping = { "TRIGGERDBR2R" : "TRIGGERDB",
                       "TRIGGERDBV2" : "TRIGGERDB_RUN1" }
    if d["DB"] in dbaliasMapping:
        d["DB"] = dbaliasMapping[ d["DB"] ]
 
    return d
 

getL1MenuAccess()

L1MenuAccess python.TriggerConfigAccess.getL1MenuAccess

(

flags = None

)

Definition at line 129 of file TriggerConfigAccess.py.

def getL1MenuAccess( flags = None ) -> L1MenuAccess:
    tc = getTrigConfigFromFlag( flags )
    if tc["SOURCE"] == "FILE":
        cfg = L1MenuAccess( filename = getL1MenuFileName( flags ) )
    elif tc["SOURCE"] == "COOL":
        """This is the case when reconstructing the data."""
        if len(flags.Input.RunNumbers) == 0:
            raise RuntimeError("No run number available in input metadata")
        keysFromCool = getKeysFromCool( flags.Input.RunNumbers[0] )
        cfg = L1MenuAccess( dbalias = keysFromCool["DB"], smkey = keysFromCool['SMK'] )
    elif tc["SOURCE"] == "DB":
        cfg = L1MenuAccess( dbalias = tc["DBCONN"], smkey = tc["SMK"] )
    elif tc["SOURCE"] == "INFILE":
        cfg = L1MenuAccess(jsonString=_getJSONFromMetadata(flags, key='TriggerMenuJson_L1'))
    else:
        raise RuntimeError("Unknown source of trigger configuration: %s" % tc["SOURCE"])
    return cfg
 
 
@AccumulatorCache

getL1PrescalesSetAccess()

L1PrescalesSetAccess python.TriggerConfigAccess.getL1PrescalesSetAccess

(

flags = None

)

Definition at line 149 of file TriggerConfigAccess.py.

def getL1PrescalesSetAccess( flags = None ) -> L1PrescalesSetAccess:
    tc = getTrigConfigFromFlag( flags )
    if tc["SOURCE"] == "FILE":
        cfg = L1PrescalesSetAccess( filename = getL1PrescalesSetFileName( flags ) )
    elif tc["SOURCE"] == "COOL":
        """This is the case when reconstructing the data."""
        if len(flags.Input.RunNumbers) == 0:
            raise RuntimeError("No run number available in input metadata")
        keysFromCool = getKeysFromCool( flags.Input.RunNumbers[0] )
        cfg = L1PrescalesSetAccess( dbalias = keysFromCool["DB"], l1pskey = keysFromCool['L1PSK'] )
    elif tc["SOURCE"] == "DB":
        cfg = L1PrescalesSetAccess( dbalias = tc["DBCONN"], l1pskey = tc["L1PSK"] )
    elif tc["SOURCE"] == "INFILE":
        cfg = L1PrescalesSetAccess(jsonString=_getJSONFromMetadata(flags, key='TriggerMenuJson_L1PS'))
    else:
        raise RuntimeError("Unknown source of trigger configuration: %s" % tc["SOURCE"])
    return cfg
 
 
@AccumulatorCache

Variable Documentation

log

python.TriggerConfigAccess.log

Definition at line 6 of file TriggerConfigAccess.py.

maxsize

python.TriggerConfigAccess.maxsize

Definition at line 51 of file TriggerConfigAccess.py.