python.TriggerConfigAccess Namespace Reference

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

Variables
	log = logging.getLogger( "TriggerConfigAccess.py" )

Function Documentation

_getJSONFromMetadata()

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

protected

Definition at line 112 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

)

Definition at line 176 of file TriggerConfigAccess.py.

def getBunchGroupSetAccess( flags ) -> 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")
        keysFromConditions = getKeysFromConditions( flags.Input.RunNumbers[0], 0, flags )
        cfg = BunchGroupSetAccess( dbalias = keysFromConditions["DB"], bgskey = keysFromConditions['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.

Example of the 'TriggerConfigInfo' metadata entry (run 453713, lb 1416):
{
    'HLT': {'key': 3221, 'name': 'PhysicsP1_pp_run3_v1'},
    'HLTPS': {'key': 6247, 'name': 'PhysicsP1_pp_run3_v1'},
    'L1': {'key': 3221, 'name': 'Physics_pp_run3_v1'},
    'L1PS': {'key': 7475, 'name': 'Physics_pp_run3_v1'}
}
@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.
 
    Example of the 'TriggerConfigInfo' metadata entry (run 453713, lb 1416):
    {
        'HLT': {'key': 3221, 'name': 'PhysicsP1_pp_run3_v1'},
        'HLTPS': {'key': 6247, 'name': 'PhysicsP1_pp_run3_v1'},
        'L1': {'key': 3221, 'name': 'Physics_pp_run3_v1'},
        'L1PS': {'key': 7475, 'name': 'Physics_pp_run3_v1'}
    }
    @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,
        'LVL1PSK': keys['L1PS']['key'] if 'L1PS' in keys else 0,
        'HLTPSK': keys['HLTPS']['key'] if 'HLTPS' 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

)

Definition at line 243 of file TriggerConfigAccess.py.

def getHLTJobOptionsAccess( flags ) -> 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")
        keysFromConditions = getKeysFromConditions( flags.Input.RunNumbers[0], 0, flags )
        cfg = HLTJobOptionsAccess( dbalias = keysFromConditions["DB"], smkey = keysFromConditions['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

)

Definition at line 203 of file TriggerConfigAccess.py.

def getHLTMenuAccess( flags ) -> 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")
        keysFromConditions = getKeysFromConditions( flags.Input.RunNumbers[0], 0, flags )
        cfg = HLTMenuAccess( dbalias = keysFromConditions["DB"], smkey = keysFromConditions['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,
			filterOnActiveChains = False )

Definition at line 263 of file TriggerConfigAccess.py.

def getHLTMonitoringAccess( flags = None, filterOnActiveChains = False ) -> 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")
        keysFromConditions = getKeysFromConditions( flags.Input.RunNumbers[0], 0, flags )
        cfg = HLTMonitoringAccess( dbalias = keysFromConditions["DB"], smkey = keysFromConditions['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 flags.Trigger.EDMVersion < 3 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:
                    if smkey < 3000:
                        log.info("Falling back on reading the HLTMonitoring from the TRIGGERDBREPR_RUN3 for SMK %i.", smkey)
                        cfg = HLTMonitoringAccess( dbalias = "TRIGGERDBREPR_RUN3", smkey = smkey )
                    else:
                        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"])
 
    # Optional deletion of chains from the monitoring JSON which were not enabled,
    # based on the first HLT PS set in the file or the HLT PS set for the RAW file's LB.
    if filterOnActiveChains:
        hltPs = getHLTPrescalesSetAccess(flags)
        if hltPs:
            for sig in cfg["signatures"]:
                for chain in list(cfg["signatures"][sig]): # Make a copy as we might be deleting
                    if chain in hltPs.chainNames() and not hltPs.enabled(chain):
                        del cfg["signatures"][sig][chain]
 
    return cfg

getHLTPrescalesSetAccess()

HLTPrescalesSetAccess python.TriggerConfigAccess.getHLTPrescalesSetAccess

(

flags

)

Definition at line 223 of file TriggerConfigAccess.py.

def getHLTPrescalesSetAccess( flags ) -> 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")
        keysFromConditions = getKeysFromConditions( flags.Input.RunNumbers[0], 0, flags )
        cfg = HLTPrescalesSetAccess( dbalias = keysFromConditions["DB"], hltpskey = keysFromConditions['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

getKeysFromConditions()

dict[str, str | int] python.TriggerConfigAccess.getKeysFromConditions	(	int	runNr,
		int	lbNr,
			flags )

Definition at line 51 of file TriggerConfigAccess.py.

def getKeysFromConditions(runNr: int, lbNr: int, flags) -> dict[str, str | int]:
    if flags.Trigger.useCrest:
        confKeys: dict[str, str | int] = getKeysFromCrest(runNr, lbNr, flags.Trigger.crestServer)
    else:
        confKeys = getKeysFromCool(runNr, lbNr)
 
    # dbalias mapping
    dbaliasMapping = {
        "TRIGGERDBR2R" : "TRIGGERDB",
        "TRIGGERDBV2" : "TRIGGERDB_RUN1"
    }
    if confKeys["DB"] in dbaliasMapping:
        confKeys["DB"] = dbaliasMapping[ confKeys["DB"] ]
 
    return confKeys
 
@lru_cache(maxsize=None)

getKeysFromCool()

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

Return dictionary of trigger keys for given run and lumiblock number

Definition at line 75 of file TriggerConfigAccess.py.

def getKeysFromCool(runNr: int, lbNr: int = 0) -> dict[str, str | int]:
    """Return dictionary of trigger keys for given run and lumiblock number
    """
    from TrigConfStorage.TriggerCoolUtil import TriggerCoolUtil
    return TriggerCoolUtil.getTrigConfKeys(runNumber=runNr, lumiBlock=lbNr)
 

getKeysFromCrest()

dict[str, str | int] python.TriggerConfigAccess.getKeysFromCrest	(	int	runNr,
		int	lbNr,
		str	crest_server )

Return dictionary of trigger keys for given run and lumiblock number from crest

Definition at line 68 of file TriggerConfigAccess.py.

def getKeysFromCrest(runNr: int, lbNr: int, crest_server: str) -> dict[str, str | int]:
    """Return dictionary of trigger keys for given run and lumiblock number from crest
    """
    from TrigConfStorage.TriggerCrestUtil import TriggerCrestUtil
    return TriggerCrestUtil.getTrigConfKeys(runNumber=runNr, lumiBlock=lbNr, server=crest_server)
 
@lru_cache(maxsize=None)

getL1MenuAccess()

L1MenuAccess python.TriggerConfigAccess.getL1MenuAccess

(

flags

)

Definition at line 136 of file TriggerConfigAccess.py.

def getL1MenuAccess( flags ) -> 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")
        keysFromConditions = getKeysFromConditions( flags.Input.RunNumbers[0], 0, flags )
        cfg = L1MenuAccess( dbalias = keysFromConditions["DB"], smkey = keysFromConditions['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

)

Definition at line 156 of file TriggerConfigAccess.py.

def getL1PrescalesSetAccess( flags ) -> 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")
        keysFromConditions = getKeysFromConditions( flags.Input.RunNumbers[0], 0, flags )
        cfg = L1PrescalesSetAccess( dbalias = keysFromConditions["DB"], l1pskey = keysFromConditions['LVL1PSK'] )
    elif tc["SOURCE"] == "DB":
        cfg = L1PrescalesSetAccess( dbalias = tc["DBCONN"], l1pskey = tc["LVL1PSK"] )
    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 = logging.getLogger( "TriggerConfigAccess.py" )

Definition at line 6 of file TriggerConfigAccess.py.