python.DigitizationSteering Namespace Reference

Functions
	HepMCVersion ()
	DigitizationMainServicesCfg (flags)
	DigitizationMainCfg (flags)
	DigitizationMainContentCfg (flags)
	DigitizationMessageSvcCfg (flags)
	DigitizationTestingPostInclude (flags, acc)

Variables
	logDigiSteering = logging.getLogger('DigitizationSteering')

Detailed Description

Main steering for the digitization jobs

Copyright (C) 2002-2026 CERN for the benefit of the ATLAS collaboration

Function Documentation

DigitizationMainCfg()

DigitizationMainCfg

(

flags

)

Definition at line 53 of file DigitizationSteering.py.

def DigitizationMainCfg(flags):
    # Construct main services
    acc = DigitizationMainServicesCfg(flags)
 
    acc.merge(DigitizationMainContentCfg(flags))
 
    return acc
 

DigitizationMainContentCfg()

DigitizationMainContentCfg

(

flags

)

Definition at line 61 of file DigitizationSteering.py.

def DigitizationMainContentCfg(flags):
 
    acc = ComponentAccumulator()
 
    acc.merge(writeDigitizationMetadata(flags))
 
    if not flags.Digitization.PileUp:
        # Old EventInfo conversion
        if "EventInfo" not in flags.Input.Collections:
            from xAODEventInfoCnv.xAODEventInfoCnvConfig import EventInfoCnvAlgCfg
            acc.merge(EventInfoCnvAlgCfg(flags,
                                        inputKey="McEventInfo",
                                        outputKey="Input_EventInfo"))
 
        from xAODEventInfoCnv.xAODEventInfoCnvConfig import EventInfoUpdateFromContextAlgCfg
        acc.merge(EventInfoUpdateFromContextAlgCfg(flags))
 
        # Decorate pile-up values
        from DigitizationConfig.PileUpConfig import NoPileUpMuWriterCfg
        acc.merge(NoPileUpMuWriterCfg(flags))
 
    # Signal-only truth information
    if flags.Digitization.PileUp:
        if HepMCVersion() == 3:
            from MCTruthSimAlgs.MCTruthSimAlgsConfig import SimpleMergeMcEventCollCfg as MergeMcEventCollCfg
            from MCTruthSimAlgs.MCTruthSimAlgsConfig import InTimeOnlySimpleMergeMcEventCollCfg as InTimeOnlyMergeMcEventCollCfg
            from MCTruthSimAlgs.MCTruthSimAlgsConfig import SignalOnlySimpleMergeMcEventCollCfg as SignalOnlyMergeMcEventCollCfg
        else:
            from MCTruthSimAlgs.MCTruthSimAlgsConfig import MergeMcEventCollCfg
            from MCTruthSimAlgs.MCTruthSimAlgsConfig import InTimeOnlyMcEventCollCfg as InTimeOnlyMergeMcEventCollCfg
            from MCTruthSimAlgs.MCTruthSimAlgsConfig import SignalOnlyMcEventCollCfg as SignalOnlyMergeMcEventCollCfg
        from MCTruthSimAlgs.MCTruthSimAlgsConfig import (
            MergeAntiKt4TruthJetsCfg,
            MergeAntiKt6TruthJetsCfg,
            MergeTruthParticlesCfg,
            MergeMuonEntryLayerCfg,
            MergeCalibHitsCfg,
            MergeHijingParsCfg,
        )
        if flags.Common.ProductionStep is not ProductionStep.FastChain and getEnabledDetectors(flags):
            if flags.Digitization.DigiSteeringConf=="StandardPileUpToolsAlg":
                acc.merge(MergeMcEventCollCfg(flags))
            elif flags.Digitization.DigiSteeringConf=="StandardInTimeOnlyTruthPileUpToolsAlg":
                acc.merge(InTimeOnlyMergeMcEventCollCfg(flags))
            else:
                acc.merge(SignalOnlyMergeMcEventCollCfg(flags))
        if flags.Digitization.EnableTruth:
            puCollections = pileupInputCollections(flags.Digitization.PU.LowPtMinBiasInputCols)
            if "AntiKt4TruthJets" in puCollections:
                acc.merge(MergeAntiKt4TruthJetsCfg(flags))
            if "AntiKt6TruthJets" in puCollections:
                acc.merge(MergeAntiKt6TruthJetsCfg(flags))
            if "TruthPileupParticles" in puCollections:
                acc.merge(MergeTruthParticlesCfg(flags))
            acc.merge(MergeMuonEntryLayerCfg(flags))
            acc.merge(MergeCalibHitsCfg(flags))
            if 'Hijing_event_params' in pileupInputCollections(flags.Digitization.PU.CavernInputCols):
                acc.merge(MergeHijingParsCfg(flags))
 
 
    from DigitizationConfig.TruthDigitizationOutputConfig import TruthDigitizationOutputCfg
    acc.merge(TruthDigitizationOutputCfg(flags))
 
    # Beam spot reweighting
    if flags.Common.ProductionStep != ProductionStep.PileUpPresampling and flags.Digitization.InputBeamSigmaZ > 0:
        from BeamEffects.BeamEffectsAlgConfig import BeamSpotReweightingAlgCfg
        acc.merge(BeamSpotReweightingAlgCfg(flags))
 
    # Inner Detector
    if flags.Detector.EnableBCM:
        from BCM_Digitization.BCM_DigitizationConfig import BCM_DigitizationCfg
        acc.merge(BCM_DigitizationCfg(flags))
    if flags.Detector.EnablePixel:
        from PixelDigitization.PixelDigitizationConfig import PixelDigitizationCfg
        acc.merge(PixelDigitizationCfg(flags))
    if flags.Detector.EnableSCT:
        from SCT_Digitization.SCT_DigitizationConfig import SCT_DigitizationCfg
        acc.merge(SCT_DigitizationCfg(flags))
    if flags.Detector.EnableTRT:
        from TRT_Digitization.TRT_DigitizationConfig import TRT_DigitizationCfg
        acc.merge(TRT_DigitizationCfg(flags))
 
    # ITk
    if flags.Detector.EnableITkPixel:
        from PixelDigitization.ITkPixelDigitizationConfig import ITkPixelDigitizationCfg
        acc.merge(ITkPixelDigitizationCfg(flags))
    if flags.Detector.EnableITkStrip:
        from StripDigitization.StripDigitizationConfig import ITkStripDigitizationCfg
        acc.merge(ITkStripDigitizationCfg(flags))
    if flags.Detector.EnablePLR:
        from PixelDigitization.PLR_DigitizationConfig import PLR_DigitizationCfg
        acc.merge(PLR_DigitizationCfg(flags))
 
    # HGTD
    if flags.Detector.EnableHGTD:
        from HGTD_Digitization.HGTD_DigitizationConfig import HGTD_DigitizationCfg
        acc.merge(HGTD_DigitizationCfg(flags))
 
    # Calorimeter
    if flags.Detector.EnableLAr:
        from LArDigitization.LArDigitizationConfig import LArTriggerDigitizationCfg
        acc.merge(LArTriggerDigitizationCfg(flags))
    if flags.Detector.EnableTile:
        from TileSimAlgs.TileDigitizationConfig import TileDigitizationCfg, TileTriggerDigitizationCfg
        acc.merge(TileDigitizationCfg(flags))
        acc.merge(TileTriggerDigitizationCfg(flags))
 
    # Muon Spectrometer
    if flags.Detector.EnableMDT:
        from MuonConfig.MDT_DigitizationConfig import MDT_DigitizationDigitToRDOCfg
        acc.merge(MDT_DigitizationDigitToRDOCfg(flags))
    if flags.Detector.EnableTGC:
        from MuonConfig.TGC_DigitizationConfig import TGC_DigitizationDigitToRDOCfg
        acc.merge(TGC_DigitizationDigitToRDOCfg(flags))
    if flags.Detector.EnableRPC:
        from MuonConfig.RPC_DigitizationConfig import RPC_DigitizationDigitToRDOCfg
        acc.merge(RPC_DigitizationDigitToRDOCfg(flags))
    if flags.Detector.EnableCSC:
        from MuonConfig.CSC_DigitizationConfig import CSC_DigitizationDigitToRDOCfg
        acc.merge(CSC_DigitizationDigitToRDOCfg(flags))
    if flags.Detector.EnablesTGC:
        from MuonConfig.sTGC_DigitizationConfig import sTGC_DigitizationDigitToRDOCfg
        acc.merge(sTGC_DigitizationDigitToRDOCfg(flags))
    if flags.Detector.EnableMM:
        from MuonConfig.MM_DigitizationConfig import MM_DigitizationDigitToRDOCfg
        acc.merge(MM_DigitizationDigitToRDOCfg(flags))
 
    # LUCID
    if flags.Detector.EnableLucid:
        from LUCID_Digitization.LUCID_DigitizationConfig import LUCID_DigitizationCfg
        acc.merge(LUCID_DigitizationCfg(flags))
 
    # AFP
    if flags.Detector.EnableAFP:
        from AFP_Digitization.AFP_DigitizationConfig import AFP_DigitizationCfg
        acc.merge(AFP_DigitizationCfg(flags))
 
    # ALFA
    if flags.Detector.EnableALFA:
        from ALFA_Digitization.ALFA_DigitizationConfig import ALFA_DigitizationCfg
        acc.merge(ALFA_DigitizationCfg(flags))
 
    # ZDC
    if flags.Detector.EnableZDC:
        from ZDC_SimuDigitization.ZDC_SimuDigitizationConfig import ZDC_DigitizationCfg
        acc.merge(ZDC_DigitizationCfg(flags))
 
    # Add in-file MetaData
    from xAODMetaDataCnv.InfileMetaDataConfig import SetupMetaDataForStreamCfg
    acc.merge(SetupMetaDataForStreamCfg(flags, "RDO"))
 
    return acc
 
 

DigitizationMainServicesCfg()

DigitizationMainServicesCfg

(

flags

)

Configure main digitization services

Definition at line 28 of file DigitizationSteering.py.

def DigitizationMainServicesCfg(flags):
    """Configure main digitization services"""
    if flags.Digitization.PileUp:
        if flags.Concurrency.NumThreads > 0:
            logDigiSteering.info("DigitizationMainServicesCfg: Attempting to run pile-up digitization AthenaMT using %s threads!", str(flags.Concurrency.NumThreads))
            logDigiSteering.info("DigitizationMainServicesCfg: Using new PileUpMT code.")
            # raise RuntimeError("DigitizationSteering.DigitizationMainServicesCfg: Running pile-up digitization with AthenaMT is not supported. Please update your configuration.")
            from DigitizationConfig.PileUpMTConfig import PileUpMTAlgCfg
            acc = MainServicesCfg(flags)
            acc.merge(PileUpMTAlgCfg(flags))
        else:
            from DigitizationConfig.PileUpConfig import PileUpEventLoopMgrCfg
            # For MP, we let MainServicesCfg create the MpEventLoopMgr
            acc = MainServicesCfg(flags, createEventLoopMgr=(flags.Concurrency.NumProcs > 0))
            acc.merge(PileUpEventLoopMgrCfg(flags))
    else:
        acc = MainServicesCfg(flags)
 
    acc.merge(PoolReadCfg(flags))
    evSel = acc.getService("EventSelector")
    evSel.CollectionType = "RootCollection"
 
    return acc
 
 

DigitizationMessageSvcCfg()

DigitizationMessageSvcCfg

(

flags

)

MessageSvc for digitization and overlay

Definition at line 215 of file DigitizationSteering.py.

def DigitizationMessageSvcCfg(flags):
    """MessageSvc for digitization and overlay"""
    MessageSvc = CompFactory.MessageSvc
    acc = ComponentAccumulator()
    acc.addService(MessageSvc(setError=["HepMcParticleLink"]))
    return acc
 
 

DigitizationTestingPostInclude()

DigitizationTestingPostInclude	(		flags,
			acc )

Testing digitization post-include

Definition at line 223 of file DigitizationSteering.py.

def DigitizationTestingPostInclude(flags, acc):
    """Testing digitization post-include"""
    # dump config
    configName = "DigiPUConfigCA" if flags.Digitization.PileUp else "DigiConfigCA"
    from AthenaConfiguration.MainServicesConfig import JobOptionsDumpCfg
    acc.merge(JobOptionsDumpCfg(flags, fileName=f"{configName}.txt"))
 
    # dump pickle
    with open(f"{configName}.pkl", "wb") as f:
        acc.store(f)

HepMCVersion()

HepMCVersion

(

)

Definition at line 19 of file DigitizationSteering.py.

def HepMCVersion():
    try:
        from AthenaPython.PyAthena import HepMC3 # noqa: F401
        HepMCVersion=3
    except ImportError:
        HepMCVersion=2
    return HepMCVersion
 
 

Variable Documentation

logDigiSteering

python.DigitizationSteering.logDigiSteering = logging.getLogger('DigitizationSteering')

Definition at line 17 of file DigitizationSteering.py.