python.TriggerCrestUtil.TriggerCrestUtil Class Reference

Collaboration diagram for python.TriggerCrestUtil.TriggerCrestUtil:

Static Public Member Functions
str|None	getCrestConnection (str oracle_db)
CrestApi	getCrestApi (str server)
dict[str, dict[str, Any]]	getConditionsForTimestamp (str tag, *int timestamp, str server="", CrestApi|None api=None, bool get_time_type=False)
list[dict[str, dict[str, Any]]]	getConditionsInRange (str tag, *int since, int until, str server="", CrestApi|None api=None, bool get_time_type=False)
str	getTagTimeType (str tag, *str server="", CrestApi|None api=None)
list[str]	getAttribs (str tag, *str server="", CrestApi|None api=None)
dict	getPayloadFromHash (str payload_hash, *str server="", CrestApi|None api=None)
dict[str, Any]|None	getEORParams (int run, *str server="", CrestApi|None api=None)
list[dict[str, dict[str, Any]]]	getHLTPrescaleKeys (int run, *str server="", CrestApi|None api=None)
list[dict[str, dict[str, Any]]]	getL1ConfigKeys (int run, *str server="", CrestApi|None api=None)
list[dict[str, dict[str, Any]]]	getBunchGroupKey (int run, *str server="", CrestApi|None api=None)
dict	getHLTConfigKeys (int run, *str server="", CrestApi|None api=None)

Static Private Member Functions
None	_update_with_time_type (dict[str, Any] result, str time_type, int since)
dict	_get_payload (payload_hash, api)
dict	_get_payload_workaround (payload_hash, server)
def	_get_iovs_for_run (str tag, *int run, CrestApi api)
IovSetDto|HTTPResponse	_get_iov_for_timestamp (str tag, *int timestamp, CrestApi api)
IovSetDto|HTTPResponse	_get_iovs_range (*int since, int until, CrestApi api, str tag)
tuple[list[Any], dict[Any, Any]]	_get_payload_spec (str tag, CrestApi api)

Detailed Description

Definition at line 13 of file TriggerCrestUtil.py.

Member Function Documentation

_get_iov_for_timestamp()

IovSetDto | HTTPResponse python.TriggerCrestUtil.TriggerCrestUtil._get_iov_for_timestamp ( str  tag, *int  timestamp, CrestApi  api  )

staticprivate

Helper to retrieve the IOV for a given timestamp

Args:
    tag (str): tag name
    timestamp (int): time-stamp in format run<32+lb or time in nanoseconds since 1.1.1970
    api (CrestApi): Crest API instance.

Returns:
    IovSetDto: set of IOVs (of size 1)

Definition at line 369 of file TriggerCrestUtil.py.

    def _get_iov_for_timestamp(tag: str, *, timestamp: int, api: CrestApi) -> IovSetDto | HTTPResponse:
        """Helper to retrieve the IOV for a given timestamp
 
        Args:
            tag (str): tag name
            timestamp (int): time-stamp in format run<32+lb or time in nanoseconds since 1.1.1970
            api (CrestApi): Crest API instance.
 
        Returns:
            IovSetDto: set of IOVs (of size 1)
        """
        # the upper limit of the iov-search is exclusive, so we need to add 1
        return api.select_iovs(tag, "0", str(timestamp+1), sort='id.since:DESC', size=1, snapshot=0)  # type: ignore
 

_get_iovs_for_run()

def python.TriggerCrestUtil.TriggerCrestUtil._get_iovs_for_run ( str  tag, *int  run, CrestApi  api  )

staticprivate

Helper to retrieve all IOVs for a run

Definition at line 362 of file TriggerCrestUtil.py.

    def _get_iovs_for_run(tag: str, *, run: int, api: CrestApi):
        """Helper to retrieve all IOVs for a run"""
        run_start = (run << 32) + 1
        run_end = ((run + 1) << 32) - 1
        return TriggerCrestUtil._get_iovs_range(since=run_start, until=run_end, api=api, tag=tag)
 

_get_iovs_range()

IovSetDto | HTTPResponse python.TriggerCrestUtil.TriggerCrestUtil._get_iovs_range ( *int  since, int  until, CrestApi  api, str  tag  )

staticprivate

Helper to retrieve the IOV in a given range

Args:
    tag (str): tag name
    since (int): start of the range (either run<32+lb or time in nanoseconds since 1.1.1970, inclusive)
    until (int): end of the range (either run<32+lb or time in nanoseconds since 1.1.1970, exclusive)
    api (CrestApi): Crest API instance.

Returns:
    IovSetDto: set of IOVs

Definition at line 384 of file TriggerCrestUtil.py.

    def _get_iovs_range(*, since: int, until: int, api: CrestApi, tag: str) -> IovSetDto | HTTPResponse:
        """Helper to retrieve the IOV in a given range
 
        Args:
            tag (str): tag name
            since (int): start of the range (either run<32+lb or time in nanoseconds since 1.1.1970, inclusive)
            until (int): end of the range (either run<32+lb or time in nanoseconds since 1.1.1970, exclusive)
            api (CrestApi): Crest API instance.
 
        Returns:
            IovSetDto: set of IOVs
        """
        # the upper limit of the iov-search is exclusive, so we need to add 1 to find the iov which includes 'since'
        iovs = api.select_iovs(tag, "0", str(since+1), sort='id.since:DESC', size=1, snapshot=0) # type: ignore
        if cast(int, iovs['size']) < 1:
            raise RuntimeError(f"Did not get an iov which includes the start of run {since}")
        firstiov = cast(list, iovs['resources'])[0]
        all_iovs = api.select_iovs(tag, str(firstiov['since']), str(until), sort='id.since:ASC', snapshot=0) # type: ignore
        return all_iovs
 

_get_payload()

dict python.TriggerCrestUtil.TriggerCrestUtil._get_payload (   payload_hash,   api  )

staticprivate

Definition at line 337 of file TriggerCrestUtil.py.

    def _get_payload(payload_hash, api) -> dict:
        return api.get_payload(hash=payload_hash).decode('utf-8')
 

_get_payload_spec()

tuple[list[Any], dict[Any, Any]] python.TriggerCrestUtil.TriggerCrestUtil._get_payload_spec ( str  tag, CrestApi  api  )

staticprivate

Helper to retrieve the payload spec for a given tag

Definition at line 405 of file TriggerCrestUtil.py.

    def _get_payload_spec(tag: str, api: CrestApi) -> tuple[list[Any], dict[Any, Any]]:
        """Helper to retrieve the payload spec for a given tag"""
        meta: TagMetaSetDto | HTTPResponse = api.find_tag_meta(tag)
        tag_info = cast(list, meta['resources'])[0]['tag_info']
        tag_info_dict = json.loads(tag_info)
        attr_list = []
        type_dict = {}
        for d in tag_info_dict['payload_spec']:
            attr_list += list(d)
            type_dict.update(d)
        return attr_list, type_dict
 
 

_get_payload_workaround()

dict python.TriggerCrestUtil.TriggerCrestUtil._get_payload_workaround (   payload_hash,   server  )

staticprivate

Definition at line 341 of file TriggerCrestUtil.py.

    def _get_payload_workaround(payload_hash, server) -> dict:
        import requests
        import json
        url = f"{server}/payloads/data"
        params = {
            "format": "BLOB", 
            "hash": payload_hash
        }
        preq = requests.Request(method='GET', url=url, params=params).prepare()
        with requests.Session() as session:
            try:
                resp = session.send(preq)
            except requests.ConnectionError as exc:
                raise RuntimeError(f"Could not connect to CREST server {server}") from exc
 
            if resp.status_code != 200:
                raise RuntimeError(f"Query {payload_hash} to crest failed with status code {resp.status_code}")
 
        return json.loads(resp.content)
 

_update_with_time_type()

None python.TriggerCrestUtil.TriggerCrestUtil._update_with_time_type ( dict[str, Any]  result, str  time_type, int  since  )

staticprivate

Definition at line 323 of file TriggerCrestUtil.py.

    def _update_with_time_type(result: dict[str, Any], time_type: str, since: int) -> None:
        result['time_type'] = time_type
        if time_type == 'run-lumi':
            result.update({
                'since_run': since >> 32,
                'since_lb': since & 0xFFFFFFFF
            })
        elif time_type == 'time':
            result.update({
                'since_formatted': dt.fromtimestamp(since / 1e9)
            })
 
 

getAttribs()

list[str] python.TriggerCrestUtil.TriggerCrestUtil.getAttribs ( str  tag, *str   server = "", CrestApi | None   api = None  )

static

list of attributes for this tag

Args:
    tag (str): tag name
    server (str, optional): crest server name. Only needed if api is not provided. Defaults to "".
    api (CrestApi, optional): Crest API instance. Defaults to None.

Raises:
    RuntimeError: if both server and api are missing

Returns:
    dict: list of attributes

Definition at line 188 of file TriggerCrestUtil.py.

    def getAttribs(tag: str, *, server: str = "", api: CrestApi | None = None) -> list[str]:
        """ list of attributes for this tag
 
        Args:
            tag (str): tag name
            server (str, optional): crest server name. Only needed if api is not provided. Defaults to "".
            api (CrestApi, optional): Crest API instance. Defaults to None.
 
        Raises:
            RuntimeError: if both server and api are missing
 
        Returns:
            dict: list of attributes
        """
        if server=="" and api is None:
            log.error("Need either crest_server name or crest api for retrieving the payload")
            raise RuntimeError("Crest access information missing")
        if api is None:
            api = CrestApi(host=server)
        attr_list, _ = TriggerCrestUtil._get_payload_spec(tag, api)
        return attr_list
 

getBunchGroupKey()

list[dict[str, dict[str, Any]]] python.TriggerCrestUtil.TriggerCrestUtil.getBunchGroupKey ( int  run, *str   server = "", CrestApi | None   api = None  )

static

Definition at line 275 of file TriggerCrestUtil.py.

    def getBunchGroupKey(run: int, *, server: str = "", api: CrestApi | None = None) -> list[dict[str, dict[str, Any]]]:
        if api is None:
            api = CrestApi(host=server)
        run_start = (run << 32)
        run_end = ((run+1) << 32)-1
        cond = TriggerCrestUtil.getConditionsInRange("TRIGGERLVL1BunchGroupKey-HEAD", since=run_start, until=run_end, api=api, get_time_type=True)
        for entry in cond:
            entry.update(entry.pop('payload')['0'])
            entry['key'] = entry.pop('Lvl1BunchGroupConfigurationKey')
        return cond
 

getConditionsForTimestamp()

dict[str, dict[str, Any]] python.TriggerCrestUtil.TriggerCrestUtil.getConditionsForTimestamp ( str  tag, *int  timestamp, str   server = "", CrestApi | None   api = None, bool   get_time_type = False  )

static

Conditions data for tag at a specific timestamp

Conditions data is returned as a dict with 'since' and 'payload' keys.
The 'payload' itself is a dict of channel:'data dicts'. The 'data dict' has attribute names as
keys and the corresponding values. If get_time_type is True, also 'time_type' key is added to 
each IOV dict and the since is further expanded into 'since_run' and 'since_lb' (for run/lb-based 
timestamps) or 'since_formatted' (for time-based timestamps).

Args:
    tag (str): The tag name
    timestamp (int): The timestamp to query (either run<32+lb or time in nanoseconds since 1.1.1970)
    server (str, optional): Crest server name. Only needed if api is not provided. Defaults to "".
    api (_type_, optional): Crest API instance. Defaults to None.
    get_time_type (bool, optional): Whether to retrieve the time type. Defaults to False.

Raises:
    RuntimeError: If both server and api are missing

Returns:
    dict[str, dict[str, any]]: Conditions data

Definition at line 59 of file TriggerCrestUtil.py.

    def getConditionsForTimestamp(tag: str, *, timestamp: int, server: str = "", api: CrestApi | None = None,
                                  get_time_type: bool = False) -> dict[str, dict[str, Any]]:
        """ Conditions data for tag at a specific timestamp
 
        Conditions data is returned as a dict with 'since' and 'payload' keys.
        The 'payload' itself is a dict of channel:'data dicts'. The 'data dict' has attribute names as
        keys and the corresponding values. If get_time_type is True, also 'time_type' key is added to 
        each IOV dict and the since is further expanded into 'since_run' and 'since_lb' (for run/lb-based 
        timestamps) or 'since_formatted' (for time-based timestamps).
 
        Args:
            tag (str): The tag name
            timestamp (int): The timestamp to query (either run<32+lb or time in nanoseconds since 1.1.1970)
            server (str, optional): Crest server name. Only needed if api is not provided. Defaults to "".
            api (_type_, optional): Crest API instance. Defaults to None.
            get_time_type (bool, optional): Whether to retrieve the time type. Defaults to False.
 
        Raises:
            RuntimeError: If both server and api are missing
 
        Returns:
            dict[str, dict[str, any]]: Conditions data
        """
        if server=="" and api is None:
            log.error("Need either crest server name or crest api specified for accessing Crest")
            raise RuntimeError("Crest access information missing")
        if api is None:
            api = CrestApi(host=server)
        # get the payload specification
        attr_list, _ = TriggerCrestUtil._get_payload_spec(tag, api)
        # get the IOVs in the given range
        iov = TriggerCrestUtil._get_iov_for_timestamp(tag, timestamp=timestamp, api=api)
 
        if get_time_type:
            time_type = TriggerCrestUtil.getTagTimeType(tag, api=api)
 
        payload_for_iov = {}
        payload_hash: str = cast(list, iov['resources'])[0]['payload_hash']
        since: int = cast(list, iov['resources'])[0]['since']
        payload = TriggerCrestUtil.getPayloadFromHash(payload_hash, api=api)
        for channel, data in payload.items():
            payload_for_iov[channel] = dict(zip(attr_list, data))
 
        result: dict[str, Any] = {
            'since': since,
            'payload': payload_for_iov
        }
        if get_time_type:
            TriggerCrestUtil._update_with_time_type(result, time_type, since)
        return result
 
 

getConditionsInRange()

list[dict[str, dict[str, Any]]] python.TriggerCrestUtil.TriggerCrestUtil.getConditionsInRange ( str  tag, *int  since, int  until, str   server = "", CrestApi | None   api = None, bool   get_time_type = False  )

static

Conditions data for tag in given range

List goes over the IOVs found. For each IOV, there is a dict with 'since' and 'payload' keys. 
The 'payload' itself is a dict of channel:'data dicts'. The 'data dict' has attribute names as
keys and the corresponding values. If get_time_type is True, also 'time_type' key is added to 
each IOV dict and the since is further expanded into 'since_run' and 'since_lb' (for run/lb-based 
timestamps) or 'since_formatted' (for time-based timestamps).

Args:
    tag (str): tag name
    since (int): start of the range (either run<32+lb or time in nanoseconds since 1.1.1970, inclusive)
    until (int): end of the range (either run<32+lb or time in nanoseconds since 1.1.1970, exclusive)
    server (str, optional): crest server name. Only needed if api is not provided. Defaults to "".
    api (CrestApi, optional): Crest API instance. Defaults to None.
    get_time_type (bool, optional): whether to retrieve the time type. Defaults to False.

Raises:
    RuntimeError: if both server and api are missing

Returns:
    list[dict[str, dict[str, any]]]: conditions data

Definition at line 112 of file TriggerCrestUtil.py.

    def getConditionsInRange(tag: str, *, since: int, until: int, server: str = "", api: CrestApi | None = None,
                             get_time_type: bool = False) -> list[dict[str, dict[str, Any]]]:
        """ Conditions data for tag in given range
 
        List goes over the IOVs found. For each IOV, there is a dict with 'since' and 'payload' keys. 
        The 'payload' itself is a dict of channel:'data dicts'. The 'data dict' has attribute names as
        keys and the corresponding values. If get_time_type is True, also 'time_type' key is added to 
        each IOV dict and the since is further expanded into 'since_run' and 'since_lb' (for run/lb-based 
        timestamps) or 'since_formatted' (for time-based timestamps).
 
        Args:
            tag (str): tag name
            since (int): start of the range (either run<32+lb or time in nanoseconds since 1.1.1970, inclusive)
            until (int): end of the range (either run<32+lb or time in nanoseconds since 1.1.1970, exclusive)
            server (str, optional): crest server name. Only needed if api is not provided. Defaults to "".
            api (CrestApi, optional): Crest API instance. Defaults to None.
            get_time_type (bool, optional): whether to retrieve the time type. Defaults to False.
 
        Raises:
            RuntimeError: if both server and api are missing
 
        Returns:
            list[dict[str, dict[str, any]]]: conditions data
        """
        if server=="" and api is None:
            log.error("Need either crest server name or crest api specified for accessing Crest")
            raise RuntimeError("Crest access information missing")
        if api is None:
            api = CrestApi(host=server)
        # get the payload specification
        attr_list, type_dict = TriggerCrestUtil._get_payload_spec(tag, api)
        # get the IOVs in the given range
        all_iovs = TriggerCrestUtil._get_iovs_range(since=since, until=until, api=api, tag=tag)
 
        if get_time_type:
            time_type = TriggerCrestUtil.getTagTimeType(tag, api=api)
 
        result: list[dict[str, dict[str, Any]]] = []
        for iov in cast(Iterable, all_iovs['resources']):
            payload_for_iov = {}
            payload_hash = iov['payload_hash']
            since = iov['since']
            payload = TriggerCrestUtil.getPayloadFromHash(payload_hash, api=api)
            for channel, data in payload.items():
                payload_for_iov[channel] = dict(zip(attr_list, data))
            entry = {'since': since, 'payload': payload_for_iov}
            if get_time_type:
                TriggerCrestUtil._update_with_time_type(entry, time_type, since)
            result += [entry]
        return result
 

getCrestApi()

CrestApi python.TriggerCrestUtil.TriggerCrestUtil.getCrestApi ( str  server )

static

Crest API object

Args:
    server (str): crest server name.

Returns:
    CrestApi: Crest API instance.

Definition at line 47 of file TriggerCrestUtil.py.

    def getCrestApi(server: str) -> CrestApi:
        """ Crest API object
 
        Args:
            server (str): crest server name.
 
        Returns:
            CrestApi: Crest API instance.
        """
        return CrestApi(host=server)
 

getCrestConnection()

str | None python.TriggerCrestUtil.TriggerCrestUtil.getCrestConnection ( str  oracle_db )

static

maps from triggerdb or triggerdb-alias to crest connection

See https://its.cern.ch/jira/browse/ATR-32030

Definition at line 16 of file TriggerCrestUtil.py.

    def getCrestConnection(oracle_db: str) -> str | None:
        """maps from triggerdb or triggerdb-alias to crest connection
        
        See https://its.cern.ch/jira/browse/ATR-32030
        """
        db_mapping: dict[str, str] = {
            "ATLAS_CONF_TRIGGER_RUN3": "CONF_DATA_RUN3",
            "ATLAS_CONF_TRIGGER_MC_RUN3": "CONF_MC_RUN3",
            "ATLAS_CONF_TRIGGER_REPR_RUN3": "CONF_REPR_RUN3",
            "ATLAS_CONF_TRIGGER_RUN4": "CONF_DATA_RUN4",
            "ATLAS_CONF_TRIGGER_MC_RUN4": "CONF_MC_RUN4",
            "ATLAS_CONF_TRIGGER_REPR_RUN4": "CONF_REPR_RUN4",
            "ATLAS_CONF_TRIGGER_LS3_DEV": "CONF_DEV_LS3",
            "ATLAS_CONF_TRIGGER_LS3_L0": "CONF_DEV_L0",
            "ATLAS_CONF_TRIGGER_RUN2_NF": "CONF_DATA_RUN2"
        }
        alias_mapping: dict[str, str] = {
            "TRIGGERDB_RUN3": "CONF_DATA_RUN3",
            "TRIGGERDBREPR_RUN3": "CONF_REPR_RUN3",
            "TRIGGERDBMC_RUN3": "CONF_MC_RUN3",
            "TRIGGERDB_RUN4": "CONF_DATA_RUN4",
            "TRIGGERDBREPR_RUN4": "CONF_REPR_RUN4",
            "TRIGGERDBMC_RUN4": "CONF_MC_RUN4",
            "TRIGGERDBLS3_DEV": "CONF_DEV_LS3",
            "TRIGGERDBLS3_L0": "CONF_DEV_L0",
            "TRIGGERDB_RUN2_NF": "CONF_DATA_RUN2_NF"
        }
        db_mapping.update(alias_mapping)
        return db_mapping.get(oracle_db, None)
    

getEORParams()

dict[str, Any] | None python.TriggerCrestUtil.TriggerCrestUtil.getEORParams ( int  run, *str   server = "", CrestApi | None   api = None  )

static

Definition at line 240 of file TriggerCrestUtil.py.

    def getEORParams(run: int, *, server: str = "", api: CrestApi | None = None) -> dict[str, Any] | None:
        if api is None:
            api = CrestApi(host=server)
        start_of_run = (run << 32)
        cond_data = TriggerCrestUtil.getConditionsForTimestamp("TDAQRunCtrlEOR-HEAD", timestamp=start_of_run, api=api, get_time_type=False)
        if not cond_data:
            log.error("No EOR params found for run %s", run)
            return None
        return cond_data['payload']['0']
 

getHLTConfigKeys()

dict python.TriggerCrestUtil.TriggerCrestUtil.getHLTConfigKeys ( int  run, *str   server = "", CrestApi | None   api = None  )

static

Definition at line 287 of file TriggerCrestUtil.py.

    def getHLTConfigKeys(run: int, *, server: str = "", api: CrestApi | None = None) -> dict:
        # helper function to turn the payload of the TRIGGERHLTHltConfigKeys into a dictionary
        def _parse_info(run, cond):
            # Format for ConfigSource (see ATR-21550):
            #   Run-4: currently like Run-3
            #   Run-3: TRIGGERDB_RUN3;22.0.101;Athena,22.0.101 --extra_args ...
            #   Run-2: TRIGGERDBR2R,21.1.24,AthenaP1
            release = 'unknown'
            if run > 379000:  # Run-3
                confsrc = cond['ConfigSource'].split(';')
                dbalias = confsrc[0]
                if len(confsrc) > 2:
                    release = confsrc[2].split(',')[0] + f",{confsrc[1]}"
            else:  # Run-2
                confsrc = cond['ConfigSource'].split(',', maxsplit=1)
                dbalias = confsrc[0]
                if len(confsrc) > 2:
                    release = f"{confsrc[2]},{confsrc[1]}"
            return {
                "SMK": cond['MasterConfigurationKey'],
                "HLTPSK": cond['HltPrescaleConfigurationKey'],
                "DB": dbalias,
                "REL": release
            }
 
        if api is None:
            api = CrestApi(host=server)
 
        run_start = (run << 32) + 1
        cond: dict[str, dict[str, Any]] = TriggerCrestUtil.getConditionsForTimestamp("TRIGGERHLTHltConfigKeys-HEAD", timestamp=run_start, api=api, get_time_type=True)
        cond.update(cond.pop('payload')['0'])
        return _parse_info(run, cond)
 
 

getHLTPrescaleKeys()

list[dict[str, dict[str, Any]]] python.TriggerCrestUtil.TriggerCrestUtil.getHLTPrescaleKeys ( int  run, *str   server = "", CrestApi | None   api = None  )

static

Definition at line 251 of file TriggerCrestUtil.py.

    def getHLTPrescaleKeys(run: int, *, server: str = "", api: CrestApi | None = None) -> list[dict[str, dict[str, Any]]]:  # -> list[dict[str, dict[str, Any]]]:
        if api is None:
            api = CrestApi(host=server)
        run_start = (run << 32)
        run_end = ((run + 1) << 32) - 1
        cond = TriggerCrestUtil.getConditionsInRange("TRIGGERHLTPrescaleKey-HEAD", since=run_start, until=run_end, api=api, get_time_type=True)
        for entry in cond:
            entry.update(entry.pop('payload')['0'])
            entry['key'] = entry.pop('HltPrescaleKey')
        return cond
 

getL1ConfigKeys()

list[dict[str, dict[str, Any]]] python.TriggerCrestUtil.TriggerCrestUtil.getL1ConfigKeys ( int  run, *str   server = "", CrestApi | None   api = None  )

static

Definition at line 263 of file TriggerCrestUtil.py.

    def getL1ConfigKeys(run: int, *, server: str = "", api: CrestApi | None = None) -> list[dict[str, dict[str, Any]]]:
        if api is None:
            api = CrestApi(host=server)
        run_start = (run << 32)
        run_end = ((run + 1) << 32) - 1
        cond = TriggerCrestUtil.getConditionsInRange("TRIGGERLVL1Lvl1ConfigKey-HEAD", since=run_start, until=run_end, api=api, get_time_type=True)
        for entry in cond:
            entry.update(entry.pop('payload')['0'])
            entry['key'] = entry.pop('Lvl1PrescaleConfigurationKey')
        return cond
 

getPayloadFromHash()

dict python.TriggerCrestUtil.TriggerCrestUtil.getPayloadFromHash ( str  payload_hash, *str   server = "", CrestApi | None   api = None  )

static

Get payload from hash

Args:
    payload_hash (str): hash of the payload
    server (str, optional): crest server name. Defaults to "".
    api (CrestApi, optional): Crest API instance. Defaults to None.

Raises:
    RuntimeError: if both server and api are missing

Returns:
    dict: payload retrieved from the hash

Definition at line 211 of file TriggerCrestUtil.py.

    def getPayloadFromHash(payload_hash: str, *, server: str = "", api: CrestApi | None = None) -> dict:
        """ Get payload from hash
 
        Args:
            payload_hash (str): hash of the payload
            server (str, optional): crest server name. Defaults to "".
            api (CrestApi, optional): Crest API instance. Defaults to None.
 
        Raises:
            RuntimeError: if both server and api are missing
 
        Returns:
            dict: payload retrieved from the hash
        """
        if server == "" and api is None:
            log.error("Need either crest_server name or crest api for retrieving the payload")
            raise RuntimeError("Crest access information missing")
        useWorkAround = True
        if useWorkAround:
            if server == "":
                server = api._host  # type: ignore
            return TriggerCrestUtil._get_payload_workaround(payload_hash, server)
        else:
            if api is None:
                api = CrestApi(host=server)
            return TriggerCrestUtil._get_payload(payload_hash, api)
 

getTagTimeType()

str python.TriggerCrestUtil.TriggerCrestUtil.getTagTimeType ( str  tag, *str   server = "", CrestApi | None   api = None  )

static

time type of the tag, either 'run-lumi' or 'time'

Args:
    tag (str): tag name
    server (str, optional): crest server name. Only needed if api is not provided. Defaults to "".
    api (CrestApi, optional): Crest API instance. Defaults to None.

Raises:
    RuntimeError: if both server and api are missing

Returns:
    str: time type of the tag ['run-lumi' or 'time']

Definition at line 164 of file TriggerCrestUtil.py.

    def getTagTimeType(tag: str, *, server: str = "", api: CrestApi | None = None) -> str:
        """ time type of the tag, either 'run-lumi' or 'time'
 
        Args:
            tag (str): tag name
            server (str, optional): crest server name. Only needed if api is not provided. Defaults to "".
            api (CrestApi, optional): Crest API instance. Defaults to None.
 
        Raises:
            RuntimeError: if both server and api are missing
 
        Returns:
            str: time type of the tag ['run-lumi' or 'time']
        """
        if server=="" and api is None:
            log.error("Need either crest_server name or crest api for retrieving the payload")
            raise RuntimeError("Crest access information missing")
        if api is None:
            api = CrestApi(host=server)
        tag_info = api.find_tag(tag)
        time_type: str = tag_info['time_type']
        return time_type
 

---

The documentation for this class was generated from the following file:

• TriggerCrestUtil.py