OverlaySteering.py

Go to the documentation of this file.

#!/usr/bin/env python
"""Main steering for MC+MC and MC+data overlay
 
Copyright (C) 2002-2025 CERN for the benefit of the ATLAS collaboration
"""
 
from AthenaConfiguration.MainServicesConfig import MainServicesCfg
from AthenaConfiguration.DetectorConfigFlags import getEnabledDetectors
from AthenaConfiguration.Enums import LHCPeriod
from AthenaKernel.EventIdOverrideConfig import IOVDbMetaDataToolWithRunNumberOverrideCfg
from AthenaPoolCnvSvc.PoolReadConfig import PoolReadCfg
from DigitizationConfig.DigitizationParametersConfig import writeDigitizationParameters
from OverlayCopyAlgs.OverlayCopyAlgsConfig import \
    CopyCaloCalibrationHitContainersCfg, CopyJetTruthInfoCfg, CopyPileupParticleTruthInfoCfg, CopyMcEventCollectionCfg, \
    CopyTrackRecordCollectionsCfg, CopyBackgroundVertexCfg
from PyJobTransforms.TransformUtils import executeFromFragment
from xAODEventInfoCnv.xAODEventInfoCnvConfig import EventInfoOverlayCfg
 
 
def OverlayMainCfg(configFlags):
    """Main overlay steering configuration"""
 
    # Construct our accumulator to run
    acc = MainServicesCfg(configFlags)
    acc.merge(PoolReadCfg(configFlags))
    acc.merge(OverlayMainContentCfg(configFlags))
    return acc
 
 
def OverlayMainContentCfg(configFlags):
    """Main overlay content"""
 
    from AthenaConfiguration.ComponentAccumulator import ComponentAccumulator
    acc = ComponentAccumulator()
 
    # Handle overlay digitization metadata writing (must come before reading setup)
    if hasattr(configFlags, '_Overlay_pileupDigitizationMetadata'):
        if configFlags.IOVDb.WriteParametersAsMetaData:
            # Direct metadata mode: use standard parameter writer
            from IOVDbMetaDataTools.ParameterWriterConfig import writeParametersToMetaData
            from DigitizationConfig.DigitizationParametersConfig import folderName
            from AthenaCommon.Logging import logging
            logOverlay = logging.getLogger('OverlayMetadataConfig')
            logOverlay.info('Writing overlay digitization parameters directly to in-file metadata (bypassing DigitParams.db)')
 
            pileupDict = configFlags._Overlay_pileupDigitizationMetadata
            runNumber = configFlags.Input.RunNumbers[0]
            runNumberEnd = configFlags.Input.RunNumbers[-1]
            if runNumberEnd == runNumber:
                runNumberEnd += 1
 
            # Convert pileup dictionary to string parameters
            params = {}
            for key in pileupDict:
                value = str(pileupDict[key])
                params[key] = value
                logOverlay.info('DigitizationMetaData: setting "%s" to be %s', key, value)
 
            acc.merge(writeParametersToMetaData(configFlags, folderName, params, runNumber, runNumberEnd))
        else:
            # Sqlite mode: create DigitParams.db first
            from OverlayConfiguration.OverlayMetadata import writeOverlayDigitizationMetadata
            writeOverlayDigitizationMetadata(configFlags, configFlags._Overlay_pileupDigitizationMetadata)
 
    # Setup reading digitization parameters (mode-dependent)
    # In sqlite mode, this reads from DigitParams.db (created above if pileup metadata exists)
    # Only needed if we actually have pileup metadata to read
    if not configFlags.IOVDb.WriteParametersAsMetaData and hasattr(configFlags, '_Overlay_pileupDigitizationMetadata'):
        acc.merge(writeDigitizationParameters(configFlags))
 
    if not configFlags.Overlay.ByteStream:
        acc.merge(IOVDbMetaDataToolWithRunNumberOverrideCfg(configFlags))
 
    # Add event info overlay
    if not configFlags.Sim.DoFullChain:
        acc.merge(EventInfoOverlayCfg(configFlags))
 
    # Add truth overlay (needed downstream)
    if not configFlags.Sim.DoFullChain and (getEnabledDetectors(configFlags) or configFlags.Digitization.EnableTruth):
        acc.merge(CopyMcEventCollectionCfg(configFlags))
    if configFlags.Digitization.EnableTruth:
        acc.merge(CopyJetTruthInfoCfg(configFlags))
        acc.merge(CopyPileupParticleTruthInfoCfg(configFlags))
        acc.merge(CopyCaloCalibrationHitContainersCfg(configFlags))
        if not configFlags.Sim.DoFullChain:
            acc.merge(CopyTrackRecordCollectionsCfg(configFlags))
 
    # Beam spot reweighting
    if configFlags.Digitization.InputBeamSigmaZ > 0:
        from BeamEffects.BeamEffectsAlgConfig import BeamSpotReweightingAlgCfg
        acc.merge(BeamSpotReweightingAlgCfg(configFlags))
 
    # Inner detector
    if configFlags.Detector.EnableBCM:
        from InDetOverlay.BCMOverlayConfig import BCMOverlayCfg
        acc.merge(BCMOverlayCfg(configFlags))
    if configFlags.Detector.EnablePixel:
        from InDetOverlay.PixelOverlayConfig import PixelOverlayCfg
        acc.merge(PixelOverlayCfg(configFlags))
    if configFlags.Detector.EnableSCT:
        from InDetOverlay.SCTOverlayConfig import SCTOverlayCfg
        acc.merge(SCTOverlayCfg(configFlags))
    if configFlags.Detector.EnableTRT:
        from InDetOverlay.TRTOverlayConfig import TRTOverlayCfg
        acc.merge(TRTOverlayCfg(configFlags))
 
    # ITk
    if configFlags.Detector.EnableITkPixel:
        from InDetOverlay.ITkPixelOverlayConfig import ITkPixelOverlayCfg
        acc.merge(ITkPixelOverlayCfg(configFlags))
    if configFlags.Detector.EnableITkStrip:
        from InDetOverlay.ITkStripOverlayConfig import ITkStripOverlayCfg
        acc.merge(ITkStripOverlayCfg(configFlags))
    if configFlags.Detector.EnablePLR:
        from InDetOverlay.PLR_OverlayConfig import PLR_OverlayCfg
        acc.merge(PLR_OverlayCfg(configFlags))
 
    # HGTD
    if configFlags.Detector.EnableHGTD:
        from HGTD_Overlay.HGTD_OverlayConfig import HGTD_OverlayCfg
        acc.merge(HGTD_OverlayCfg(configFlags))
 
    # Calorimeters
    if configFlags.Detector.EnableLAr:
        from LArDigitization.LArDigitizationConfig import LArOverlayCfg, LArSuperCellOverlayCfg
        acc.merge(LArOverlayCfg(configFlags))
        if configFlags.Detector.EnableL1Calo:
            if configFlags.Overlay.DataOverlay:
                pass  # TODO: not supported for now
            else:
                from TrigT1CaloSim.TTL1OverlayConfig import LArTTL1OverlayCfg
                acc.merge(LArTTL1OverlayCfg(configFlags))
                if configFlags.GeoModel.Run >= LHCPeriod.Run3:
                    acc.merge(LArSuperCellOverlayCfg(configFlags))
 
    if configFlags.Detector.EnableTile:
        from TileSimAlgs.TileDigitizationConfig import TileDigitizationCfg, TileOverlayTriggerDigitizationCfg
        acc.merge(TileDigitizationCfg(configFlags))
        if configFlags.Detector.EnableL1Calo:
            if configFlags.Overlay.DataOverlay:
                pass  # TODO: not supported for now
            else:
                from TrigT1CaloSim.TTL1OverlayConfig import TileTTL1OverlayCfg
                acc.merge(TileTTL1OverlayCfg(configFlags))
                acc.merge(TileOverlayTriggerDigitizationCfg(configFlags))
 
    # Muon system
    from MuonConfig.MuonOverlayConfig import MuonOverlayCfg
    acc.merge(MuonOverlayCfg(configFlags))
 
    # Track overlay
    if configFlags.Overlay.doTrackOverlay:
        #need this to ensure that the ElementLinks to the PRDs are handled correctly (since the name is hardcoded in the converters)
        from TrkEventCnvTools.TrkEventCnvToolsConfig import TrkEventCnvSuperToolCfg
        acc.merge(TrkEventCnvSuperToolCfg(configFlags))
        if configFlags.Detector.GeometryITk:
           from OverlayCopyAlgs.OverlayCopyAlgsConfig import CopyITkTrackCollectionsCfg, CopyITkPixelClusterContainerCfg, CopyITkStripClusterContainerCfg
           acc.merge(CopyITkTrackCollectionsCfg(configFlags))
           acc.merge(CopyITkPixelClusterContainerCfg(configFlags))
           acc.merge(CopyITkStripClusterContainerCfg(configFlags))
        else:
           from OverlayCopyAlgs.OverlayCopyAlgsConfig import CopyTrackCollectionsCfg,CopyPixelClusterContainerCfg, CopySCT_ClusterContainerCfg,\
            CopyTRT_DriftCircleContainerCfg
           acc.merge(CopyTrackCollectionsCfg(configFlags))
           acc.merge(CopyPixelClusterContainerCfg(configFlags))
           acc.merge(CopySCT_ClusterContainerCfg(configFlags))
           acc.merge(CopyTRT_DriftCircleContainerCfg(configFlags))
 
    if configFlags.Overlay.DataOverlay:
        # Copy background vertex collection
        acc.merge(CopyBackgroundVertexCfg(configFlags))
        # Override conditions for data overlay
        if configFlags.Overlay.DataOverlayConditions:
            executeFromFragment(configFlags.Overlay.DataOverlayConditions, configFlags, acc)
 
    # Add in-file MetaData
    from xAODMetaDataCnv.InfileMetaDataConfig import SetupMetaDataForStreamCfg
    if configFlags.Output.doWriteRDO:
        acc.merge(SetupMetaDataForStreamCfg(configFlags, "RDO"))
    if configFlags.Output.doWriteRDO_SGNL:
        acc.merge(SetupMetaDataForStreamCfg(configFlags, "RDO_SGNL"))
 
    return acc