python.TrigValSteering.CheckSteps Namespace Reference

Classes
class	ChainCompStep
class	ChainDumpStep
class	CheckFileStep
class	CheckLogStep
class	DownloadRefStep
class	HistCountStep
class	InputDependentStep
class	LogMergeStep
class	MessageCountStep
class	PerfMonStep
class	RefComparisonStep
class	RegTestStep
class	RootCompStep
class	RootMergeStep
class	TailStep
class	TrigTestJsonStep
class	ZeroCountsStep
class	ZipStep

Functions
def	produces_log (step)
def	default_check_steps (test)
def	add_step_after_type (step_list, ref_type, step_to_add)

Function Documentation

add_step_after_type()

def python.TrigValSteering.CheckSteps.add_step_after_type	(		step_list,
			ref_type,
			step_to_add
	)

Insert step_to_add into step_list after the last step of type ref_type.
If the list has no steps of type ref_type, append step_to_add at the end of the list.

Definition at line 834 of file CheckSteps.py.

def add_step_after_type(step_list, ref_type, step_to_add):
    '''
    Insert step_to_add into step_list after the last step of type ref_type.
    If the list has no steps of type ref_type, append step_to_add at the end of the list.
    '''
    index_to_add = -1
    for index, step in enumerate(step_list):
        if isinstance(step, ref_type):
            index_to_add = index+1
    if index_to_add > 0:
        step_list.insert(index_to_add, step_to_add)
    else:
        step_list.append(step_to_add)

default_check_steps()

def python.TrigValSteering.CheckSteps.default_check_steps

(

test

)

Create the default list of check steps for a test. The configuration
depends on the package name and the type of exec steps (athena or
athenaHLT or transforms).

Definition at line 715 of file CheckSteps.py.

def default_check_steps(test):
    '''
    Create the default list of check steps for a test. The configuration
    depends on the package name and the type of exec steps (athena or
    athenaHLT or transforms).
    '''
 
    check_steps = []
 
    # Log merging
    if len(test.exec_steps) == 1:
        exec_step = test.exec_steps[0]
        if exec_step.type == 'athenaHLT' and produces_log(exec_step):
            logmerge = LogMergeStep()
            logmerge.merged_name = 'athena.log'
            logmerge.log_files = ['athenaHLT.log']
            nforks = 1 if exec_step.forks is None else exec_step.forks
            for n in range(1, 1+nforks):
                logmerge.log_files.append('athenaHLT:{:02d}.out'.format(n))
                logmerge.log_files.append('athenaHLT:{:02d}.err'.format(n))
            check_steps.append(logmerge)
    else:
        logmerge = LogMergeStep()
        logmerge.merged_name = 'athena.log'
        logmerge.log_files = []
        for exec_step in test.exec_steps:
            if not produces_log(exec_step):
                continue
            logmerge.log_files.append(exec_step.get_log_file_name())
            if exec_step.type == 'athenaHLT':
                logmerge.extra_log_regex = 'athenaHLT:.*(.out|.err)'
        check_steps.append(logmerge)
    log_to_check = None
    log_to_zip = None
    if len(check_steps) > 0 and isinstance(check_steps[-1], LogMergeStep):
        log_to_check = check_steps[-1].merged_name
        log_to_zip = check_steps[-1].merged_name
 
    # Reco_tf log merging
    reco_tf_steps = [step for step in test.exec_steps if step.type in ['Reco_tf', 'Trig_reco_tf', 'Derivation_tf']]
    if len(reco_tf_steps) > 0:
        reco_tf_logmerge = LogMergeStep('LogMerge_Reco_tf')
        reco_tf_logmerge.warn_if_missing = False
        # FIXME: drop AODtoDAOD once test_trigAna_AODtoDAOD_run2_build.py is migrated to Derivation_tf
        tf_names = ['HITtoRDO', 'Overlay', 'RDOtoRDOTrigger', 'RAWtoESD', 'ESDtoAOD',
                    'PhysicsValidation', 'RAWtoALL',
                    'BSRDOtoRAW', 'DRAWCOSTtoNTUPCOST', 'AODtoNTUPRATE', 'Derivation', 'AODtoDAOD']
        reco_tf_logmerge.log_files = ['log.'+tf_name for tf_name in tf_names]
        if not get_step_from_list('LogMerge', check_steps):
            for step in reco_tf_steps:
                reco_tf_logmerge.log_files.append(step.get_log_file_name())
        reco_tf_logmerge.merged_name = 'athena.merged.log'
        log_to_zip = reco_tf_logmerge.merged_name
        if log_to_check is not None:
            reco_tf_logmerge.log_files.append(log_to_check)
            log_to_check = reco_tf_logmerge.merged_name
        log_to_check = reco_tf_logmerge.merged_name
        check_steps.append(reco_tf_logmerge)
 
    # Histogram merging for athenaHLT forks
    num_athenaHLT_steps = sum([1 for step in test.exec_steps if step.type == 'athenaHLT'])
    if num_athenaHLT_steps > 0:
        histmerge = RootMergeStep('HistMerge')
        histmerge.merged_file = 'expert-monitoring.root'
        histmerge.input_file = 'athenaHLT_workers/*/expert-monitoring.root expert-monitoring-mother.root'
        histmerge.rename_suffix = '-mother'
        check_steps.append(histmerge)
 
    # CheckLog for errors
    checklog = CheckLogStep('CheckLog')
    if log_to_check is not None:
        checklog.log_file = log_to_check
    check_steps.append(checklog)
 
    # CheckLog for warnings
    checkwarn = CheckLogStep('Warnings')
    checkwarn.check_errors = False
    checkwarn.check_warnings = True
    if log_to_check is not None:
        checkwarn.log_file = log_to_check
    check_steps.append(checkwarn)
 
    # MessageCount
    msgcount = MessageCountStep('MessageCount')
    for logmerge in [step for step in check_steps if isinstance(step, LogMergeStep)]:
        msgcount.skip_logs.append(logmerge.merged_name)
    check_steps.append(msgcount)
 
    # Tail (probably not so useful these days)
    tail = TailStep()
    if log_to_check is not None:
        tail.log_file = log_to_check
    check_steps.append(tail)
 
    # Histogram-based steps
    check_steps.append(RootCompStep())
    check_steps.append(ChainDumpStep())
    check_steps.append(HistCountStep())
 
    # ZeroCounts
    check_steps.append(ZeroCountsStep())
 
    # Extra JSON
    check_steps.append(TrigTestJsonStep())
 
    # CheckFile
    check_steps.append(CheckFileStep())
 
    # Zip the merged log (can be large and duplicates information)
    if log_to_zip is not None:
        zip_step = ZipStep()
        zip_step.zip_input = log_to_zip
        zip_step.zip_output = log_to_zip+'.tar.gz'
        check_steps.append(zip_step)
 
    # return the steps
    return check_steps
 
 

produces_log()

def python.TrigValSteering.CheckSteps.produces_log

(

step

)

Helper function checking whether a Step output_stream value
indicates that it will produce a log file

Definition at line 706 of file CheckSteps.py.

def produces_log(step):
    '''
    Helper function checking whether a Step output_stream value
    indicates that it will produce a log file
    '''
    return step.output_stream == Step.OutputStream.FILE_ONLY or \
           step.output_stream == Step.OutputStream.FILE_AND_STDOUT