trigbs_dumpHLTContentInBS_run3.py

Go to the documentation of this file.

#!/usr/bin/env python
#
# Copyright (C) 2002-2024 CERN for the benefit of the ATLAS collaboration
#
 
'''
Dump content of the HLT result and HLT related details from the event header
'''
 
import argparse
import eformat
from datetime import datetime
from TrigByteStreamTools import hltResultMT
 
from AthenaCommon.Logging import logging
log = logging.getLogger('dumpHLTContentInBS')
 
 
def get_parser():
    parser = argparse.ArgumentParser(usage='%(prog)s [options] files',
                                     description=__doc__)
    parser.add_argument('files',
                        metavar='FILE', nargs='+',
                        help='RAW file to inspect')
    parser.add_argument('-n', '--events',
                        metavar='N', action='store', type=int,
                        help='Process N events')
    parser.add_argument('-s', '--skip',
                        metavar='N', action='store', type=int,
                        help='Skip N events')
    parser.add_argument('--ctp', nargs="?", metavar="MODULE_ID", default=False, const=1,
                        help="CTP ROB details of ROB with MODULE_ID [default=%(const)s]")
    parser.add_argument('--l1',
                        action='store_true', default=False,
                        help='L1 trigger bits (from event header)')
    parser.add_argument('--ef', '--hlt',
                        action='store_true', default=False,
                        help='EF/HLT trigger bits (from event header)')
    parser.add_argument('--stag',
                        action='store_true', default=False,
                        help='stream tag')
    parser.add_argument('--efres', '--hltres',
                        action='store_true', default=False,
                        help='details of EF/HLT ROB payload')
    parser.add_argument('--sizes',
                        action='store_true', default=False,
                        help='dump info about EDM sizes per result; implies --hltres')
    parser.add_argument('--deserialize',
                        action='store_true', default=False,
                        help='deserialize EDM collections (slow!); implies --hltres')
    parser.add_argument('--sizeSummary',
                        action='store_true', default=False,
                        help='dump summary info about EDM sizes at the end')
    parser.add_argument('--confKeys',
                        action='store_true', default=False,
                        help='dump TrigConfKeys stored in the events; implies --hltres')
    parser.add_argument('--runtimeMetadata',
                        action='store_true', default=False,
                        help='dump HLT_RuntimeMetadata stored in the events; implies --hltres')
    return parser
 
 
def decodeTriggerBits(words, num_sets, base=32):
    assert len(words) % num_sets == 0
    n_words_per_set = len(words) // num_sets
    result = []
    for iset in range(num_sets):
        words_in_set = words[iset*n_words_per_set:(iset+1)*n_words_per_set]
        bit_indices = []
        for iw in range(len(words_in_set)):
            bit_indices.extend([base*iw+i for i in range(base) if words_in_set[iw] & (1 << i)])
        result.append(bit_indices)
    return result
 
 
def header_info(event):
    '''Return a string with basic information about the event from the header'''
 
    info_str = 'RunNumber: {:6d}, '.format(event.run_no())
    info_str += 'LB: {:4d}, '.format(event.lumi_block())
    info_str += 'Global_ID: {:10d}, '.format(event.global_id())
    info_str += 'LVL1_ID: {:10d}, '.format(event.lvl1_id())
    info_str += 'BC_ID: {:4d}, '.format(event.bc_id())
    info_str += 'Time: {}, '.format(datetime.fromtimestamp(event.bc_time_seconds()).isoformat())
    info_str += 'TT: 0x{:2x}, '.format(event.lvl1_trigger_type())
    info_str += 'Status: ' + decode_status(event)
    return info_str
 
 
def decode_status(fragment):
    '''Return event/ROB fragment status and string representation'''
 
    if not fragment.status():
        return 'None'
 
    info_str = '[' + ', '.join('0x%x' % s for s in fragment.status()) + ']'
 
    # First word:
    status = eformat.helper.Status(fragment.status()[0])
    info_str += ''.join(f' {item.name}' for code, item in eformat.helper.GenericStatus.values.items()
                        if status.generic() & code)
 
    # The specific bits of the first word are only relevant for the full event:
    if isinstance(fragment, eformat.FullEventFragment):
        info_str += ''.join(f' {item.name}' for code, item in eformat.helper.FullEventStatus.values.items()
                            if status.specific() & code)
 
    # HLT-specific event status:
    if len(fragment.status())>1:
        import cppyy
        cppyy.gbl.HLT.OnlineErrorCode  # trigger auto-loading via enum
        info_str += ' '+ cppyy.gbl.HLT.OnlineErrorCodeToString(fragment.status()[1]).data()
 
    return info_str
 
 
def ctp_rob(event, module_id=1):
    '''Return a string with information from the CTP ROB'''
 
    ctp_robs = [rob for rob in event \
                if rob.source_id().subdetector_id() == eformat.helper.SubDetector.TDAQ_CTP \
                and rob.source_id().module_id()==module_id]
 
    if len(ctp_robs)==0:
        log.warning('No CTP ROB found')
        return ''
 
    assert len(ctp_robs)==1
    rob = ctp_robs[0]
 
    from TrigByteStreamTools import CTPfragment
    x = CTPfragment.getExtraPayloadObject(rob)
    info_str = (f'ROB 0x{rob.source_id().code():x}, '
                f'L1ID {event.lvl1_id():10d}, '
                f'LB {event.lumi_block():4d}, '
                f'Version {CTPfragment.ctpFormatVersion(rob)}, '
                f'Bunch {CTPfragment.lvl1AcceptBunch(rob)}, '
                f'HLT counter: {CTPfragment.hltCounter(rob):3d}, '
                f'Payload #{CTPfragment.numberHltExtraPayloadWords(rob)} '
                f'{CTPfragment.hltExtraPayloadWords(rob)} '
                f'L1PSK {x.getL1PSK()} BGK {x.getBGK()}')
    folderUpdates = CTPfragment.getFolderUpdates(x)
    if len(folderUpdates)>0:
        info_str += ' COOLUPD '+''.join([f'[{f.second.folderIndex}, {f.second.lumiBlock}]' for f in folderUpdates])
    return info_str
 
 
def lvl1_bits(event):
    '''Return a string with information about LVL1 bits (IDs of items passed at TBP, TAP, TAV)'''
 
    info = event.lvl1_trigger_info()
    lvl1_bits = decodeTriggerBits(info, 3)  # TBP, TAP, TAV
    info_str = 'L1 CTP IDs - TBP: {:s}\n'.format(str(lvl1_bits[0]))
    info_str += 'L1 CTP IDs - TAP: {:s}\n'.format(str(lvl1_bits[1]))
    info_str += 'L1 CTP IDs - TAV: {:s}'.format(str(lvl1_bits[2]))
    return info_str
 
 
def hlt_bits(event, version=(1,1)):
    '''Return a string with information about passed chain IDs at HLT'''
 
    # Version 1.0 has {passed, prescaled, rerun}, 1.1 and later only {passed, prescaled}
    num_sets = 3 if version[0] < 1 or version==(1,0) else 2
 
    info = event.event_filter_info()
    hlt_bits = decodeTriggerBits(info, num_sets)
    info_str = 'HLT passed chain IDs: {:s}'.format(str(hlt_bits[0]))
    info_str += '\nHLT prescaled chain IDs: {:s}'.format(str(hlt_bits[1]))
    if num_sets==3:
        info_str += '\nHLT rerun chain IDs: {:s}'.format(str(hlt_bits[2]))
    return info_str
 
 
def stream_tags(event):
    def hex_list(nums):
        return '[' + ', '.join([hex(num) for num in nums]) + ']'
    info_str = 'Stream Tags:'
    for st in event.stream_tag():
        info_str += '\n-- {}_{}'.format(st.type, st.name)
        robs = list(st.robs)
        dets = list(st.dets)
        if len(robs) == 0 and len(dets) == 0:
            info_str += ' - Full Event Building'
        else:
            info_str += ' - Partial Event Building, robs={}, dets={}'.format(hex_list(robs), hex_list(dets))
    return info_str
 
 
def hlt_rod_minor_version(rob):
    version = rob.rod_version()
    minor_version = version.minor_version()
    minor_version_M = (minor_version & 0xff00) >> 8
    minor_version_L = minor_version & 0xff
    return (minor_version_M, minor_version_L)
 
 
def hlt_rod_minor_version_from_event(event):
    for rob in event.children():
        if rob.source_id().subdetector_id() == eformat.helper.SubDetector.TDAQ_HLT:
            return hlt_rod_minor_version(rob)
 
 
def hlt_result(event, print_sizes=False, deserialize=False, conf_keys=False, runtime_metadata=False):
    num_hlt_robs = 0
    info_str = ""
    for rob in event.children():
        if rob.source_id().subdetector_id() != eformat.helper.SubDetector.TDAQ_HLT:
            continue
        num_hlt_robs += 1
        version = hlt_rod_minor_version(rob)
        info_str += '\n-- {:s} SourceID: {:s}, Version: {:s}, Size: {:d} bytes, Status: {:s}'.format(
            rob.__class__.__name__,
            rob.source_id().human(),
            f'{version[0]:d}.{version[1]:d}',
            rob.fragment_size_word()*4,
            decode_status(rob)
        )
        if print_sizes or deserialize or conf_keys or runtime_metadata:
            if version[0] < 1:
                raise RuntimeError('Cannot decode data from before Run 3, HLT ROD minor version needs to be >= 1.0')
            skip_payload = not conf_keys and not runtime_metadata and not deserialize
            collections = hltResultMT.get_collections(rob, skip_payload=skip_payload)
            if conf_keys:
                conf_list = [c for c in collections if 'xAOD::TrigConfKeys_v' in c.name_persistent]
                conf_available = False
                for conf in conf_list:
                    conf_obj = conf.deserialise()
                    if not conf_obj:
                        continue
                    conf_available = True
                    info_str += '\n---- {:s}#{:s} SMK: {:d}, L1PSK: {:d}, HLTPSK: {:d}'.format(
                        conf.name_persistent, conf.name_key,
                        conf_obj.smk(), conf_obj.l1psk(), conf_obj.hltpsk())
                if not conf_available:
                    info_str += '\n---- TrigConfKeys unavailable in this ROB'
            if runtime_metadata:
                decorations = ['hostname']
                meta_list = [c for c in collections if any([f'RuntimeMetadataAux.{decor}' in c.name() for decor in decorations])]
                meta_available = False
                for meta in meta_list:
                    meta_obj = meta.deserialise()
                    if not meta_obj:
                        continue
                    meta_available = True
                    info_str += '\n---- RuntimeMetadata {:s}: {:s}'.format(
                        meta.name_key, meta_obj.at(0))
                if not meta_available:
                    info_str += '\n---- RuntimeMetadata unavailable in this ROB'
            if print_sizes or deserialize:
                for coll in collections:
                    indent = 4 if not coll.is_xAOD_decoration() else 6
                    info_str += '\n{:s} {:s}'.format('-'*indent, str(coll))
                    if deserialize and (coll_obj := coll.deserialise()) is not None:
                        try:
                            length = coll_obj.size()  # all collections should have this method
                        except Exception:
                            length = None
                        if length is not None:
                            info_str += f' ({length} element' + ('s)' if length!=1 else ')')
                        info_str += '\n{:s} {:s}'.format(' '*indent, str(coll_obj))
 
 
    info_str = 'Found {:d} HLT ROBs'.format(num_hlt_robs) + info_str
    return info_str
 
 
def size_summary(events):
    data = {}
    # Format of the data dictionary:
    # {
    #   0: {
    #     'total': 123,
    #     'collections': {
    #       'collA': 22,
    #       'collB': 35
    #     }
    #   }
    # }
    for event in events:
        for rob in event.children():
            if rob.source_id().subdetector_id() != eformat.helper.SubDetector.TDAQ_HLT:
                continue
            version = hlt_rod_minor_version(rob)
            if version[0] < 1:
                raise RuntimeError('Cannot decode data from before Run 3, HLT ROD minor version needs to be >= 1.0')
            module = rob.source_id().module_id()
            if module not in data.keys():
                data[module] = {'total': 0}
            data[module]['total'] += rob.fragment_size_word()*4
            if 'collections' not in data[module].keys():
                data[module]['collections'] = {}
            for coll in hltResultMT.get_collections(rob, skip_payload=True):
                coll_name = coll.name()
                if coll_name in data[module]['collections'].keys():
                    data[module]['collections'][coll_name] += coll.size_bytes
                else:
                    data[module]['collections'][coll_name] = coll.size_bytes
    info_str = '='*20 + '\nSize summary:\n' + '='*20
    for module in data.keys():
        module_size = data[module]['total']
        module_size_per_evt = module_size / float(len(events))
        info_str += '\n-- TDAQ_HLT module {:d} total size {:d} bytes, {:.3f} bytes/event'.format(
            module, module_size, module_size_per_evt)
        sorted_colls = sorted(data[module]['collections'].items(),
                              key=lambda kv: kv[1],
                              reverse=True)
        max_name_len = len(max(data[module]['collections'].keys(), key=len))
        max_name_len = min(120, max_name_len)  # Make a reasonable limit to avoid line breaks
        for coll_name, coll_size in sorted_colls:
            coll_size_per_evt = coll_size / float(len(events))
            info_str += '\n---- {:{width}} {:12d} B {:12.3f} B/ev'.format(
                coll_name, coll_size, coll_size_per_evt, width=max_name_len)
    return info_str
 
def dump_info(bsfile, args):
    log.info('Opening %s', bsfile)
    input = eformat.istream(bsfile)
    offset = args.skip if args.skip else 0
    max_events = min(args.events, len(input)) if args.events else len(input)
    event_count = 0
    events = []
    version = None
 
    # Loop over events
    for event in input:
        event_count += 1
        if event_count <= offset:
            continue
        if event_count > offset+max_events:
            break
        events.append(event)
 
        if version is None:
            version = hlt_rod_minor_version_from_event(event)
 
        # Print header info
        print('{sep:s} Event: {:{width}d}, {:s} {sep:s}'.format(
              event_count, header_info(event),
              sep='===', width=len(str(max_events))))
 
        # Print CTP ROB
        if args.ctp is not False:
            print(ctp_rob(event, 1))
 
        # Print L1/L2/HLT bits
        if args.l1:
            print(lvl1_bits(event))
        if args.ef and version is not None:
            print(hlt_bits(event, version))
 
        # Print Stream Tags
        if args.stag:
            print(stream_tags(event))
 
        # HLT Result
        if args.efres or args.sizes or args.deserialize or args.confKeys or args.runtimeMetadata:
            print(hlt_result(event, args.sizes, args.deserialize, args.confKeys, args.runtimeMetadata))
 
    # Size summary (after the loop over events)
    if args.sizeSummary:
        print(size_summary(events))
 
 
if '__main__' in __name__:
    args = get_parser().parse_args()
    for f in args.files:
        dump_info(f, args)