Skip to content
Snippets Groups Projects
Commit 456c9f38 authored by Ahmed Tarek's avatar Ahmed Tarek
Browse files

Merge branch 'IDPVM_legacy' into '24.0'

IDPVM legacy config clean up

See merge request !69019
parents 4e3a79e0 12bd32e1
No related branches found
No related tags found
2 merge requests!691252024-02-22: merge of 24.0 into main,!69019IDPVM legacy config clean up
# Copyright (C) 2002-2021 CERN for the benefit of the ATLAS collaboration
from InDetRecExample.TrackingCommon import setDefaults
def getHistogramDefinitionSvc(**kwargs):
import InDetPhysValMonitoring.InDetPhysValMonitoringConf
kwargs = setDefaults(kwargs,
DefinitionSource="InDetPVMPlotDefRun2.xml",
DefinitionFormat="text/xml")
return InDetPhysValMonitoring.InDetPhysValMonitoringConf.HistogramDefinitionSvc(**kwargs)
# Copyright (C) 2002-2022 CERN for the benefit of the ATLAS collaboration
from __future__ import print_function
import InDetPhysValMonitoring.InDetPhysValMonitoringConf
from .ConfigUtils import createPublicTool
from .ConfigUtils import createExtendNameIfNotDefault
from .ConfigUtils import toolFactory
from InDetRecExample.TrackingCommon import setDefaults
# ---- definitions
def metaDataKey():
'''
Meta data key to store the file source on which the InDet decoration alg has been running.
'''
return 'InDetPhysValDecoration'
def monManName():
'''
Name of the monitoring manager
'''
return 'PhysValMonManager'
def findAlg(alg_name, search_outputstream_otherwise=True):
'''
Find the algorithm alg_name in the top sequence and return its index.
if no algorithm of the given name is found the index just before the AhtenaOutputStream is returned.
if all that fails None is returned.
'''
if not isinstance(alg_name, (list, tuple)):
raise Exception(
'logic error findAlg called with a non list argument %s / %s' % (alg_name, type(alg_name)))
from AthenaCommon.AlgSequence import AlgSequence
topSequence = AlgSequence()
count = 0
mon_man_index = None
for child in topSequence.getChildren():
# if child.getName() in alg_name :
if child.getName() in alg_name:
# print ('DEBUG findAlg %s =?= %s ' %(child.getName(),alg_name))
mon_man_index = count
if len(alg_name) == 1:
break
count += 1
if mon_man_index is None and search_outputstream_otherwise:
count = 0
exclude_streams = ['StreamBS']
for child in topSequence.getChildren():
if child.getType() == 'AthenaOutputStream' and child.getName() not in exclude_streams:
# print ('DEBUG found %s at postion %i/%i' % (child.getFullName(),count,len(topSequence.getChildren() ) ))
mon_man_index = count
break
count += 1
return mon_man_index
def findMonMan():
'''
Find the monitoring manager and return its index in the top sequence or None
'''
return findAlg([monManName()], search_outputstream_otherwise=False)
def setMetaData():
'''
Write the input source on which the InDet decoration algorithm was running to the meta data
'''
from AthenaCommon.AppMgr import ServiceMgr as svcMgr
if metaDataKey() in svcMgr.TagInfoMgr.ExtraTagValuePairs:
return
from RecExConfig.RecFlags import rec
str = ''
if rec.readRDO():
str = 'fromRDO'
elif rec.readESD():
str = 'fromESD'
elif rec.readAOD():
str = 'fromAOD'
if metaDataKey() not in svcMgr.TagInfoMgr.ExtraTagValuePairs:
svcMgr.TagInfoMgr.ExtraTagValuePairs[metaDataKey()] = str
# print ('DEBUG add meta data %s.' % svcMgr.TagInfoMgr.ExtraTagValuePairs)
def getPhysValMonInDetHoleSearchTool(**kwargs):
from InDetRecExample.TrackingCommon import getInDetHoleSearchTool
kwargs = setDefaults(kwargs,
name='PhysValMonInDetHoleSearchTool')
return getInDetHoleSearchTool(**kwargs)
def getInDetPhysHitDecoratorAlg(**kwargs):
'''
create decoration algorithm which decorates track particles with the unbiased hit residuals and pulls.
If the collection name TrackParticleContainerName is specified and differs from the default, the name
of the algorithm will be extended by the collection name
'''
# @TODO use track particles from ? from InDetRecExample.InDetKeys import InDetKeys
from InDetRecExample.TrackingCommon import getPixelLorentzAngleTool
return createExtendNameIfNotDefault(InDetPhysValMonitoring.InDetPhysValMonitoringConf.InDetPhysHitDecoratorAlg,
'TrackParticleContainerName', 'InDetTrackParticles',
kwargs,
InDetTrackHoleSearchTool=toolFactory(
getPhysValMonInDetHoleSearchTool),
Updator='Trk::KalmanUpdator/TrkKalmanUpdator',
ResidualPullCalculator='Trk::ResidualPullCalculator/ResidualPullCalculator',
LorentzAngleTool=getPixelLorentzAngleTool(),
TrackParticleContainerName='InDetTrackParticles')
def getParameterErrDecoratorAlg(**kwargs):
'''
create decoration algorithm which decorates track particles with the uncertainties of the track parameters.
If the collection name TrackParticleContainerName is specified and differs from the default, the name
of the algorithm will be extended by the collection name
'''
return createExtendNameIfNotDefault(InDetPhysValMonitoring.InDetPhysValMonitoringConf.ParameterErrDecoratorAlg,
'TrackParticleContainerName', 'InDetTrackParticles',
kwargs)
def getInDetPhysValTruthDecoratorAlg(**kwargs):
'''
create decoration algorithm which decorates truth particles with track parameters at the perigee.
If the collection name TruthParticleContainerName is specified and differs from the default, the name
of the algorithm will be extended by the collection name
'''
return createExtendNameIfNotDefault(
InDetPhysValMonitoring.InDetPhysValMonitoringConf.InDetPhysValTruthDecoratorAlg,
'TruthParticleContainerName', 'TruthParticles',
kwargs,
Extrapolator='Trk::Extrapolator/AtlasExtrapolator')
def getInDetRttTruthSelectionTool(**kwargs):
return createPublicTool(
InDetPhysValMonitoring.InDetPhysValMonitoringConf.AthTruthSelectionTool,
kwargs,
# @TODO change name ? name = 'InDetRttTruthSelectionTool',
requireStatus1=True,
requireCharged=True,
requireSiHit=0,
requireOnlyPrimary=(
True if kwargs.pop("OnlyDressPrimaryTracks", True) else False),
maxProdVertRadius=300.,
maxEta=2.5,
minPt=500.)
def getInDetTruthSelectionTool(**kwargs):
return getInDetRttTruthSelectionTool(
**setDefaults(kwargs,
name='InDetTruthSelectionTool',
maxProdVertRadius=-1)) # disable production radius cut
def getHardScatterSelectionTool(**kwargs):
from InDetHardScatterSelectionTool.InDetHardScatterSelectionToolConf import InDet__InDetHardScatterSelectionTool
kwargs = setDefaults(kwargs, name="InDetHardScatterSelectionTool")
kwargs = setDefaults(kwargs, JetContainer="AntiKt4EMTopoJets")
return InDet__InDetHardScatterSelectionTool(**kwargs)
def getInDetPhysValTruthDecoratorAlgStableParticles(**kwargs):
'''
Create a decoration algorithm which decorates stable truth particles only with track parameters at the perigee
see @ref getInDetPhysValTruthDecoratorAlg for details.
'''
return getInDetPhysValTruthDecoratorAlg(**setDefaults(
kwargs,
TruthSelectionTool=getInDetTruthSelectionTool()))
def getTruthClassDecoratorAlg(**kwargs):
'''
create decoration algorithm which decorates truth particles with origin and type from truth classifier.
if the collection name TruthParticleContainerName is specified and differs from the default, the name
of the algorithm will be extended by the collection name
'''
return createExtendNameIfNotDefault(
InDetPhysValMonitoring.InDetPhysValMonitoringConf.TruthClassDecoratorAlg,
'TruthParticleContainerName', 'TruthParticles',
kwargs)
def getTrackDecorators(**kwargs):
'''
Get track particle decorators needed for the InDetPhysValMonitoring tool
If the collection name TrackParticleContainerName is specified and differs from the default, the name
of the algorithms will be extended by the collection name.
'''
# only valid kwarg : TrackParticleContainerName
from RecExConfig.AutoConfiguration import IsInInputFile
if not IsInInputFile('TrackCollection', 'CombinedInDetTracks'):
return [getParameterErrDecoratorAlg(**kwargs)]
else:
return [getInDetPhysHitDecoratorAlg(**kwargs),
getParameterErrDecoratorAlg(**kwargs)]
def getTruthDecorators(**kwargs):
'''
Get truth particle decorators needed for the InDetPhysValMonitoring tool.
If the collection name TruthParticleContainerName is specified and differs from the default, the name
of the algorithms will be extended by the collection name.
'''
# only valid kwarg : TruthParticleContainerName
return [getInDetPhysValTruthDecoratorAlg(**kwargs)
# , getTruthClassDecoratorAlg(**kwargs)
]
def getDecorators(**kwargs):
decorators = []
import InDetPhysValMonitoring.InDetPhysValDecoration
decorators += InDetPhysValMonitoring.InDetPhysValDecoration.getTrackDecorators(
TrackParticleContainerName=kwargs.pop('TrackParticleContainerName', 'InDetTrackParticles'))
truth_particle_container_name = kwargs.pop(
'TruthParticleContainerName', 'TruthParticles')
from InDetPhysValMonitoring.InDetPhysValJobProperties import isMC
if isMC():
decorators += InDetPhysValMonitoring.InDetPhysValDecoration.getTruthDecorators(
TruthParticleContainerName=truth_particle_container_name)
if len(kwargs) > 0:
raise Exception('Unused kwargs: %s' % kwargs)
return decorators
class InDetPhysValKeys:
'''
Keys to decorate and monitor GSF Tracks/TrackParticles
'''
# the name of unslimmed GSF Tracks defined here
GSFTracksUnslimmed = 'GSFTracksUnslimmed'
# the name of GSF Tracks given by egamma
GSFTracks = 'GSFTracks'
# the name of GSF TrackParticles given by egamma
# @todo should be used also in InDetPhysValMonitoring.py
GSFTrackParticles = 'GSFTrackParticles'
def getGSFTrackDecorators(**kwargs):
return getTrackDecorators(
TrackParticleContainerName=InDetPhysValKeys.GSFTrackParticles)
# Debugging
# import trace
# tracer = trace.Trace()
# tracer.runfunc( )
def _addDecorators(decorator_alg_list, add_after=None):
'''
Add the given decorator algorithms to the top sequence.
The algorithm is to be run on RAW/RDO since it depends on full hit information
which is generally not available at later stages. The decorations added by this
algorithm are used by InDetPhysValMonitoring tool.
'''
if add_after is not None and not isinstance(add_after, (list, tuple)):
raise Exception(' logic error _addDecorators called with a non list argument %s / %s' %
(add_after, type(add_after)))
# Access the algorithm sequence:
from AthenaCommon.AlgSequence import AlgSequence
topSequence = AlgSequence()
# print ('DEBUG add _addDecorators add after %s ' % (add_after))
# if there is a monitoring manager add decorator before
mon_man_index = findMonMan()
if mon_man_index is None and add_after is not None:
alg_index = findAlg(add_after)
if alg_index is not None:
# add after the found algorithm
mon_man_index = alg_index + 1
# print ('DEBUG _addDecorators after this %s ' % (mon_man_index))
if mon_man_index is None:
for decorator_alg in decorator_alg_list:
if findAlg([decorator_alg.getName()], search_outputstream_otherwise=False) is not None:
print('DEBUG decorator %s already in sequence. Not adding again.' % (
decorator_alg.getFullName()))
continue
print('DEBUG add decorator %s at end of top sequence:' %
(decorator_alg.getFullName()))
topSequence += decorator_alg
else:
for decorator_alg in decorator_alg_list:
if findAlg([decorator_alg.getName()], search_outputstream_otherwise=False) is not None:
print('DEBUG decorator %s already in sequence. Not inserting again.' % (
decorator_alg.getFullName()))
continue
print('DEBUG insert decorator %s at position %i' %
(decorator_alg.getFullName(), mon_man_index))
topSequence.insert(mon_man_index, decorator_alg)
mon_man_index += 1
setMetaData()
def addGSFTrackDecoratorAlg():
'''
Search egamma algorithm and add the GSF TrackParticle decorator after the it.
'''
from InDetPhysValMonitoring.InDetPhysValDecoration import _addDecorators
from InDetPhysValMonitoring.InDetPhysValJobProperties import InDetPhysValFlags
if InDetPhysValFlags.doValidateGSFTracks():
# print ('DEBUG add addGSFTrackDecoratorAlg')
decorators = getGSFTrackDecorators()
from InDetPhysValMonitoring.InDetPhysValJobProperties import InDetPhysValFlags
from InDetPhysValMonitoring.ConfigUtils import extractCollectionPrefix
for col in InDetPhysValFlags.validateExtraTrackCollections():
prefix = extractCollectionPrefix(col)
decorators += getTrackDecorators(
TrackParticleContainerName=prefix+"TrackParticles")
# add the InDetPhysValDecoratorAlgGSF after the egamma algorithms ran
# they build the GSF track particles.
_addDecorators(decorators, ['egamma', 'egammaTruthAssociationAlg'])
# To allow for proper decoration of GSF TrackParticles
# - have to switch of slimming for GSF Tracks at time of creation
# - must slim GSF Tracks after decoration since the unslimmed GSF Tracks cannot be persistified
from AthenaCommon.AppMgr import ToolSvc
# print (ToolSvc)
# print ('DEBUG has EMBremCollectionBuilder %s' % hasattr(ToolSvc,'EMBremCollectionBuilder'))
if hasattr(ToolSvc, 'EMBremCollectionBuilder'):
decor_index = findAlg(
[decorators[0].getName()], search_outputstream_otherwise=False)
if decor_index is not None:
from TrkTrackSlimmer.TrkTrackSlimmerConf import Trk__TrackSlimmer as ConfigurableTrackSlimmer
slimmer = ConfigurableTrackSlimmer(
name="RealGSFTrackSlimmer",
TrackLocation=[
InDetPhysValKeys.GSFTracksUnslimmed],
SlimmedTrackLocation=[
InDetPhysValKeys.GSFTracks],
TrackSlimmingTool=ToolSvc.EMBremCollectionBuilder.TrackSlimmingTool)
from InDetPhysValMonitoring.InDetPhysValMonitoringConf import DummyTrackSlimmingTool
slimming_tool = DummyTrackSlimmingTool()
ToolSvc += slimming_tool
ToolSvc.EMBremCollectionBuilder.TrackSlimmingTool = slimming_tool
ToolSvc.EMBremCollectionBuilder.OutputTrackContainerName = InDetPhysValKeys.GSFTracksUnslimmed
# ToolSvc.ResidualPullCalculator.OutputLevel = 1
from AthenaCommon.AlgSequence import AlgSequence
topSequence = AlgSequence()
topSequence.insert(decor_index+1, slimmer)
# import sys
# sys.exit(1)
def addDecorator():
'''
Add the track particle decoration algorithm to the top sequence.
The algorithm is to be run on RAW/RDO since it depends on full hit information
which is generally not available at later stages. The decorations added by this
algorithm are used by InDetPhysValMonitoring tool.
'''
decorators = []
decorators += getTrackDecorators()
from InDetPhysValMonitoring.InDetPhysValJobProperties import isMC
if isMC():
decorators += getTruthDecorators()
from InDetPhysValMonitoring.InDetPhysValJobProperties import InDetPhysValFlags
InDetPhysValFlags.init()
if InDetPhysValFlags.doValidateGSFTracks():
# cannot add the decorator for GSF tracks at this moment because the egamma algorithm has not been
# constructed yet.
# so this does not work: decorators.append( InDetPhysValDecoratorAlg.InDetPhysValDecoratorAlgGSF() )
# Can only schedule a user algorithm :
from RecExConfig.RecFlags import rec
rec.UserExecs += ['from InDetPhysValMonitoring.InDetPhysValDecoration import addGSFTrackDecoratorAlg;addGSFTrackDecoratorAlg();']
_addDecorators(decorators)
def canAddDecorator():
'''
check whether the decorator can be added.
A decorator can be added if a track particle converter alg is in the sequence or
if ESDs or AODs are read.
'''
from RecExConfig.RecFlags import rec
if not rec.doInDet:
return False
if rec.readAOD:
return True
if rec.readESD:
return True
if rec.readTAG:
return False
if rec.readRDO:
try:
from AthenaCommon.AlgSequence import AlgSequence
topSequence = AlgSequence()
import re
pat = re.compile('.*(TrackParticleCnvAlg).*')
for alg in topSequence.getChildren():
if pat.match(alg.getFullName()) is not None:
return True
except Exception:
pass
return False
def addDecoratorIfNeeded():
'''
Run the InDet decoration algorithm if it has not been ran yet.
'''
if not canAddDecorator():
print('DEBUG addDecoratorIfNeeded ? Stage is too early or too late for running the decoration. Needs reconstructed tracks. Try again during next stage ?')
return
addDecorator()
# if GSF tracks need to be monitored schedule addExtraMonitoring as user algorithm, so that
# the monitoring manager exists already.
from InDetPhysValMonitoring.InDetPhysValJobProperties import InDetPhysValFlags
InDetPhysValFlags.init()
# from AthenaCommon.AppMgr import ServiceMgr as svcMgr
# print ('DEBUG addDecoratorIfNeeded add meta data %s.' % svcMgr.TagInfoMgr.ExtraTagValuePairs)
# Copyright (C) 2002-2020 CERN for the benefit of the ATLAS collaboration
#
# purpose Python module to hold common flags to configure the InDetPhysValMonitoring
##
from __future__ import print_function
from InDetRecExample.InDetJobProperties import Enabled
from AthenaCommon.JobProperties import jobproperties
from AthenaCommon.JobProperties import JobProperty, JobPropertyContainer
""" InDetPhysValJobProperties
Python module to hold common flags to configure InDetPhysValMonitoring JobOptions.
"""
__doc__ = "InDetPhysValJobProperties"
__all__ = ["InDetPhysValJobProperties"]
# import AthenaCommon.SystemOfUnits as Units
def isMC():
'''
Test whether the input is monte carlo.
@return true if this is simulation and should have truth information
'''
from AthenaCommon.GlobalFlags import globalflags
return globalflags.DataSource() != 'data'
class InDetPhysValFlagsJobProperty(JobProperty):
""" This class stores if a user changed a property once in the variable setByUser
"""
setByUser = False
def _do_action(self):
self.setByUser = True
def _undo_action(self):
self.setByUser = True
def get_Value(self):
if (self.allowedTypes[0] == 'bool'):
return self.statusOn and self.StoredValue
else:
return self.StoredValue
class doValidateLargeD0Tracks(InDetPhysValFlagsJobProperty):
statusOn = True
allowedTypes = ['bool']
StoredValue = False
class doValidateMergedLargeD0Tracks(InDetPhysValFlagsJobProperty):
statusOn = True
allowedTypes = ['bool']
StoredValue = False
class doRecoOnly(InDetPhysValFlagsJobProperty):
statusOn = True
allowedTypes = ['bool']
StoredValue = False
class doValidateGSFTracks(InDetPhysValFlagsJobProperty):
statusOn = True
allowedTypes = ['bool']
StoredValue = False
class doValidateMuonMatchedTracks(InDetPhysValFlagsJobProperty):
statusOn = True
allowedTypes = ['bool']
StoredValue = False
class doValidateElectronMatchedTracks(InDetPhysValFlagsJobProperty):
statusOn = True
allowedTypes = ['bool']
StoredValue = False
class doValidateLooseTracks(InDetPhysValFlagsJobProperty):
statusOn = True
allowedTypes = ['bool']
StoredValue = False
class doValidateTightPrimaryTracks(InDetPhysValFlagsJobProperty):
statusOn = True
allowedTypes = ['bool']
StoredValue = False
class doValidateTracksInJets(InDetPhysValFlagsJobProperty):
statusOn = True
allowedTypes = ['bool']
StoredValue = False
class doValidateTracksInBJets(InDetPhysValFlagsJobProperty):
statusOn = True
allowedTypes = ['bool']
StoredValue = False
class doValidateTruthToRecoNtuple(InDetPhysValFlagsJobProperty):
statusOn = True
allowedTypes = ['bool']
StoredValue = False
class doTruthOriginPlots(InDetPhysValFlagsJobProperty):
statusOn = True
allowedTypes = ['bool']
StoredValue = False
class doPerAuthorPlots(InDetPhysValFlagsJobProperty):
statusOn = True
allowedTypes = ['bool']
StoredValue = False
class doHitLevelPlots(InDetPhysValFlagsJobProperty):
statusOn = True
allowedTypes = ['bool']
StoredValue = False
class validateExtraTrackCollections(InDetPhysValFlagsJobProperty):
"""List of extra track collection names to be validated in addition to Tracks."""
statusOn = True
allowedTypes = ['list']
StoredValue = []
class doPhysValOutput(InDetPhysValFlagsJobProperty):
statusOn = True
allowedTypes = ['bool']
StoredValue = False
class doExpertOutput(InDetPhysValFlagsJobProperty):
statusOn = True
allowedTypes = ['bool']
StoredValue = False
class setTruthStrategy(InDetPhysValFlagsJobProperty):
statusOn = True
allowedTypes = ['string']
StoredValue = 'HardScatter'
class ancestorIDs(InDetPhysValFlagsJobProperty):
statusOn = True
allowedTypes = ['list']
StoredValue = []
class requiredSiHits(InDetPhysValFlagsJobProperty):
statusOn = True
allowedTypes = ['int']
StoredValue = 0
class maxProdVertRadius(InDetPhysValFlagsJobProperty):
statusOn = True
allowedTypes = ['float']
StoredValue = 300
class hardScatterStrategy(InDetPhysValFlagsJobProperty):
"""The hard-scatter vertex selection strategy to use when running hard-scatter efficiency / performance plots in IDPVM. 0 corresponds to sumPt^2, 1 corresponds to sumPt"""
statusOn = True
allowedTypes = ['int']
StoredValue = 0 # default to sum(pt²)
class doIDTIDE(InDetPhysValFlagsJobProperty):
statusOn = True
allowedTypes = ['bool']
StoredValue = False
class doTechnicalEfficiency(InDetPhysValFlagsJobProperty):
''' enables the filling of the technical efficiency '''
statusOn = True
allowedTypes = ['bool']
StoredValue = False
# -----------------------------------------------------------------------------
# 2nd step
# Definition of the InDet flag container
class InDetPhysValJobProperties(JobPropertyContainer):
"""Container for the InDetPhysValMonitoring flags
"""
def checkThenSet(self, jp, value):
# checks if a variable has been changed by the user before
if not jp.setByUser:
jp.set_Value(value)
def setupDefaults(self):
pass
def init(self):
# Method to do the final setup of the flags according to user input before.
# This method MUST ONLY BE CALLED once in InDetRecExample/InDetRec_jobOptions.py!!
if not self.Enabled:
print('InDetPhysValFlags.init(): ID PhysValFlags are disabled. Locking container and not doing anything else.')
else:
self.setup()
# do this also if Enabled == False
print("Initialization of InDetFlags finished - locking container!")
self.lock_JobProperties()
def setup(self):
print('Initializing InDetJobPhysValProperties with InDetFlags.')
# THIS METHOD MUST BE THE FIRST TO BE CALLED. DO NOT MOVE IT OR ADD THINGS IN FRONT
self.setupDefaults()
print(self)
def printInfo(self):
pass
jobproperties.add_Container(InDetPhysValJobProperties)
# -----------------------------------------------------------------------------
# 4th step
# adding ID flags to the InDetJobProperties container
_list_InDetPhysValJobProperties = [
Enabled,
doValidateGSFTracks,
doValidateLooseTracks,
doValidateTightPrimaryTracks,
doValidateTracksInJets,
doValidateTracksInBJets,
doValidateTruthToRecoNtuple,
validateExtraTrackCollections,
doValidateMuonMatchedTracks,
doValidateElectronMatchedTracks,
doPhysValOutput,
doExpertOutput,
setTruthStrategy,
doValidateLargeD0Tracks,
doValidateMergedLargeD0Tracks,
doRecoOnly,
doTruthOriginPlots,
doPerAuthorPlots,
doHitLevelPlots,
ancestorIDs,
requiredSiHits,
hardScatterStrategy,
doIDTIDE,
doTechnicalEfficiency
]
for j in _list_InDetPhysValJobProperties:
jobproperties.InDetPhysValJobProperties.add_JobProperty(j)
InDetPhysValFlags = jobproperties.InDetPhysValJobProperties
print('DEBUG InDetPhysValJobProberties')
print(InDetPhysValFlags)
# Copyright (C) 2002-2024 CERN for the benefit of the ATLAS collaboration
from __future__ import print_function
from InDetPhysValMonitoring.ConfigUtils import serviceFactory, toolFactory
from InDetRecExample.TrackingCommon import setDefaults
import InDetPhysValMonitoring.InDetPhysValMonitoringConf
def removePhysValExample():
print('DEBUG no AntiKt4EMTopoJets in input file.')
from InDetPhysValMonitoring.InDetPhysValDecoration import findMonMan
mon_index = findMonMan()
if mon_index is not None:
import re
pattern = re.compile('.*PhysValExample')
from AthenaCommon.AlgSequence import AlgSequence
topSequence = AlgSequence()
mon_manager = topSequence.getChildren()[mon_index]
print('DEBUG Found mon_manager %s with the following tools: %s ' % (
mon_manager.getName(), mon_manager.AthenaMonTools.toStringProperty()))
for idx in range(0, len(mon_manager.AthenaMonTools)):
tool_name = mon_manager.AthenaMonTools[idx].getName()
# print ('DEBUG %s AthenaMonTools %s' % (mon_manager.getName(),tool_name))
if pattern.match(tool_name) is not None:
print('DEBUG removing %s from %s .' %
(tool_name, mon_manager.getName()))
del mon_manager.AthenaMonTools[idx]
break
else:
for child in topSequence.getChildren():
print('DEBUG top sequence has %s' % (child.getName()))
def getInDetPhysValMonitoringTool(**kwargs):
kwargs = setDefaults(
kwargs,
useTrackSelection=False,
EnableLumi=False)
# create the HistogramDefinitionSvc
# at the moment there can only be one HistogramDefinitionSvc
from InDetPhysValMonitoring.HistogramDefinitionSvc import getHistogramDefinitionSvc
# self.HistogramDefinitionSvc =
serviceFactory(getHistogramDefinitionSvc)
from InDetPhysValMonitoring.InDetPhysValJobProperties import isMC, InDetPhysValFlags
if isMC():
from InDetPhysValMonitoring.InDetPhysValDecoration import getInDetRttTruthSelectionTool
from InDetPhysValMonitoring.InDetPhysValDecoration import getHardScatterSelectionTool
kwargs = setDefaults(
kwargs, TruthParticleContainerName="TruthParticles")
if 'TruthSelectionTool' not in kwargs:
kwargs = setDefaults(
kwargs, TruthSelectionTool=getInDetRttTruthSelectionTool())
if 'hardScatterSelectionTool' not in kwargs:
kwargs = setDefaults(
kwargs, hardScatterSelectionTool=getHardScatterSelectionTool(RedoHardScatter=True, SelectionMode=InDetPhysValFlags.hardScatterStrategy()))
if InDetPhysValFlags.doValidateTracksInJets():
jets_name = 'AntiKt4LCTopoJets'
kwargs = setDefaults(kwargs,
JetContainerName=jets_name,
FillTrackInJetPlots=True)
from InDetPhysValMonitoring.addTruthJets import addTruthJetsIfNotExising
addTruthJetsIfNotExising(jets_name)
if InDetPhysValFlags.doValidateTracksInBJets():
kwargs = setDefaults(
kwargs,
FillTrackInBJetPlots=True)
else:
kwargs = setDefaults(
kwargs,
JetContainerName='',
FillTrackInJetPlots=False)
if InDetPhysValFlags.doValidateTruthToRecoNtuple():
kwargs = setDefaults(
kwargs,
FillTruthToRecoNtuple=True)
if InDetPhysValFlags.doTruthOriginPlots():
kwargs = setDefaults(
kwargs,
doTruthOriginPlots=True )
if InDetPhysValFlags.doPerAuthorPlots():
kwargs = setDefaults(
kwargs,
doPerAuthorPlots=True )
if InDetPhysValFlags.doHitLevelPlots():
kwargs = setDefaults(
kwargs,
doHitLevelPlots=True )
# adding the VeretxTruthMatchingTool
from InDetTruthVertexValidation.InDetTruthVertexValidationConf import InDetVertexTruthMatchTool
kwargs = setDefaults(
kwargs,
useVertexTruthMatchTool=True,
VertexTruthMatchTool=toolFactory(InDetVertexTruthMatchTool))
# Options for Truth Strategy : Requires full pile-up truth containers for some
if InDetPhysValFlags.setTruthStrategy() == 'All' or InDetPhysValFlags.setTruthStrategy() == 'PileUp':
from RecExConfig.AutoConfiguration import IsInInputFile
if IsInInputFile('xAOD::TruthPileupEventContainer', 'TruthPileupEvents'):
kwargs = setDefaults(
kwargs,
PileupSwitch=InDetPhysValFlags.setTruthStrategy())
else:
print('WARNING Truth Strategy for InDetPhysValMonitoring set to %s but TruthPileupEvents are missing in the input; resetting to HardScatter only' % (
InDetPhysValFlags.setTruthStrategy()))
elif InDetPhysValFlags.setTruthStrategy() != 'HardScatter':
print('WARNING Truth Strategy for for InDetPhysValMonitoring set to invalid option %s; valid flags are ["HardScatter", "All", "PileUp"]' % (
InDetPhysValFlags.setTruthStrategy()))
else:
# disable truth monitoring for data
kwargs = setDefaults(
kwargs,
TruthParticleContainerName='',
TruthVertexContainerName='',
TruthEvents='',
TruthPileupEvents='',
TruthSelectionTool='',
# the jet container is actually meant to be a truth jet container
JetContainerName='',
FillTrackInJetPlots=False,
FillTrackInBJetPlots=False,
FillTruthToRecoNtuple=False)
# Control the number of output histograms
if InDetPhysValFlags.doPhysValOutput():
kwargs = setDefaults(kwargs,
DetailLevel=100)
elif InDetPhysValFlags.doExpertOutput():
kwargs = setDefaults(kwargs,
DetailLevel=200)
# hack to remove example physval monitor
from RecExConfig.AutoConfiguration import IsInInputFile
if not IsInInputFile('xAOD::JetContainer', 'AntiKt4EMTopoJets'):
add_remover = True
from RecExConfig.RecFlags import rec
try:
for elm in rec.UserExecs:
if elm.find('removePhysValExample') > 0:
add_remover = False
break
except Exception:
pass
if add_remover:
rec.UserExecs += ['from InDetPhysValMonitoring.InDetPhysValMonitoringTool import removePhysValExample;removePhysValExample();']
return InDetPhysValMonitoring.InDetPhysValMonitoringConf.InDetPhysValMonitoringTool(**kwargs)
def getInDetPhysValMonitoringToolLoose(**kwargs):
if 'TrackSelectionTool' not in kwargs:
from InDetPhysValMonitoring.TrackSelectionTool import getInDetTrackSelectionToolLoose
kwargs = setDefaults(kwargs, TrackSelectionTool=toolFactory(
getInDetTrackSelectionToolLoose))
kwargs = setDefaults(
kwargs,
name='InDetPhysValMonitoringToolLoose',
SubFolder='Loose/',
useTrackSelection=True)
return getInDetPhysValMonitoringTool(**kwargs)
def getInDetPhysValMonitoringToolTightPrimary(**kwargs):
if 'TrackSelectionTool' not in kwargs:
from InDetPhysValMonitoring.TrackSelectionTool import getInDetTrackSelectionToolTightPrimary
kwargs = setDefaults(kwargs, TrackSelectionTool=toolFactory(
getInDetTrackSelectionToolTightPrimary))
kwargs = setDefaults(
kwargs,
name='InDetPhysValMonitoringToolTightPrimary',
SubFolder='TightPrimary/',
useTrackSelection=True)
return getInDetPhysValMonitoringTool(**kwargs)
def getInDetPhysValMonitoringToolHILoose(**kwargs):
if 'TrackSelectionTool' not in kwargs:
from InDetPhysValMonitoring.TrackSelectionTool import getInDetTrackSelectionToolHILoose
kwargs = setDefaults(kwargs, TrackSelectionTool=toolFactory(
getInDetTrackSelectionToolHILoose))
kwargs = setDefaults(
kwargs,
name='InDetPhysValMonitoringToolHILoose',
SubFolder='HILoose/',
useTrackSelection=True)
return getInDetPhysValMonitoringTool(**kwargs)
def getInDetPhysValMonitoringToolGSF(**kwargs):
kwargs = setDefaults(
kwargs,
name='InDetPhysValMonitoringToolGSF',
SubFolder='GSF/',
TrackParticleContainerName='GSFTrackParticles',
useTrackSelection=True)
return getInDetPhysValMonitoringTool(**kwargs)
def getInDetPhysValMonitoringToolElectrons(**kwargs):
from InDetPhysValMonitoring.InDetPhysValDecoration import getInDetRttTruthSelectionTool
kwargs = setDefaults(
kwargs,
TruthSelectionTool=getInDetRttTruthSelectionTool(name="AthTruthSelectionToolForIDPVM_Electrons",pdgId=11,minPt=5000.),
name='InDetPhysValMonitoringToolElectrons',
onlyFillTruthMatched=True,
SubFolder='Electrons/')
return getInDetPhysValMonitoringTool(**kwargs)
def getInDetPhysValMonitoringToolMuons(**kwargs):
from InDetPhysValMonitoring.InDetPhysValDecoration import getInDetRttTruthSelectionTool
kwargs = setDefaults(
kwargs,
TruthSelectionTool=getInDetRttTruthSelectionTool(name="AthTruthSelectionToolForIDPVM_Muons",pdgId=13,minPt=5000.),
name='InDetPhysValMonitoringToolMuons',
onlyFillTruthMatched=True,
SubFolder='Muons/')
return getInDetPhysValMonitoringTool(**kwargs)
def getInDetLargeD0PhysValMonitoringTool(**kwargs):
from InDetRecExample.InDetJobProperties import InDetFlags
from InDetRecExample.InDetKeys import InDetKeys
from InDetPhysValMonitoring.InDetPhysValDecoration import getInDetRttTruthSelectionTool
from InDetPhysValMonitoring.InDetPhysValJobProperties import InDetPhysValFlags
kwargs = setDefaults(
kwargs,
name='InDetPhysValMonitoringToolLargeD0',
SubFolder='LRT/',
TruthSelectionTool=getInDetRttTruthSelectionTool(name="AthTruthSelectionToolForIDPVM_LargeD0",maxProdVertRadius = InDetPhysValFlags.maxProdVertRadius(),minPt=InDetPhysValFlags.truthMinPt(),ancestorList=InDetPhysValFlags.ancestorIDs(), requireSiHit=InDetPhysValFlags.requiredSiHits()),
TrackParticleContainerName=InDetKeys.xAODLargeD0TrackParticleContainer(
) if InDetFlags.storeSeparateLargeD0Container() else InDetKeys.xAODTrackParticleContainer(),
useTrackSelection=True)
return getInDetPhysValMonitoringTool(**kwargs)
# Copyright (C) 2002-2021 CERN for the benefit of the ATLAS collaboration
from InDetRecExample.TrackingCommon import setDefaults
from InDetTrackSelectionTool.InDetTrackSelectionToolConf import InDet__InDetTrackSelectionTool
def getInDetTrackSelectionToolLoose(
name="InDetTrackSelectionToolLoose",
**kwargs):
kwargs = setDefaults(kwargs, CutLevel="Loose")
return InDet__InDetTrackSelectionTool(name, **kwargs)
def getInDetTrackSelectionToolTightPrimary(
name="InDetTrackSelectionToolTightPrimary",
**kwargs):
kwargs = setDefaults(kwargs, CutLevel="TightPrimary")
return InDet__InDetTrackSelectionTool(name, **kwargs)
# Copyright (C) 2002-2022 CERN for the benefit of the ATLAS collaboration
def addTruthJetsIfNotExising(truth_jets_name):
'''
Add algorithm to create the truth jets collection unless the
collection exists already, or a truth jet finder is already running
'''
from RecExConfig.AutoConfiguration import IsInInputFile
# the jet collection name does not exist in the input file
# add a jet finder algorithm in front of the monitoring if the algorithm
# does not yet exist.
if not IsInInputFile('xAOD::JetContainer', truth_jets_name):
try:
from AthenaCommon.Logging import logging
log = logging.getLogger('InDetPhysValMonitoring/addTruthJets.py')
from PyUtils.MetaReaderPeeker import convert_itemList, metadata
eventdata_itemsDic = convert_itemList(layout='dict')
log.info('DEBUG addTruthJetsIfNotExising {} not in {} [file_type={}]'.format(
truth_jets_name, eventdata_itemsDic, metadata['file_type']))
if truth_jets_name in eventdata_itemsDic:
return
except Exception:
pass
# Access the algorithm sequence:
from AthenaCommon.AlgSequence import AlgSequence
topSequence = AlgSequence()
# extract the jet finder type and main parameter
import re
extract_alg = re.search('^([^0-9]+)([0-9]+)TruthJets', truth_jets_name)
if extract_alg is not None:
alg_type = extract_alg.group(1)
alg_param_str = extract_alg.group(2)
else:
alg_type = 'AntiKt'
alg_param_str = "4"
jet_finder_alg_name = "jetalg"+alg_type+alg_param_str+'TruthJets'
# add the jet finder unless it exists already in the alg sequence
from InDetPhysValMonitoring.InDetPhysValDecoration import findAlg, findMonMan
alg_pos = findAlg([jet_finder_alg_name])
if alg_pos is None:
from JetRec.JetRecStandard import jtm
mon_man_index = findMonMan()
# configure truth jet finding ?
from JetRec.JetRecFlags import jetFlags
jetFlags.useTruth = True
jetFlags.useTracks = False
jetFlags.truthFlavorTags = ["BHadronsInitial", "BHadronsFinal", "BQuarksFinal",
"CHadronsInitial", "CHadronsFinal", "CQuarksFinal",
"TausFinal",
"Partons",
]
# tool to create truth jet finding inputs
truth_part_copy_name = 'truthpartcopy'
dir(jtm)
create_truth_jet_input = None
if not hasattr(jtm, truth_part_copy_name):
from MCTruthClassifier.MCTruthClassifierConf import MCTruthClassifier
truth_classifier_name = 'JetMCTruthClassifier'
if not hasattr(jtm, truth_classifier_name):
from AthenaCommon.AppMgr import ToolSvc
if not hasattr(ToolSvc, truth_classifier_name):
truthClassifier = MCTruthClassifier(
name=truth_classifier_name,
ParticleCaloExtensionTool="")
else:
truthClassifier = getattr(
ToolSvc, truth_classifier_name)
jtm += truthClassifier
else:
truthClassifier = getattr(jtm, truth_classifier_name)
from ParticleJetTools.ParticleJetToolsConf import CopyTruthJetParticles
create_truth_jet_input = CopyTruthJetParticles(truth_part_copy_name, OutputName="JetInputTruthParticles",
MCTruthClassifier=truthClassifier)
jtm += create_truth_jet_input
else:
create_truth_jet_input = getattr(jtm, truth_part_copy_name)
jet_tools = []
from JetRec.JetFlavorAlgs import scheduleCopyTruthParticles
jet_tools += scheduleCopyTruthParticles()
jet_tools += [create_truth_jet_input]
jet_tools += jtm.jetrecs
# add the jet finder in front of the monitoring
from JetRec.JetRecConf import JetAlgorithm
from JetRec.JetRecConf import JetToolRunner
jtm += JetToolRunner("jetrun",
Tools=jet_tools,
EventShapeTools=[]
)
# jet_finder_alg = JetAlgorithm(jet_finder_alg_name, jet_tools)
jet_finder_alg = JetAlgorithm(jet_finder_alg_name)
# jet_finder_alg.OutputLevel = 1
jet_finder_alg.Tools = [jtm.jetrun]
if mon_man_index is not None:
topSequence.insert(mon_man_index, jet_finder_alg)
else:
topSequence += jet_finder_alg
# Original author: Goetz Gaycken
# Uploaded by shaun roe, 19 Jan 2016
# The actual reco tf below pre-includes some python snippits
# which are generated in the first few lines.
# create a python script which switches off everything which is not strictly
# needed for InDetPhysValMonitoring or the InDetStandardPlots
# It seems that the main PhysVal monitoring will fail if standard objects
# are not available in 20.1.8.5, so only the trigger can be disabled to speed up
# processing
cat - > switch_off_non_id.py <<EOF
# print rec
# rec.doCalo = False
# rec.doEgamma = False
# rec.doForwardDet = False
# rec.doInDet = True
# rec.doJetMissingETTag = False
# rec.doLArg = False
# rec.doLucid = False
# rec.doMuon = False
# rec.doMuonCombined = False
# rec.doRecoTiming = True
# rec.doSemiDetailedPerfMon = True
# rec.doTau = False
# rec.doTile = False
rec.doTrigger = False
EOF
# switch of track slimming so that residuals can be plotted
# this is needed.
cat - > slimming.py <<EOF
from InDetRecExample.InDetJobProperties import InDetFlags;InDetFlags.doSlimming.set_Value_and_Lock(False)
EOF
# switch on the indet standard plots
# this is needed to get the plots from the InDetPerformanceRTT
cat - > indet_std_plots <<EOF
# from AthenaMonitoring import DQMonFlags;
# DQMonFlags.doStreamAwareMon=False
from InDetRecExample.InDetJobProperties import InDetFlags
InDetFlags.doPhysValMon.set_Value_and_Lock(True);
EOF
# run the reco transform
Reco_tf.py --checkEventCount False --ignoreErrors True --maxEvents 5 --valid True --inputRDOFile root://eosatlas.cern.ch//eos/atlas/atlascerngroupdisk/perf-idtracking/rtt/mc10_7TeV.105200.T1_McAtNlo_Jimmy.digit.RDO.e598_s933_s946_d366/RDO.197091._000015.pool.root.1 --outputNTUP_PHYSVALFile physval.root --validationFlags doInDet --outputAODFile physval.AOD.root --preInclude switch_off_non_id.py r2e:slimming.py,indet_std_plots.py
# --checkEventCount False --ignoreErrors True : to prevent the Reco_tf to abort after a stage due to errors showing up in the log file.
# --maxEvents 5 : to only run on 5 events
# --inputRDOFile ... : specify an RDO input file
# --valid True : needed for InDetPhysValMonitoring. Otherwise the decorator be run too late
# --outputNTUP_PHYSVALFile physval.root : request the PhysVal monitoring histograms and set the histogram file name
# --validationFlags doInDet : request InDet PhysVal monitoring histograms
# --outputAODFile ... : [NOT NECESSARY] request an AOD which includes the decorations by the decrorator of the InDetPhysValMonitoring
# --preInclude r2e:switch_off_non_id.py,slimming.py,indet_std_plots.py e2a:switch_off_non_id.py :
# run these python scripts before processing the normal job options of a certain processing staage
# r2a: after this label the python scripts which are to be run at the very beginning of the RAWtoALL stage (or RDO to ESD)
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment