Skip to content

Update data_file_check_utilities to use FragmentStatusBits #142

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
eflumerf merged 1 commit into from
Feb 24, 2026
Merged

Update data_file_check_utilities to use FragmentStatusBits #142

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 20 additions & 20 deletions src/integrationtest/data_file_check_utilities.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@

import daqdataformats
import trgdataformats
from daqdataformats import FragmentErrorBits
from daqdataformats import FragmentStatusBits
from hdf5libs import HDF5RawDataFile

def get_TC_types(h5_file, record_id) -> list:
Expand Down Expand Up @@ -152,25 +152,25 @@ def get_fragment_size_limits(params, tc_type_string, record_ordinal_strings):

return [min_size, max_size]

def get_fragment_error_bitmask(params, tc_type_string, record_ordinal_strings):
def get_fragment_status_bitmask(params, tc_type_string, record_ordinal_strings):
# Set default value (all bits allowed)
error_bitmask = 0xFFFFFFFF
status_bitmask = 0xFFFFFFFF

# get first-level defaults from any top-level parameters specified by the user
if 'error_bitmask' in params.keys():
error_bitmask = params['error_bitmask']
if 'status_bitmask' in params.keys():
status_bitmask = params['status_bitmask']

# check for bitmasks that are specified by TC type
if 'frag_bitmasks_by_TC_type' in params.keys():
tc_type_dict = params['frag_bitmasks_by_TC_type']
if tc_type_string in tc_type_dict.keys():
bitmask_dict = tc_type_dict[tc_type_string]
if 'error_bitmask' in bitmask_dict.keys():
error_bitmask = bitmask_dict['error_bitmask']
if 'status_bitmask' in bitmask_dict.keys():
status_bitmask = bitmask_dict['status_bitmask']
elif 'default' in tc_type_dict.keys():
bitmask_dict = tc_type_dict['default']
if 'error_bitmask' in bitmask_dict.keys():
error_bitmask = bitmask_dict['error_bitmask']
if 'status_bitmask' in bitmask_dict.keys():
status_bitmask = bitmask_dict['status_bitmask']

# check for bitmasks that are specified by record number
# (obviously, if both bitmasks-by-TC-type and bitmasks-by-record-ordinal are
Expand All @@ -180,23 +180,23 @@ def get_fragment_error_bitmask(params, tc_type_string, record_ordinal_strings):
record_number_dict = params['frag_bitmasks_by_record_ordinal']
if rno_string in record_number_dict.keys():
bitmask_dict = record_number_dict[rno_string]
if 'error_bitmask' in bitmask_dict.keys():
error_bitmask = bitmask_dict['error_bitmask']
if 'status_bitmask' in bitmask_dict.keys():
status_bitmask = bitmask_dict['status_bitmask']
elif 'default' in record_number_dict.keys():
bitmask_dict = record_number_dict['default']
if 'error_bitmask' in bitmask_dict.keys():
error_bitmask = bitmask_dict['error_bitmask']
if 'status_bitmask' in bitmask_dict.keys():
status_bitmask = bitmask_dict['status_bitmask']

return error_bitmask
return status_bitmask

def get_set_error_bit_names(error_bits):
error_bits_temp = error_bits
def get_set_status_bit_names(status_bits):
status_bits_temp = status_bits
names = []
current_bit = 0
while error_bits_temp != 0:
if error_bits_temp & 0x1 == 0x1:
names.append(FragmentErrorBits(current_bit).name)
error_bits_temp = error_bits_temp >> 1
while status_bits_temp != 0:
if status_bits_temp & 0x1 == 0x1:
names.append(FragmentStatusBits(current_bit).name)
status_bits_temp = status_bits_temp >> 1
current_bit = current_bit + 1
return names

Expand Down
24 changes: 12 additions & 12 deletions src/integrationtest/data_file_checks.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,8 +9,8 @@
get_record_ordinal_strings,
get_fragment_count_limits,
get_fragment_size_limits,
get_fragment_error_bitmask,
get_set_error_bit_names,
get_fragment_status_bitmask,
get_set_status_bit_names,
sid_key,
get_TR_trigger_types,
unpack_TR_trigger_types,
Expand Down Expand Up @@ -264,12 +264,12 @@ def check_fragment_sizes(datafile, params):
# is that each type of fragment can be tested individually, by calling this routine for
# each type. The test is driven by a set of parameters that describe both the fragments
# to be tested (e.g. the Fragment type) and the characteristics that they should have
# (e.g. any allowed error bits).
# (e.g. any allowed status bits).
#
# The parameters that are required by this routine are the following:
# * fragment_type_description - descriptive text for the fragment type, e.g. "WIB" or "PDS" or "Raw TP"
# * fragment_type - Type of the Fragment, e.g. "ProtoWIB" or "Trigger_Primitive"
# * error_bitmask - A mask to be applied to the error bits of the Fragment (default: 0xFFFFFFFF)
# * status_bitmask - A mask to be applied to the status bits of the Fragment (default: 0xFFFFFFFF)
def check_fragment_error_flags(datafile, params):
if params['expected_fragment_count'] == 0:
return True
Expand All @@ -288,23 +288,23 @@ def check_fragment_error_flags(datafile, params):
for rec in records:
trigger_type_string = get_trigger_type_string(h5_file, rec)
rno_strings = get_record_ordinal_strings(rec, records)
error_bitmask = get_fragment_error_bitmask(params, trigger_type_string, rno_strings)
status_bitmask = get_fragment_status_bitmask(params, trigger_type_string, rno_strings)
if (debug_mask & 0x4) != 0:
print(f'DataFileChecks Debug: the fragment error bitmask is {hex(error_bitmask)} for TC type {trigger_type_string} and record ordinal strings {rno_strings}')
if error_bitmask not in error_mask_list:
error_mask_list.append(error_bitmask)
print(f'DataFileChecks Debug: the fragment status bitmask is {hex(status_bitmask)} for TC type {trigger_type_string} and record ordinal strings {rno_strings}')
if status_bitmask not in error_mask_list:
error_mask_list.append(status_bitmask)
if subdet_string == "":
src_ids = h5_file.get_source_ids_for_fragment_type(rec, params['fragment_type'])
else:
src_ids = h5_file.get_source_ids_for_fragtype_and_subdetector(rec, params['fragment_type'], subdet_string)
for src_id in src_ids:
frag=h5_file.get_frag(rec,src_id);
error_bits=frag.get_error_bits()
status_bits=frag.get_status_bits()
if (debug_mask & 0x8) != 0:
print(f' DataFileChecks Debug: fragment error bits for SourceID {src_id} are {hex(error_bits)}')
if error_bits & error_bitmask != 0:
print(f' DataFileChecks Debug: fragment status bits for SourceID {src_id} are {hex(status_bits)}')
if status_bits & status_bitmask != 0:
passed=False
print(f" \N{POLICE CARS REVOLVING LIGHT} {params['fragment_type_description']} fragment for SrcID {src_id.to_string()} in record {rec} has the following unmasked error flags set: {get_set_error_bit_names(error_bits & error_bitmask)} \N{POLICE CARS REVOLVING LIGHT}")
print(f" \N{POLICE CARS REVOLVING LIGHT} {params['fragment_type_description']} fragment for SrcID {src_id.to_string()} in record {rec} has the following unmasked error flags set: {get_set_status_bit_names(status_bits & status_bitmask)} \N{POLICE CARS REVOLVING LIGHT}")
if passed:
error_mask_list.sort()
print(f"\N{WHITE HEAVY CHECK MARK} All {params['fragment_type_description']} fragments in {len(records)} records have no error flags set (after applying bitmasks)")
Expand Down