Skip to content
Snippets Groups Projects

Update JetAnalysisCommon

Merged Pierre Antoine Delsart requested to merge delsart/athena:testAB into master
@@ -34,88 +34,18 @@ driver.submit( job, "out")
"""
import logging
from types import SimpleNamespace, ModuleType
import ROOT
#*******************************************************************
### configure logger module ----------------------------------------
logging.VERBOSE = logging.DEBUG - 1
logging.ALL = logging.DEBUG - 2
logging.addLevelName( logging.VERBOSE, 'VERBOSE' )
logging.addLevelName( logging.ALL, 'ALL' )
def _verbose(self, message, *args, **kws):
if self.isEnabledFor(logging.VERBOSE):
# Yes, logger takes its '*args' as 'args'.
self._log(logging.VERBOSE, message, args, **kws)
logging.Logger.verbose = _verbose
Logging = ModuleType("Logging")
Logging.logging = logging
from types import ModuleType
import ROOT
#*******************************************************************
class SystemOfUnits:
"""Recopied from AthenaCommon.SystemOfUnits. Only energies for now """
#
# Energy [E]
#
megaelectronvolt = 1.
electronvolt = 1.e-6*megaelectronvolt
kiloelectronvolt = 1.e-3*megaelectronvolt
gigaelectronvolt = 1.e+3*megaelectronvolt
teraelectronvolt = 1.e+6*megaelectronvolt
petaelectronvolt = 1.e+9*megaelectronvolt
#joule = electronvolt/e_SI # joule = 6.24150 e+12 * MeV
# symbols
MeV = megaelectronvolt
eV = electronvolt
keV = kiloelectronvolt
GeV = gigaelectronvolt
TeV = teraelectronvolt
PeV = petaelectronvolt
from enum import Enum
class BeamType(Enum):
"""Recopied from AthenaConfiguration.Enums."""
Collisions = 'collisions'
SingleBeam = 'singlebeam'
Cosmics = 'cosmics'
class Project(Enum):
"""Recopied from AthenaConfiguration.Enums."""
Athena = 'Athena'
AthAnalysis = 'AthAnalysis'
AthDerivation = 'AthDerivation'
AthGeneration = 'AthGeneration'
AthSimulation = 'AthSimulation'
AnalysisBase = 'AnalysisBase'
@classmethod
def determine(cls):
import os
if "AthSimulation_DIR" in os.environ:
return cls.AthSimulation
if "AthGeneration_DIR" in os.environ:
return cls.AthGeneration
if "AthAnalysis_DIR" in os.environ:
return cls.AthAnalysis
if "AthDerivation_DIR" in os.environ:
return cls.AthDerivation
if "AnalysisBase_DIR" in os.environ:
return cls.AnalysisBase
return cls.Athena
Enums_mock = ModuleType("Enum")
Enums_mock.BeamType = BeamType
Enums_mock.Project = Project
# #*******************************************************************
# reset the base logging class : the one set by xAH breaks AthenaCommon.Logging
import logging
logging.setLoggerClass( logging.Logger )
# #*******************************************************************
#*******************************************************************
def stringPropValue( value ):
"""Helper function producing a string property value"""
@@ -324,7 +254,7 @@ class ConfNameSpace:
# A replacement for CompFactory
CompFactory = ConfNameSpace()
# Add known namespaces :
CompFactory.addNameSpaces( 'Analysis', 'Trk', 'Jet', 'Sim')
CompFactory.addNameSpaces( 'Analysis', 'Trk', 'Jet', 'Sim',)
# Make a pseudo-Module :
ComponentFactory = ModuleType("ComponentFactory")
@@ -333,53 +263,6 @@ ComponentFactory.CompFactory = CompFactory
ComponentFactory.isRun3Cfg = lambda : True
#*******************************************************************
# replacements for ConfigFlags
class AthConfigFlags_mock(SimpleNamespace):
def addFlag(self, name, value ):
keys = name.split('.')
namespace = self
for k in keys[:-1]:
# build nested namespace if needed
namespace2 = getattr(namespace, k,None )
if namespace2 is None:
namespace2 = AthConfigFlags_mock()
setattr(namespace,k, namespace2 )
namespace = namespace2
# set the value on the last key
setattr(namespace,keys[-1], value)
# setup a default ConfigFlags and a few default values needed for running jets
ConfigFlags = AthConfigFlags_mock(
Input = AthConfigFlags_mock(),
)
ConfigFlags.addFlag("Beam.Type", BeamType.Collisions)
ConfigFlags.addFlag("Concurrency.NumThreads",0)
# put it in a fake module
AllConfigFlags = ModuleType("AllConfigFlags")
AllConfigFlags.ConfigFlags= ConfigFlags
AllConfigFlags.AthConfigFlags = AthConfigFlags_mock # alias for AthConfigFlags
def setupFlags(inputFiles, ):
"""Setup the ConfigFlags according to the input files content.
This is required for some part of the Athena-style config to work.
"""
ConfigFlags.Input.Files = inputFiles
f = ROOT.TFile(ConfigFlags.Input.Files[0])
tree = f.CollectionTree
ConfigFlags.Input.Collections = [br.GetName() for br in tree.GetListOfBranches() if '.' not in br.GetName()]
# terrible hack to have truth jets working :
ConfigFlags.Input.TypedCollections = []
if "GEN_EVENT" in ConfigFlags.Input.Collections:
ConfigFlags.Input.TypedCollections = ["McEventCollection#GEN_EVENT"]
return ConfigFlags
#*******************************************************************
# replacements for CFElements
def parOR(name):
@@ -422,7 +305,6 @@ ComponentAccumulator.ComponentAccumulator = ComponentAccumulatorMockUp
#*******************************************************************
#
def addManyAlgs(job, algList):
@@ -439,28 +321,22 @@ ROOT.EL.Job.addManyAlgs = addManyAlgs
# to what we have defined in this module
import sys
JetAnalysisCommon = sys.modules[__name__]
sys.modules['AthenaCommon'] = JetAnalysisCommon
sys.modules['AthenaCommon.Logging'] = JetAnalysisCommon.Logging
sys.modules['AthenaCommon.SystemOfUnits'] = JetAnalysisCommon.SystemOfUnits
sys.modules['AthenaConfiguration'] = JetAnalysisCommon
#import AthenaConfiguration, AthenaCommon
sys.modules['AthenaConfiguration.ComponentFactory'] = JetAnalysisCommon.ComponentFactory
sys.modules['AthenaConfiguration.ComponentAccumulator'] = JetAnalysisCommon.ComponentAccumulator
sys.modules['AthenaConfiguration.Enums'] = JetAnalysisCommon.Enums_mock
sys.modules['AthenaCommon.CFElements'] = JetAnalysisCommon.CFElements
sys.modules['AthenaConfiguration.AllConfigFlags'] = JetAnalysisCommon.AllConfigFlags
sys.modules['AthenaConfiguration.AthConfigFlags'] = JetAnalysisCommon.AllConfigFlags
def mock_JetRecTools():
"""Allows to ignore JetRecTools in case this package is not checked out on top of AnalysisBase"""
sys.modules['JetRecTools'] = ModuleType('JetRecTools')
sys.modules['JetRecTools.JetRecToolsConfig'] = ModuleType('JetRecToolsConfig')
#*******************************************************************
# hack specific to jets
from JetRecConfig.JetConfigFlags import createJetConfigFlags
# copy standard jet flags
ConfigFlags.Jet = createJetConfigFlags().Jet
#*******************************************************************
# hacks specific to jets
import JetRecConfig.JetRecConfig as JetRecConfig
@@ -505,3 +381,11 @@ def JetRecCfg_reorder(jetdef, configFlags, returnFinalJetDef=False):
return res
JetRecConfig.JetRecCfg = JetRecCfg_reorder
#*******************************************************************
# Some flags are not available in AnalysisBase.
# Add them manually :
from AthenaConfiguration.AllConfigFlags import ConfigFlags
ConfigFlags.addFlag('Reco.EnableTracking',True)
ConfigFlags.addFlag('Reco.EnableCombinedMuon',True)
Loading