StandardTests.py

Go to the documentation of this file.

# Copyright (C) 2002-2026 CERN for the benefit of the ATLAS collaboration
from .Checks import (
    AODContentCheck,
    AODDigestCheck,
    FrozenTier0PolicyCheck,
    MetadataCheck,
)
from .Inputs import (
    input_AOD,
    input_BS_minimum_bias_overlay,
    input_EVNT,
    input_EVNT_data_overlay,
    input_HITS,
    input_HITS_data_overlay,
    input_HITS_minbias_high,
    input_HITS_minbias_high_fulltruth,
    input_HITS_minbias_low,
    input_HITS_minbias_low_fulltruth,
    input_HITS_neutrino,
    input_HITS_unfiltered,
    input_RDO_BKG,
    input_RDO_BKG_data,
)
from .Test import TestSetup, WorkflowRun, WorkflowTest, WorkflowType
 
 
class QTest(WorkflowTest):
    """General workflow q-test."""
 
    def __init__(self, ID: str, run: WorkflowRun, type: WorkflowType, steps: list[str], setup: TestSetup, extra_args: str = "") -> None:
        if "maxEvents" not in extra_args:
            if type == WorkflowType.MCPileUpReco or run == WorkflowRun.Run4:
                extra_args += " --maxEvents 5"
            else:
                extra_args += " --maxEvents 20"
 
        if type == WorkflowType.MCPileUpReco:
            if "inputHITSFile" not in extra_args:
                extra_args += f" --inputHITSFile {input_HITS[run]}"
            if "inputRDO_BKGFile" not in extra_args:
                extra_args += " --inputRDO_BKGFile ../run_d*/myRDO.pool.root"
 
        threads = 1
        threads_argument = '--multithreaded'
        if setup.custom_threads is not None:
            threads = setup.custom_threads
        if threads <= 0:
            threads_argument = ''
 
        self.command = \
            (f"ATHENA_CORE_NUMBER={threads} Reco_tf.py {threads_argument} --AMIConfig {ID}"
             f" --imf False {extra_args}")
 
        self.output_checks = []
        # TODO: disable RDO comparison for now
        # if type == WorkflowType.MCReco:
        #     self.output_checks.append(FrozenTier0PolicyCheck(setup, "RDO", 10))
        self.output_checks.append(FrozenTier0PolicyCheck(setup, "AOD", 60))
        self.output_checks.append(FrozenTier0PolicyCheck(setup, "ESD", 20))
        self.output_checks.append(MetadataCheck(setup, "AOD"))
 
        self.digest_checks = []
        if not setup.disable_output_checks:
            self.digest_checks.append(AODContentCheck(setup))
        self.digest_checks.append(AODDigestCheck(setup))
 
        super().__init__(ID, run, type, steps, setup)
 
 
class SimulationTest(WorkflowTest):
    """Simulation workflow test."""
 
    def __init__(self, ID: str, run: WorkflowRun, type: WorkflowType, steps: list[str], setup: TestSetup, extra_args: str = "") -> None:
        if "maxEvents" not in extra_args:
            extra_args += " --maxEvents 20"
 
        if "jobNumber" not in extra_args and run is WorkflowRun.Run3 and type is WorkflowType.FullSim:
            extra_args += " --jobNumber 5"
 
        input_argument = ""
        if "inputEVNTFile" not in extra_args and "inputHITSFile" not in extra_args:
            if type is WorkflowType.HitsFilter:
                input_argument = f"--inputHITSFile {input_HITS_unfiltered[run]}"
            elif type is WorkflowType.HitsMerge:
                input_argument = f"--inputHITSFile {input_HITS[run]}"
            else:
                input_argument = f"--inputEVNTFile {input_EVNT[run]}"
 
        threads = 0
        threads_argument = '--multithreaded'
        if setup.custom_threads is not None:
            threads = setup.custom_threads
        if threads <= 0:
            threads_argument = ''
 
        if type is WorkflowType.HitsMerge:
            self.command = \
                (f"ATHENA_CORE_NUMBER={threads} HITSMerge_tf.py {threads_argument} --AMIConfig {ID}"
                f" {input_argument} --outputHITS_MRGFile myHITS.pool.root"
                f" --imf False {extra_args}")
        elif type is WorkflowType.HitsFilter:
            self.command = \
                (f"ATHENA_CORE_NUMBER={threads} FilterHit_tf.py {threads_argument} --AMIConfig {ID}"
                f" {input_argument} --outputHITS_FILTFile myHITS.pool.root"
                f" --imf False {extra_args}")
        else:
            self.command = \
                (f"ATHENA_CORE_NUMBER={threads} Sim_tf.py {threads_argument} --AMIConfig {ID}"
                f" {input_argument} --outputHITSFile myHITS.pool.root"
                f" --imf False {extra_args}")
 
        self.output_checks = [
            FrozenTier0PolicyCheck(setup, "HITS", 10),
            MetadataCheck(setup, "HITS"),
        ]
 
        super().__init__(ID, run, type, steps, setup)
 
 
class OverlayTest(WorkflowTest):
    """MC overlay workflow test."""
 
    def __init__(self, ID: str, run: WorkflowRun, type: WorkflowType, steps: list[str], setup: TestSetup, extra_args: str = "") -> None:
        if "maxEvents" not in extra_args:
            extra_args += " --maxEvents 10"
 
        self.command = \
            (f"Overlay_tf.py --AMIConfig {ID}"
             f" --inputHITSFile {input_HITS[run]} --inputRDO_BKGFile {input_RDO_BKG[run]} --outputRDOFile myRDO.pool.root"
             f" --imf False {extra_args}")
 
        # skip performance checks for now
        self.skip_performance_checks = True
 
        self.output_checks = [
            FrozenTier0PolicyCheck(setup, "RDO", 10),
            MetadataCheck(setup, "RDO"),
        ]
 
        super().__init__(ID, run, type, steps, setup)
 
 
class DataOverlayTest(WorkflowTest):
    """Data overlay workflow test."""
 
    def __init__(self, ID: str, run: WorkflowRun, type: WorkflowType, steps: list[str], setup: TestSetup, extra_args: str = "") -> None:
        if "maxEvents" not in extra_args:
            extra_args += " --maxEvents 10"
 
        if type is WorkflowType.DataOverlayChain:
            self.command = \
                (f"FastChain_tf.py --AMIConfig {ID}"
                f" --inputEVNTFile {input_EVNT_data_overlay[run]} --inputRDO_BKGFile {input_RDO_BKG_data[run]} --outputHITSFile myHITS.pool.root --outputRDOFile myRDO.pool.root"
                f" --imf False {extra_args}")
        else:
            self.command = \
                (f"Overlay_tf.py --AMIConfig {ID}"
                f" --inputHITSFile {input_HITS_data_overlay[run]} --inputRDO_BKGFile {input_RDO_BKG_data[run]} --outputRDOFile myRDO.pool.root"
                f" --imf False {extra_args}")
 
        self.output_checks = []
        if type is WorkflowType.DataOverlayChain:
            self.output_checks.extend([
                FrozenTier0PolicyCheck(setup, "HITS", 10),
                MetadataCheck(setup, "HITS"),
            ])
 
        self.output_checks.extend([
            FrozenTier0PolicyCheck(setup, "RDO", 10),
            MetadataCheck(setup, "RDO"),
        ])
 
        super().__init__(ID, run, type, steps, setup)
 
 
class PileUpTest(WorkflowTest):
    """Digitization with pile-up workflow test."""
 
    def __init__(self, ID: str, run: WorkflowRun, type: WorkflowType, steps: list[str], setup: TestSetup, extra_args: str = "") -> None:
        if "maxEvents" not in extra_args:
            extra_args += " --maxEvents 5"
 
        if "StandardInTimeOnlyTruth" in extra_args:
            input_high = input_HITS_minbias_high_fulltruth[run]
            input_low = input_HITS_minbias_low_fulltruth[run]
        else:
            input_high = input_HITS_minbias_high[run]
            input_low = input_HITS_minbias_low[run]
 
        self.command = \
            (f"Digi_tf.py --AMIConfig {ID} --jobNumber 1 --digiSeedOffset1 1 --digiSeedOffset2 1"
             f" --inputHITSFile {input_HITS_neutrino[run]} --inputHighPtMinbiasHitsFile {input_high} --inputLowPtMinbiasHitsFile {input_low} --outputRDOFile myRDO.pool.root"
             f" --imf False {extra_args}")
 
        self.output_checks = [
            FrozenTier0PolicyCheck(setup, "RDO", 5),
            MetadataCheck(setup, "RDO"),
        ]
 
        super().__init__(ID, run, type, steps, setup)
 
 
class DataOverlayPreparationTest(WorkflowTest):
    """Data overlay minimum bias preprocessing test."""
 
    def __init__(self, ID: str, run: WorkflowRun, type: WorkflowType, steps: list[str], setup: TestSetup, extra_args: str = "") -> None:
        if "maxEvents" not in extra_args:
            extra_args += " --maxEvents 5"
 
        self.command = \
            (f"Overlay_tf.py --AMIConfig {ID}"
             f" --inputBSFile {input_BS_minimum_bias_overlay[run]} --outputRDO_BKGFile myRDO_BKG.pool.root"
             f" --imf False {extra_args}")
 
        # TODO: enable once the output stabilises
        # self.output_checks = [
        #     FrozenTier0PolicyCheck(setup, "RDO_BKG", 5),
        #     MetadataCheck(setup, "RDO_BKG"),
        # ]
 
        super().__init__(ID, run, type, steps, setup)
 
 
class DerivationTest(WorkflowTest):
    """Derivations test."""
 
    def __init__(self, ID: str, run: WorkflowRun, type: WorkflowType, steps: list[str], setup: TestSetup, extra_args: str = "") -> None:
        self.diff_tolerances = "PROCTools/DAOD_diff-tolerances.yaml"
 
        test_def = ID.split("_")
        multithreaded = False
        if test_def[-2] == "MT":
            multithreaded = True
            test_run = test_def.pop()
            test_def.pop()  # remove MT
            test_def.append(test_run)  # add run back to the end
        data_type = test_def[0].lower()
        formats = [format.upper() for format in test_def[1:-1]]
 
        threads = 0
        if setup.custom_threads is not None:
            threads = setup.custom_threads
 
        # If we're running a DAOD to DAOD job, the output stream is called D2AOD
        out_stream = "DAOD"
        parallel_compression = "True"
        if "inputDAOD_PHYSFile" in extra_args:
            out_stream = "D2AOD"
            parallel_compression = "False" # Disable parallel compression for DAOD to DAOD tests (ATEAM-1194)
 
        if "maxEvents" not in extra_args:
            base_events = 100
            events = threads * base_events + 1
            extra_args += f" --maxEvents {events}"
 
            if not multithreaded:
                flush = 80
                format_flush = ", ".join([f"\"{out_stream}_{format}\": {flush}" for format in formats])
                extra_args += f" --preExec 'flags.Output.TreeAutoFlush={{{format_flush}}}'"
 
        if "inputAODFile" not in extra_args and "inputDAOD_PHYSFile" not in extra_args:
            extra_args += f" --inputAODFile {input_AOD[run][data_type]}"
 
        # could also use p5503
        if not multithreaded:
            self.command = \
                (f"ATHENA_CORE_NUMBER={threads} Derivation_tf.py"
                f" --formats {' '.join(formats)}"
                " --multiprocess --multithreadedFileValidation True"
                f" --athenaMPMergeTargetSize '{out_stream}_*:0'"
                " --sharedWriter True"
                f" --parallelCompression {parallel_compression}"
                f" --output{out_stream}File myOutput.pool.root"
                f" --imf False {extra_args}")
        else:
            self.command = \
                (f"ATHENA_CORE_NUMBER={threads} Derivation_tf.py"
                f" --formats {' '.join(formats)}"
                " --multithreaded"
                f" --output{out_stream}File myOutput.pool.root"
                f" --imf False {extra_args}")
 
        # skip performance checks for now
        self.skip_performance_checks = True
 
        self.output_checks = []
        for format in formats:
            self.output_checks.append(FrozenTier0PolicyCheck(setup, f"{out_stream}_{format}", 10))
            self.output_checks.append(MetadataCheck(setup, f"{out_stream}_{format}"))
 
        super().__init__("_".join(test_def), run, type, steps, setup)
 
 
class GenerationTest(WorkflowTest):
    """Generation test."""
 
    def __init__(self, ID: str, run: WorkflowRun, type: WorkflowType, steps: list[str], setup: TestSetup, extra_args: str = "") -> None:
        if "output" not in extra_args:
            extra_args += " --outputEVNTFile myEVNT.pool.root"
 
        if "maxEvents" not in extra_args:
            extra_args += " --maxEvents 10"
 
        if "ecmEnergy" not in extra_args:
            if run is WorkflowRun.Run2:
                extra_args += " --ecmEnergy 13000"
            elif run is WorkflowRun.Run3:
                extra_args += " --ecmEnergy 13600"
            else:
                extra_args += " --ecmEnergy 14000"
 
        dsid = ID.replace("gen", "")
 
        self.command = \
            (f"Gen_tf.py --jobConfig {dsid}"
             f" --imf False {extra_args}")
 
        super().__init__(ID, run, type, steps, setup)