python.TrigP1TestSteps Namespace Reference

Classes
class	ExtractExpertMonitoring
class	TrigBSDumpGrepStep

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

Function Documentation

check_hlt_properties()

def python.TrigP1TestSteps.check_hlt_properties

(

filename = 'HLTJobOptions.db.json'

)

Check a few important job options in the JSON file

Definition at line 108 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()

def python.TrigP1TestSteps.decodeBS	(		stream_name,
			moduleID = 0
	)

Deserialise HLT data from ByteStream and save to an ESD file

Definition at line 165 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('*'+stream_name+'*._athenaHLT*.data')
    return decodeStep

default_check_steps_OHMon()

def python.TrigP1TestSteps.default_check_steps_OHMon	(		test,
			hist_path
	)

Definition at line 135 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()

def python.TrigP1TestSteps.filterBS

(

stream_name

)

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

Definition at line 154 of file TrigP1TestSteps.py.

def filterBS(stream_name):
    '''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 = '-s ' + stream_name + ' ' + find_file('*_HLTMPPy_output.*.data')
    return filterStep