python.TrigP1TestSteps Namespace Reference

Classes
class	TrigBSDumpGrepStep
class	ExtractExpertMonitoring

Functions
	check_hlt_properties (filename='HLTJobOptions.db.json')
	default_check_steps_OHMon (test, hist_path)
	filterBS (stream_name, extra_args='')
	decodeBS (stream_name, moduleID=0)

Detailed Description

Definitions of additional validation steps in Trigger ART tests relevant only for TrigP1Test.
The main common check steps are defined in the TrigValSteering.CheckSteps module.

Function Documentation

check_hlt_properties()

check_hlt_properties

(

filename = 'HLTJobOptions.db.json'

)

Check a few important job options in the JSON file

Definition at line 109 of file TrigP1TestSteps.py.

def check_hlt_properties(filename='HLTJobOptions.db.json'):
    '''Check a few important job options in the JSON file'''
 
    log = get_logger()
 
    with open(filename) as f:
        props = json.load(f)['properties']
 
    def checkprop(comp, prop, regex):
        value = props[comp][prop]
        if re.match(regex, value) is None:
            log.error('Property "%s.%s" does not match "%s". Current value: "%s"', comp, prop, regex, value)
            return False
        return True
 
    checks = [
        checkprop('IOVDbSvc', 'CacheAlign', '0'),
        checkprop('IOVDbSvc', 'CacheRun', '0'),
        checkprop('IOVDbSvc', 'CacheTime', '0'),
        checkprop('AtlasFieldMapCondAlg', 'LoadMapOnStart', 'True'),
        checkprop('IOVDbSvc', 'Folders', r'.*/TRIGGER/LUMI/HLTPrefLumi\s*<extensible/>.*'),
        checkprop('IOVDbSvc', 'Folders', r'.*/Indet/Onl/Beampos\s*<extensible/>.*'),
        checkprop('IOVDbSvc', 'Folders', r'.*/TRIGGER/HLT/PrescaleKey <tag>HEAD</tag>\s*<extensible/>.*'),
    ]
    return 0 if all(checks) else 1
 
 

decodeBS()

decodeBS	(		stream_name,
			moduleID = 0 )

Deserialise HLT data from ByteStream and save to an ESD file

Definition at line 166 of file TrigP1TestSteps.py.

def decodeBS(stream_name, moduleID=0):
    '''Deserialise HLT data from ByteStream and save to an ESD file'''
    from TrigValTools.TrigValSteering import ExecStep
    from TrigValTools.TrigValSteering.Common import find_file
    decodeStep = ExecStep.ExecStep('DecodeBS_'+stream_name)
    decodeStep.type = 'other'
    decodeStep.executable = 'python'
    decodeStep.input = ''
    decodeStep.explicit_input = True
    decodeStep.args = f'-m TrigP1Test.DecodeBS --moduleID={moduleID} --filesInput=' + find_file(f'data*_{stream_name}.*.data')
    return decodeStep

default_check_steps_OHMon()

default_check_steps_OHMon	(		test,
			hist_path )

Definition at line 136 of file TrigP1TestSteps.py.

def default_check_steps_OHMon(test, hist_path):
    steps = []
    # Extract expert-monitoring.root file from OH server output
    extract_hist = ExtractExpertMonitoring()
    hist_path_split = hist_path.split(':')
    if len(hist_path_split) > 1:
        extract_hist.input_file = hist_path_split[0]
        extract_hist.path_prefix = hist_path_split[1]
    else:
        extract_hist.input_file = hist_path
    steps.append(extract_hist)
    # Default check steps
    steps.extend(CheckSteps.default_check_steps(test))
    # Remove histogram merging step
    matches = [step for step in steps if step.name == 'HistMerge']
    for hm_step in matches:
        steps.remove(hm_step)
    return steps
 

filterBS()

filterBS	(		stream_name,
			extra_args = '' )

Extract ByteStream data for a given stream from a file with multiple streams

Definition at line 155 of file TrigP1TestSteps.py.

def filterBS(stream_name, extra_args = ''):
    '''Extract ByteStream data for a given stream from a file with multiple streams'''
    from TrigValTools.TrigValSteering import ExecStep
    from TrigValTools.TrigValSteering.Common import find_file
    filterStep = ExecStep.ExecStep('FilterBS_'+stream_name)
    filterStep.type = 'other'
    filterStep.executable = 'trigbs_extractStream.py'
    filterStep.input = ''
    filterStep.args = f'{extra_args} -s {stream_name} ' + find_file('data*_SingleStream.daq.RAW.*_output.*.data', max_files=None)
    return filterStep