OverlaySkeleton.py

Go to the documentation of this file.

# Copyright (C) 2002-2026 CERN for the benefit of the ATLAS collaboration
 
import sys
 
from AthenaConfiguration.AutoConfigFlags import GetFileMD
from AthenaConfiguration.Enums import ProductionStep
from PyJobTransforms.CommonRunArgsToFlags import commonRunArgsToFlags
from PyJobTransforms.TransformUtils import processPreExec, processPreInclude, processPostExec, processPostInclude
from SimuJobTransforms.CommonSimulationSteering import specialConfigPreInclude, specialConfigPostInclude
 
# force no legacy job properties
from AthenaCommon import JobProperties
JobProperties.jobPropertiesDisallowed = True
 
 
def setOverlayInputFiles(runArgs, flags, log):
    hasRDO_BKGInput = hasattr(runArgs, 'inputRDO_BKGFile')
    hasEVNT_Input   = hasattr(runArgs, 'inputEVNTFile')
 
    if flags.Common.ProductionStep == ProductionStep.Overlay and not hasattr(runArgs, 'inputHITSFile'):
        raise RuntimeError('No input HITS file defined')
 
    if not hasRDO_BKGInput:
        raise RuntimeError('No input RDO_BKG file defined')
 
    if hasattr(runArgs, 'skipSecondaryEvents'):
        flags.Overlay.SkipSecondaryEvents = runArgs.skipSecondaryEvents
 
    flags.Input.Files = runArgs.inputRDO_BKGFile
 
    if flags.Input.isMC:
        log.info('Running MC+MC overlay')
        flags.Overlay.DataOverlay = False
    else:
        log.info('Running MC+data overlay')
        flags.Overlay.DataOverlay = True
 
    if flags.Common.ProductionStep == ProductionStep.Overlay:
        flags.Input.SecondaryFiles = runArgs.inputHITSFile
    elif flags.Common.ProductionStep == ProductionStep.FastChain:
        if not hasEVNT_Input:
            raise RuntimeError('No input EVNT file defined')
        else:
            flags.Input.SecondaryFiles = runArgs.inputEVNTFile
    else:
        raise RuntimeError('No secondaryFiles are defined')
 
    # take MCChannelNumber from secondary input:
    flags.Input.MCChannelNumber = GetFileMD(flags.Input.SecondaryFiles, allowEmpty=False).get("mc_channel_number", 0)
 
    # runNumber is MC channel number in reco
    if hasattr(runArgs, 'runNumber'):
        if flags.Input.MCChannelNumber != runArgs.runNumber:
            log.warning('Got different MC channel number (%d) from runNumber than from metadata (%d)', runArgs.runNumber, flags.Input.MCChannelNumber)
            flags.Input.MCChannelNumber = runArgs.runNumber
        else:
            log.info('MC channel number: %d', flags.Input.MCChannelNumber)
 
 
def fromRunArgs(runArgs):
    from AthenaCommon.Logging import logging
    logOverlay = logging.getLogger('Overlay')
    logOverlay.info('****************** STARTING OVERLAY *****************')
 
    logOverlay.info('**** Transformation run arguments')
    logOverlay.info(str(runArgs))
 
    logOverlay.info('**** Setting-up configuration flags')
    from AthenaConfiguration.AllConfigFlags import initConfigFlags
    flags = initConfigFlags()
 
    commonRunArgsToFlags(runArgs, flags)
 
    flags.Common.ProductionStep = ProductionStep.Overlay
 
    # Setting input files for Overlay
    setOverlayInputFiles(runArgs, flags, logOverlay)
 
    # Setting output files for Overlay
    if hasattr(runArgs, 'outputRDOFile'):
        if runArgs.outputRDOFile == 'None':
            flags.Output.RDOFileName = ''
        else:
            flags.Output.RDOFileName = runArgs.outputRDOFile
    else:
        raise RuntimeError('No output RDO file defined')
 
    if hasattr(runArgs, 'outputRDO_SGNLFile'):
        flags.Output.RDO_SGNLFileName = runArgs.outputRDO_SGNLFile
 
    # Autoconfigure enabled subdetectors
    if hasattr(runArgs, 'detectors'):
        detectors = runArgs.detectors
    else:
        detectors = None
 
    # Setup digitization flags
    from DigitizationConfig.DigitizationConfigFlags import digitizationRunArgsToFlags
    digitizationRunArgsToFlags(runArgs, flags)
 
    # Setup detector flags
    from AthenaConfiguration.DetectorConfigFlags import setupDetectorFlags
    setupDetectorFlags(flags, detectors, use_metadata=True, toggle_geometry=True)
 
    # Disable LVL1 trigger if triggerConfig explicitly set to 'NONE'
    if hasattr(runArgs, 'triggerConfig') and runArgs.triggerConfig == 'NONE':
        flags.Detector.EnableL1Calo = False
 
    # Setup perfmon flags from runargs
    from PerfMonComps.PerfMonConfigHelpers import setPerfmonFlagsFromRunArgs
    setPerfmonFlagsFromRunArgs(flags, runArgs)
 
    # Special Configuration preInclude
    specialConfigPreInclude(flags)
 
    # Pre-include
    processPreInclude(runArgs, flags)
 
    # Pre-exec
    processPreExec(runArgs, flags)
 
    # To respect --athenaopts
    flags.fillFromArgs()
 
    # Moving here so that it is ahead of flags being locked. Need to
    # iterate on exact best position w.r.t. above calls
    from OverlayConfiguration.OverlayMetadata import overlayMetadataCheck
    # Handle metadata correctly
    overlayMetadataCheck(flags)
 
    # Lock flags
    flags.lock()
 
    # Main overlay steering
    from OverlayConfiguration.OverlaySteering import OverlayMainCfg
    cfg = OverlayMainCfg(flags)
 
    # Special message service configuration
    from DigitizationConfig.DigitizationSteering import DigitizationMessageSvcCfg
    cfg.merge(DigitizationMessageSvcCfg(flags))
 
    # Special Configuration postInclude
    specialConfigPostInclude(flags, cfg)
 
    # Post-include
    processPostInclude(runArgs, flags, cfg)
 
    # Post-exec
    processPostExec(runArgs, flags, cfg)
 
    # Write AMI tag into in-file metadata
    from PyUtils.AMITagHelperConfig import AMITagCfg
    cfg.merge(AMITagCfg(flags, runArgs))
 
    # Run the final configuration
    sc = cfg.run()
    sys.exit(not sc.isSuccess())