PostExec.py

Go to the documentation of this file.

# Copyright (C) 2002-2026 CERN for the benefit of the ATLAS collaboration
#
# This module contains postExec commands that can be used with the CA-based
# runHLT in athena(EF).
#
# Assumptions:
#  - the final ComponentAccumulator instance is called 'cfg'
#  - the ConfigFlags instance is called 'flags'
#
# Example usage:
#   athenaEF -C 'from TriggerJobOpts import PostExec; PostExec.foo([args])' ... TriggerJobOpts.runHLT
#   athena --postExec 'from TriggerJobOpts import PostExec; PostExec.foo([args])' TriggerJobOpts/runHLT.py
#
 
from AthenaCommon.Logging import logging
log = logging.getLogger('TriggerJobOpts.PostExec')
 
# For convenience we provide access to the CA of the calling frame.
# This is where the `exec` of the post-commmand happens in athena or athenaEF.
cfg = None
 
import inspect
__postExec_frame = next(filter(lambda f : ('TrigPSCPythonCASetup.py' in f.filename or
                                           'athenaEF.py' in f.filename or
                                           'runHLT.py' in f.filename), inspect.stack()), None)
if __postExec_frame is not None:
   __frame_members = dict(inspect.getmembers(__postExec_frame[0]))
   # in athenaEF we find the CA in the locals() otherwise globals()
   cfg =  __frame_members['f_globals'].get('cfg') or __frame_members['f_locals'].get('cfg')
 
 
#
# PostExec functions
#
 
def forceConditions(run, lb, timestamp=None, iovDbSvc=None):
   """Force all conditions (except prescales) to match the given run and LB number, a timestamp can also be provided for MC"""
 
   log.info(forceConditions.__doc__)
 
   if iovDbSvc is None:
      iovDbSvc = cfg.getService('IOVDbSvc')
 
   # Do not override these folders:
   ignore = ['/TRIGGER/HLT/PrescaleKey']   # see ATR-22143
 
   # All time-based folders (from IOVDbSvc DEBUG message, see athena!38274)
   timebased = ['/TDAQ/OLC/CALIBRATIONS',
                '/TDAQ/Resources/ATLAS/SCT/Robins',
                '/SCT/DAQ/Config/ChipSlim',
                '/SCT/DAQ/Config/Geog',
                '/SCT/DAQ/Config/MUR',
                '/SCT/DAQ/Config/Module',
                '/SCT/DAQ/Config/ROD',
                '/SCT/DAQ/Config/RODMUR',
                '/SCT/HLT/DCS/HV',
                '/SCT/HLT/DCS/MODTEMP',
                '/MUONALIGN/Onl/MDT/BARREL',
                '/MUONALIGN/Onl/MDT/ENDCAP/SIDEA',
                '/MUONALIGN/Onl/MDT/ENDCAP/SIDEC',
                '/MUONALIGN/Onl/TGC/SIDEA',
                '/MUONALIGN/Onl/TGC/SIDEC',
                '/TRIGGER/L1Calo/V1/Calibration/EfexNoiseCuts',
                '/TRIGGER/L1Calo/V1/Calibration/EfexEnergyCalib',
                '/TRIGGER/L1Calo/V1/Calibration/JfexModuleSettings',
                '/TRIGGER/L1Calo/V1/Calibration/JfexNoiseCuts',
                '/TRIGGER/L1Calo/V1/Calibration/JfexSystemSettings',
                '/TRIGGER/L1Calo/V1/Calibration/GfexModuleSettings']
 
   if timestamp is None:
      from TrigCommon.AthHLT import get_sor_params
      sor = get_sor_params(run)
      timestamp = sor['SORTime'] // int(1e9)
 
   for i,f in enumerate(iovDbSvc.Folders):
      if any(name in f for name in ignore):
         continue
      if any(name in f for name in timebased):
         iovDbSvc.Folders[i] += f'<forceTimestamp>{timestamp:d}</forceTimestamp>'
      else:
         iovDbSvc.Folders[i] += f'<forceRunNumber>{run:d}</forceRunNumber> <forceLumiblockNumber>{lb:d}</forceLumiblockNumber>'
 
 
def reverseViews():
   """Process views in reverse order"""
 
   log.info(reverseViews.__doc__)
 
   from TriggerJobOpts.TriggerConfig import collectViewMakers
   viewMakers = collectViewMakers( cfg.getSequence() )
   for alg in viewMakers:
      alg.ReverseViewsDebug = True
 
 
def dbmod_BFieldAutoConfig():  # DB modifier for debug recovery when using an online SMK
   """Use DCS currents to configure magnetic field"""
 
   log.info(dbmod_BFieldAutoConfig.__doc__)
 
   from GaudiPython.Bindings import iProperty
   # Add the DCS folder
   f = '<db>COOLOFL_DCS/CONDBR2</db> /EXT/DCS/MAGNETS/SENSORDATA'
   iProperty('IOVDbSvc').Folders += [f]
   iProperty('CondInputLoader').Load.add(('CondAttrListCollection','/EXT/DCS/MAGNETS/SENSORDATA'))
   # Configure CondAlgs
   iProperty('AtlasFieldCacheCondAlg').UseDCS = True
   iProperty('AtlasFieldMapCondAlg').LoadMapOnStart = False
   iProperty('AtlasFieldMapCondAlg').UseMapsFromCOOL = True
   iProperty('HltEventLoopMgr').setMagFieldFromPtree = False