TrigInDetArtSteps.py

Go to the documentation of this file.

#
# Copyright (C) 2002-2026 CERN for the benefit of the ATLAS collaboration
#
 
'''
Definitions of additional validation steps in Trigger ART tests relevant only for TrigInDetValidation
The main common check steps are defined in the TrigValSteering.CheckSteps module.
'''
 
import os
import subprocess
import json
 
from TrigValTools.TrigValSteering.ExecStep import ExecStep
from TrigValTools.TrigValSteering.Step import Step
from TrigValTools.TrigValSteering.CheckSteps import RefComparisonStep
from TrigValTools.TrigValSteering.Common import find_file
from AthenaCommon.Utils.unixtools import FindFile
from AthenaConfiguration.TestDefaults import defaultConditionsTags
 

 
class TrigInDetReco(ExecStep):
 
    def __init__(self, name='TrigInDetReco', postinclude_file='', preinclude_file=''):
        ExecStep.__init__(self, name)

        self.type = 'Reco_tf'
        self.max_events=-1
        self.required = True
        self.threads = 1 # TODO: change to 4
        self.concurrent_events = 1 # TODO: change to 4
        self.perfmon = False
        self.timeout = 18*3600
        self.slices = []
        self.preexec_trig = ''
        self.postinclude_trig = postinclude_file
        self.preinclude_trig  = preinclude_file
        self.release = 'current'
        self.preexec_reco =  ';'.join([
            'flags.Reco.EnableEgamma=True',
            'flags.Reco.EnableCombinedMuon=True',
            'flags.Reco.EnableJet=False',
            'flags.Reco.EnableMet=False',
            'flags.Reco.EnableBTagging=False',
            'flags.Reco.EnablePFlow=False',
            'flags.Reco.EnableTau=False',
            'flags.Reco.EnablePostProcessing=False',
        ])
        self.preexec_all = ';'.join([
            'flags.Trigger.AODEDMSet=\'ESD\'',
        ])
        self.postexec_trig = ''
        self.postexec_reco = ''
        self.args = '--outputAODFile=AOD.pool.root --steering "doRDO_TRIG"'
        self.args += ' --CA'
       
        if ( self.postinclude_trig != '' ) : 
            print( "postinclude_trig: ", self.postinclude_trig )
 
        if ( self.preinclude_trig != '' ) : 
            print( "preinclude_trig:  ", self.preinclude_trig  )
 
 
    def configure(self, test):
        from TrigInDetValidation.Chains import Chains
        tc = Chains()
        self.preexec_trig += tc.get_compiled_flag_str(self.slices)
 
        AVERSION = ""
        
        if (self.release != 'current'):
            # get the current atlas base release, and the previous base release
            import os
            DVERSION=os.getenv('Athena_VERSION')
            if (self.release == 'latest'):
                if ( DVERSION is None ) :
                    AVERSION = "22.0.20"
                else:
                    AVERSION=str(subprocess.Popen(["getrelease.sh",DVERSION],stdout=subprocess.PIPE).communicate()[0],'utf-8')
                    if AVERSION == "":
                        print( "cannot get last stable release - will use current release" )
            else:
                AVERSION = self.release
 
        # would use AVERSION is not None, but the return from a shell function with no printout 
        # gets set as an empty string rather than None        
        if AVERSION != "":
            self.args += ' --asetup "RAWtoALL:Athena,'+AVERSION+'" '
            print( "remapping athena base release version for offline Reco steps: ", DVERSION, " -> ", AVERSION )
        else:
            print( "Using current release for offline Reco steps  " )
 
        if self.preexec_trig != '' or self.preexec_reco != '' or self.preexec_all != '':
            self.args += ' --preExec'
            if self.preexec_trig != '':
                self.args += ' "RDOtoRDOTrigger:{:s};"'.format(self.preexec_trig)
            if self.preexec_reco != '':
                self.args += ' "RAWtoALL:{:s};"'.format(self.preexec_reco)
            if self.preexec_all != '':
                self.args += ' "all:{:s};"'.format(self.preexec_all)
        if self.postexec_trig != '' or self.postexec_reco != '':
            self.args += ' --postExec'
            if self.postexec_trig != '':
                self.args += ' "RDOtoRDOTrigger:{:s};"'.format(self.postexec_trig)
            if self.postexec_reco != '':
                self.args += ' "RAWtoALL:{:s};"'.format(self.postexec_reco)
        if (self.postinclude_trig != ''):
            self.args += ' --postInclude "{:s}"'.format(self.postinclude_trig)
        if (self.preinclude_trig != ''):
            self.args += ' --preInclude "{:s}"'.format(self.preinclude_trig)
        super(TrigInDetReco, self).configure(test)
 
 

 
class TrigInDetAna(ExecStep):
    def __init__(self, name='TrigInDetAna', extraArgs=None):
        ExecStep.__init__(self, name )
        self.type = 'other'
        self.executable = 'python'
        self.args = ' -m TrigInDetValidation.TrigInDetValidation_AODtoTrkNtuple_CA '
        self.max_events=-1
        self.required = True
        self.depends_on_previous = True
        #self.input = 'AOD.pool.root'
        self.input = ''
        if extraArgs is not None:
            self.args += extraArgs
 
 

 
class TrigCostStep(Step):
    def __init__(self, name='TrigCostStep'):
        super(TrigCostStep, self).__init__(name)
        self.required = True
        self.depends_on_previous = True
        self.input = 'tmp.RDO_TRIG'
        self.args = '  --monitorChainAlgorithm --MCCrossSection=0.5 Input.Files=\'["tmp.RDO_TRIG"]\' '
        self.executable = 'RunTrigCostAnalysis.py'
 
 

class TrigInDetRecoData(ExecStep):
    def __init__(self, name='TrigInDetRecoData'):
#        super(TrigInDetRecoData, self).__init__(name)
        ExecStep.__init__(self, name)
        self.type = 'athenaHLT'
        self.job_options = 'TriggerJobOpts.runHLT'
        self.max_events=-1
        self.required = True
        self.threads = 1 # TODO: change to 4
        self.concurrent_events = 1 # TODO: change to 4
        self.perfmon = False
        self.timeout = 18*3600
        self.input = ''
        self.perfmon=False
        self.imf=False
        self.args = '-c "setMenu=\'Cosmic_run3_v1\';doCosmics=True;doL1Sim=False;forceEnableAllChains=True;"'
        self.args = '-c "flags.Trigger.forceEnableAllChains=True;flags.Trigger.triggerMenuSetup=\'Cosmic_run3_v1\';from AthenaConfiguration.Enums import BeamType;flags.Beam.Type=BeamType.Cosmics;"'
        self.args += ' -o output'
 
 

 
class TrigTZReco(ExecStep):
    def __init__(self, name='TrigTZReco'):
        super(TrigTZReco, self).__init__(name)
        self.type = 'Reco_tf'
        tzrecoPreExec = ' '.join([
            "flags.Trigger.triggerMenuSetup=\'Cosmic_run3_v1\';",
            "flags.Trigger.AODEDMSet=\'AODFULL\';",
            ])
        self.threads = 1
        self.concurrent_events = 1
        self.input = ''
        self.explicit_input = True
        self.max_events = -1
        self.args = '--inputBSFile=' + find_file('*.physics_Main.*.data')  # output of the previous step
        self.args += ' --outputAODFile=AOD.pool.root'
        self.args += f' --conditionsTag="{defaultConditionsTags.RUN3_DATA23}" --geometryVersion=\'ATLAS-R3S-2021-03-02-00\''
        self.args += ' --preExec="{:s}"'.format(tzrecoPreExec)
        self.args += ' --CA'
 
 

 
class TrigInDetRdictStep(Step):
    '''
    Execute TIDArdict for TrkNtuple files.
    '''
    def __init__(self, name='TrigInDetdict', args=None, testbin='Test_bin.dat', config=False ):
        super(TrigInDetRdictStep, self).__init__(name)
        self.args=args + "  -b " + testbin + " "
        self.auto_report_result = True
        self.required = True
        self.executable = 'TIDArdict'
        self.timeout = 5*3600
        self.config = config
 
    def configure(self, test):
        if not self.config :
            os.system( 'get_files -data TIDAbeam.dat &> /dev/null' )
            os.system( 'get_files -data Test_bin.dat &> /dev/null' )
            os.system( 'get_files -data Test_bin_larged0.dat &> /dev/null' )
            os.system( 'get_files -data Test_bin_lrt.dat &> /dev/null' )
            os.system( 'get_files -data TIDAdata-run3.dat &> /dev/null' )
            os.system( 'get_files -data TIDAdata-run3-larged0.dat &> /dev/null' )
            os.system( 'get_files -data TIDAdata-run3-larged0-el.dat &> /dev/null' )
            os.system( 'get_files -data TIDAdata-run3-lrt.dat &> /dev/null' )
            os.system( 'get_files -data TIDAdata-run3-fslrt.dat &> /dev/null' )
            os.system( 'get_files -data TIDAdata-run3-minbias.dat &> /dev/null' )
            os.system( 'get_files -data TIDAdata_cuts.dat &> /dev/null' )
            os.system( 'get_files -data TIDAdata-run3-offline.dat &> /dev/null' )
            os.system( 'get_files -data TIDAdata-run3-offline-rzMatcher.dat &> /dev/null' )
            os.system( 'get_files -data TIDAdata-run3-offline-vtxtrack.dat &> /dev/null' )
            os.system( 'get_files -data TIDAdata-run3-offline-larged0.dat &> /dev/null' )
            os.system( 'get_files -data TIDAdata-run3-offline-larged0-el.dat &> /dev/null' )
            os.system( 'get_files -data TIDAdata-run3-offline-lrt.dat &> /dev/null' )
            os.system( 'get_files -data TIDAdata-run3-offline-fslrt.dat &> /dev/null' )
            os.system( 'get_files -data TIDAdata-run3-offline-vtx.dat &> /dev/null' )
            os.system( 'get_files -data TIDAdata-run3-minbias-offline.dat &> /dev/null' )
            os.system( 'get_files -data TIDAdata-run3-offline-cosmic.dat &> /dev/null' )
            os.system( 'get_files -data TIDAdata_cuts-offline.dat &> /dev/null' )
            os.system( 'get_files -data TIDAdata-chains-run3.dat &> /dev/null' )
            os.system( 'get_files -data TIDAdata-chains-run3-lrt.dat &> /dev/null' )
            os.system( 'get_files -data TIDAdata-run4.dat &> /dev/null' )
            os.system( 'get_files -data TIDAdata-run4-offline.dat &> /dev/null' )
            os.system( 'get_files -data TIDAdata-run4-offline-vtx.dat &> /dev/null' )
        super(TrigInDetRdictStep, self).configure(test)
 
 
def json_chains( slice ) : 
    json_file     = 'TrigInDetValidation/comparitor.json'
    json_fullpath = FindFile(json_file, os.environ['DATAPATH'].split(os.pathsep), os.R_OK)
 
    if not json_fullpath:
        print('Failed to determine full path for input JSON %s', json_file)
        return None
        
    with open(json_fullpath) as f:
        try:
            data = json.load(f)
        except json.decoder.JSONDecodeError as e:
            print(f"Failed to load json file {json_fullpath}")
            raise e
 
    chainmap = data[slice]
 
    return chainmap['chains']
 
 
 
class TrigInDetCompStep(RefComparisonStep):
    '''
    Execute TIDAcomparitor for data.root files.
    '''
    def __init__( self, name='TrigInDetComp', slice=None, args=None, file=None ):
        super(TrigInDetCompStep, self).__init__(name)
 
        self.input_file = file    
        self.slice = slice 
        self.auto_report_result = True
        self.required   = True
        self.args = args
        self.executable = 'TIDAcomparitor'
        os.system( 'get_files -data TIDAhisto-panel.dat &> /dev/null' )
        os.system( 'get_files -data TIDAhisto-panel-vtx.dat &> /dev/null' )
        os.system( 'get_files -data TIDAhistos-vtx.dat &> /dev/null' )
        os.system( 'get_files -data TIDAhisto-panel-TnP.dat &> /dev/null' )
        os.system( 'get_files -data TIDAhisto-tier0.dat &> /dev/null' )
        os.system( 'get_files -data TIDAhisto-tier0-vtx.dat &> /dev/null' )
        os.system( 'get_files -data TIDAhisto-tier0-TnP.dat &> /dev/null' )    
 
    def configure(self, test):
        RefComparisonStep.configure(self, test)
        if self.reference is None :
            self.args  = self.args + " " + self.input_file + "  " + self.input_file + " --noref --oldrms "
        else:
            self.args  = self.args + " " + self.input_file + "  " + self.reference + " --oldrms "
        self.chains = json_chains( self.slice )
        self.args += " " + self.chains
        print( "\033[0;32mTIDAcomparitor "+self.args+" \033[0m" )
        Step.configure(self, test)
 
 
 
 
class TrigInDetCpuCostStep(RefComparisonStep):
    '''
    Execute TIDAcpucost for data.root files.
    '''
    def __init__( self, name='TrigInDetCpuCost', outdir=None, infile=None, extra=None ):
        super(TrigInDetCpuCostStep, self).__init__(name)
 
        self.input_file = infile
        self.output_dir = outdir
        self.auto_report_result = True
        self.required   = True
        self.extra = extra
        self.executable = 'TIDAcpucost'
    
 
    def configure(self, test):
        RefComparisonStep.configure(self, test)
        if self.reference is None :
            self.args  = self.input_file + " -o " + self.output_dir + " " + self.extra + " --noref  "
        else:
            self.args  = self.input_file + " " + self.reference + " -o " + self.output_dir + " " + self.extra
        Step.configure(self, test)