###############################################################################
# (c) Copyright 2022-2024 CERN for the benefit of the LHCb Collaboration #
# #
# This software is distributed under the terms of the GNU General Public #
# Licence version 3 (GPL Version 3), copied verbatim in the file "COPYING". #
# #
# In applying this licence, CERN does not waive the privileges and immunities #
# granted to it by virtue of its status as an Intergovernmental Organization #
# or submit itself to any jurisdiction. #
###############################################################################
"""
Functor collections for the FunTuple framework.
Refer to the docstrings of the collections in this submodule
for examples of typical usage.
"""
from typing import Optional
from GaudiKernel import SystemOfUnits # type: ignore[import]
from GaudiConf.LbExec import HltSourceID # type: ignore[import]
import Functors as F # type: ignore[import]
from PyConf.reading import get_odin, get_decreports, get_hlt1_selreports, get_particles, tes_root_for_tistos, get_rec_summary # type: ignore[import]
from PyConf.dataflow import DataHandle # type: ignore[import]
from PyConf.Algorithms import Hlt1TisTosAlg, Hlt2TisTosAlg, Hlt1TrueSimEffAlg, Hlt2TrueSimEffAlg # type: ignore[import]
import DaVinciMCTools # type: ignore[import]
from DecayTreeFitter import DecayTreeFitter # type: ignore[import]
from IsolationTools import VertexAndConeIsolation # type: ignore[import]
from .FunctorCollection import FunctorCollection
__all__ = (
"EventInfo",
"SelectionInfo",
"HltTisTos",
"Kinematics",
"MCHierarchy",
"MCKinematics",
"MCVertexInfo",
"MCPromptDecay",
"MCReconstructible",
"MCReconstructed",
"ParticleIsolation",
"ConeIsolation",
"VertexIsolation",
"NeutralCaloInfo",
"ChargedCaloInfo",
"DecayTreeFitterResults",
"ParticleID",
"MCPrimaries",
"RecSummary",
"FlavourTaggingResults",
)
[docs]
def EventInfo() -> FunctorCollection:
"""
Event-level collection of most-used functors for event information.
By default the FunTuple framework stores the `RUN_NUMBER`
and `EVENT_NUMBER` variables.
When running over truth samples (MCParticles) i.e. xgen files,
the `RUN_NUMBER` and `EVENT_NUMBER` are not yet available
(see issue https://gitlab.cern.ch/lhcb/DaVinci/-/issues/90).
Output variables:
- `BUNCHCROSSING_ID` : `F.BUNCHCROSSING_ID(odin)`
- `BUNCHCROSSING_TYPE` : `F.BUNCHCROSSING_TYPE(odin)`
- `ODINTCK` : `F.ODINTCK(odin)`
- `GPSTIME` : `F.GPSTIME(odin)`
Example::
from FunTuple import FunTuple_Particles as Funtuple
import FunTuple.functorcollections as FC
from DaVinci import options
variables = FC.EventInfo()
# ...
tuple = Funtuple(name="MyTuple",
fields=...,
variables=variables,
...
)
"""
odin = get_odin()
event_info = {}
event_info["BUNCHCROSSING_ID"] = F.BUNCHCROSSING_ID(odin)
event_info["BUNCHCROSSING_TYPE"] = F.BUNCHCROSSING_TYPE(odin)
event_info["ODINTCK"] = F.ODINTCK(odin)
event_info["GPSTIME"] = F.GPSTIME(odin)
return FunctorCollection(event_info)
[docs]
def SelectionInfo(
*, selection_type: HltSourceID, trigger_lines: list[str]
) -> FunctorCollection:
"""
Event-level collection of most-used functors for tupling trigger/Sprucing information.
Args:
selection_type (HltSourceID): Name of the selection type i.e. "Hlt1" or "Hlt2" or "Spruce".
Used as branch name prefix when tupling and as source ID to get decision reports.
trigger_lines (list(str)): List of line names for which the decision is requested.
Output variables:
- `_TCK` : `F.TCK(dec_report)` - for each source ID
Example::
from FunTuple import FunTuple_Particles as Funtuple
from DaVinci import options
import FunTuple.functorcollections as FC
# List of lines you want decisions for
selection_type = HltSourceiD.Hlt2
hlt2_lines = ['Hlt2CharmD0ToKmPipLineDecision']
variables = FC.SelectionInfo(selection_type, hlt2_lines)
tuple = Funtuple(name="MyTuple",
fields=...,
variables=variables,
...
)
"""
selection_type = HltSourceID(selection_type)
# get decreports
dec_report = get_decreports(selection_type)
# check that the code ends with decision
add_suff_to_elem = lambda s: s + "Decision" if not s.endswith("Decision") else s
trigger_lines = list(map(add_suff_to_elem, trigger_lines))
trigger_info = FunctorCollection({selection_type.name + "_TCK": F.TCK(dec_report)})
trigger_info += FunctorCollection(
{l: F.DECISION(dec_report, l) for l in trigger_lines}
)
return trigger_info
def _HltMCTrueSimEff(
*,
selection_type: HltSourceID,
trigger_lines: list[str],
mcp_data: DataHandle,
dec_reports: DataHandle,
mc2IDLink: DataHandle,
cand_locations: list[str] = ["dummy_line_name_Decision_loc"],
):
"""
Candidate-level collection to store TrueSim efficiencies for HltEfficiencyChecker
Configured in MooreAnalysis/HltEfficiencyChecker/python/HltEfficiencyChecker/config.py
*Only* intended for usage in HltEfficienyChecker.
These efficiencies are described in the HltEfficiencyChecker documentation:
https://lhcbdoc.web.cern.ch/lhcbdoc/moore/master/tutorials/hltefficiencychecker.html
Note that for HLT1, this is not configured by HltEfficiencyChecker due to MooreAnalysis#41.
Use of this in HltEfficiencyChecker will currently crash.
Args:
selection_type (HltSourceID): Name of the selection type i.e. "Hlt1" or "Hlt2".
Used as branch name prefix when tupling and as source ID to get decision reports.
trigger_lines (list(str)): List of Hlt1/Hlt2 decision names for computing TrueSim efficiencies.
Should be post-fixed with 'Decision'.
mcp_data (DataHandle): TES location of MC particles e.g. output of
`get_mc_particles("/Event/MC/Particles")
dec_reports (DataHandle): TES location of DecReports.
mc2IDLink (DataHandle): TES location of linker between MC particles and MC LHCbIDs of
their tracks. Output of e.g. make_links_lhcbids_mcparticles_tracking_system()
cand_locations (list(str), optional): List of TES locations to the reconstructed candidates
of each line. Only relevant for HLT2, so has a dummy default value.
Example:
Only intended usage is in MooreAnalysis/HltEfficiencyChecker/python/HltEfficiencyChecker/config.py.
"""
if selection_type == HltSourceID.Hlt1:
true_sim_relns = Hlt1TrueSimEffAlg(
name="Hlt1TrueSimEffAlgName_{hash}",
InputMCParticles=mcp_data,
DecReports=dec_reports,
MC2IDLink=mc2IDLink,
SelReports=get_hlt1_selreports(),
TriggerLines=trigger_lines,
).P2TrueSimEffTable
elif selection_type == HltSourceID.Hlt2:
true_sim_relns = Hlt2TrueSimEffAlg(
name="Hlt2TrueSimEffAlgName_{hash}",
InputMCParticles=mcp_data,
DecReports=dec_reports,
MC2IDLink=mc2IDLink,
TriggerLines=trigger_lines,
CandidateLocations=cand_locations,
).P2TrueSimEffTable
else:
raise ValueError(
f'selection_type "{selection_type}" is not supported in _HltMCTrueSimEff functor collection.'
)
return FunctorCollection(
{
f"{line}TrueSim": F.VALUE_OR(False)
@ F.VALUE_FROM_DICT(line)
@ F.MAP_TO_RELATED(true_sim_relns)
for line in trigger_lines
}
)
[docs]
def HltTisTos(
*,
selection_type: HltSourceID,
trigger_lines: list[str],
data: DataHandle,
trigger_particles: Optional[list[DataHandle]] = None,
) -> FunctorCollection:
"""
Candidate-level collection to store TIS (Trigger Independent of Signal) or TOS (Trigger On Signal)
information of candidates for specified Hlt1/Hlt2 lines.
For TOB (Trigger On Both), the condition is (!TIS && !TOS && event_decision).
To store "event decision" of a line use the `SelectionInfo` functor collection.
Args:
selection_type (HltSourceID): Name of the selection type i.e. "Hlt1" or "Hlt2". Currently only "Hlt1" is supported.
Used as branch name prefix when tupling and as source ID to get decision reports.
trigger_lines (list(str)): List of Hlt1/Hlt2 line names for
which the decision is requested.
data (DataHandle): TES location of particles from a Sprucing/Hlt2 line
i.e. output of for example `get_particles("/Event/HLT2/<linename>/Particles")`.
trigger_particles (list(DataHandle), optional): List of TES location of trigger particles.
This could be specified for example when running over "Hlt2" process directly (where only TOS information might be useful).
When running over "Spruce" or "TurboPass" process this is not needed
as the path is inferred from the DV configuration.
Output variables:
- `_TOS` : `F.IS_TOS(line, TisTosRelations)` - for each provided line
- `_TIS` : `F.IS_TIS(line, TisTosRelations)` - for each provided line
Example::
from DaVinci import options
import FunTuple.functorcollections as FC
# Get input data from a particular "<linename>" (can be Sprucing line or Hlt2 line)
data = get_particles(f"/Event/HLT2/<linename>/Particles")
# List of lines for which Tis/Tos information is requested
trigger_lines = ["Hlt1TwoTrackMVADecision", "Hlt1TrackMVADecision"]
variables = FC.HltTisTos(selection_type="Hlt1", trigger_lines=trigger_lines, data=data)
# Store event decision to check if a candidate is "TOB" wrt to a given line
evt_variables = FC.SelectionInfo(selection_type="Hlt1", trigger_lines)
tuple = Funtuple(name="MyTuple",
fields=...,
variables=variables,
event_variables=evt_variables,
...
)
"""
# Helper function to check if the trigger_lines contain some prefix
def _check_if_lines_startswith(lines, prefix):
line_wo_hlt1prefix = [line for line in lines if not line.startswith(prefix)]
if len(line_wo_hlt1prefix) != 0:
raise ValueError(
f"The 'selection_type' is {selection_type}, "
"but the following 'trigger_lines' do not "
f"contain '{prefix}' in their name: {line_wo_hlt1prefix}"
)
selection_type = HltSourceID(selection_type)
if selection_type == HltSourceID.Spruce:
raise ValueError("The selection type can only be Hlt1 or Hlt2.")
# Get the selection and decision reports
hlt1_dec = get_decreports(selection_type)
# Check if the trigger_lines contain the selection_type prefix
_check_if_lines_startswith(trigger_lines, selection_type)
# check if trigger lines end with Decision
# if not add Decision to the end of the line
add_suff_to_elem = lambda s: s + "Decision" if not s.endswith("Decision") else s
trigger_lines = list(map(add_suff_to_elem, trigger_lines))
if selection_type == HltSourceID.Hlt1:
hlt1_sel_report = get_hlt1_selreports()
TisTosRelations = Hlt1TisTosAlg(
name="Hlt1TisTosAlgName_{hash}",
InputParticles=data,
SelReports=hlt1_sel_report,
DecReports=hlt1_dec,
TriggerLines=trigger_lines,
).P2TisTosTable
elif selection_type == HltSourceID.Hlt2:
if trigger_particles is None:
trigger_particles = [
get_particles(
f"{tes_root_for_tistos()}/{line.replace('Decision','')}/Particles"
)
for line in trigger_lines
]
TisTosRelations = Hlt2TisTosAlg(
name="Hlt2TisTosAlgName_{hash}",
InputParticles=data,
DecReports=hlt1_dec,
TriggerLines=trigger_lines,
TriggerParticles=trigger_particles,
).P2TisTosTable
hlttistos_info = FunctorCollection(
{f"{line}_TOS": F.IS_TOS(line, TisTosRelations) for line in trigger_lines}
)
hlttistos_info += FunctorCollection(
{f"{line}_TIS": F.IS_TIS(line, TisTosRelations) for line in trigger_lines}
)
return hlttistos_info
[docs]
def Kinematics() -> FunctorCollection:
"""
Candidate-level collection of most-used functors on kinematics.
Output variables:
- `M` : `F.MASS`
- `P` : `F.P`
- `P[T,X,Y,Z]` : `F.P[T,X,Y,Z]`
- `ENERGY` : `F.ENERGY`
Example::
from FunTuple import FunTuple_Particles as Funtuple
import FunTuple.functorcollections as FC
variables = FC.Kinematics()
tuple = Funtuple(name="MyTuple",
fields=...,
variables=variables,
...
)
"""
kinematics = {
"M": F.MASS,
"P": F.P,
"PT": F.PT,
"PX": F.PX,
"PY": F.PY,
"PZ": F.PZ,
"ENERGY": F.ENERGY,
}
return FunctorCollection(kinematics)
[docs]
def MCKinematics(*, mctruth_alg: DaVinciMCTools.MCTruthAndBkgCat) -> FunctorCollection:
"""
Candidate-level collection of most-used functors on kinematics on the related MC particle.
Args:
mctruth_alg (DaVinciMCTools.MCTruthAndBkgCat): An instance of the `MCTruthAndBkgCat` helper class
that is used for retrieving MC truth information of a reconstructed particle.
Output variables:
- `TRUEP` : `mctruth_alg(F.P)`
- `TRUEPT` : `mctruth_alg(F.PT)`
- `TRUEP[X,Y,Z]` : `mctruth_alg(F.P[X,Y,Z])`
- `TRUEENERGY` : `mctruth_alg(F.ENERGY)`
Example::
from FunTuple import FunTuple_Particles as Funtuple
import FunTuple.functorcollections as FC
from DaVinciMCTools import MCTruthAndBkgCat
# Load particles onto TES location for a given event cycle
input_data = get_particles("/Event/HLT2/<linename>/Particles")
# Define an algorithm that builds one-to-one relation table b/w Reco Particle -> Truth MC Particle.
# Here "options" is the DV options object (see DaVinciExamples or DaVinciTutorials)
mctruth_alg = MCTruthAndBkgCat(input_data)
variables = FC.MCKinematics(mctruth_alg)
tuple = Funtuple(name="MyTuple",
fields=...,
variables=variables,
...
)
"""
mckinematics = {
"TRUEP": mctruth_alg(F.P),
"TRUEPT": mctruth_alg(F.PT),
"TRUEPX": mctruth_alg(F.PX),
"TRUEPY": mctruth_alg(F.PY),
"TRUEPZ": mctruth_alg(F.PZ),
"TRUEENERGY": mctruth_alg(F.ENERGY),
}
return FunctorCollection(mckinematics)
def ParticlePVInfo() -> FunctorCollection:
"""
Candidate-level collection of functors yielding information of the primary vertex associated to the particle.
Args:
Output variables:
- `OWNPV_[X,Y,Z]`: `F.OWNPV[X,Y,Z]`
- `OWNPV_NDOF`: `F.NDOF @ F.OWNPV`
Example::
from FunTuple import FunTuple_Particles as Funtuple
import FunTuple.functorcollections as FC
variables = FC.ParticlePVInfo()
tuple = Funtuple(name="MyTuple",
fields=...,
inputs=...,
variables=variables,
...
)
"""
pvinfo = {
"OWNPV_X": F.OWNPVX,
"OWNPV_Y": F.OWNPVY,
"OWNPV_Z": F.OWNPVZ,
"OWNPV_NDOF": F.NDOF @ F.OWNPV,
}
return FunctorCollection(pvinfo)
[docs]
def DecayTreeFitterResults(
*,
DTF: DecayTreeFitter,
decay_origin: bool = True,
with_lifetime: bool = False,
with_kinematics: bool = True,
prefix: str = "DTF",
) -> FunctorCollection:
"""
Candidate-level collection of functors that access DecayTreeFitter fit results.
Args:
DTF (DecayTreeFitter.DecayTreeFitter): An instance of the `DecayTreeFitter` helper class
that builds the necessary functors.
decay_origin (bool), optional: Returns functorcollection for decay origin. Defaults to True.
with_lifetime (bool, optional): The returned functors comprise lifetime and flight distance. Defaults to False.
with_kinematics (bool, optional): The returned functors comprise kinematics. Defaults to True.
prefix (str, optional): Additional prefix to output functors. Defaults to 'DTF'.
Output variables:
- `_CHI2`
- `_NDOF`
- `_CHI2DOF`
- `_NITER`
- `_MASS`
- `_MASSERR`
- `_P`
- `_PERR`
If with PV constraint:
- `_PV_KEY`
- `_PV_[X,Y,Z]`
If with lifetime:
- `_CTAU`
- `_CTAUERR`
- `_FD`
- `_FDERR`
If with kinematics:
- `_P[X,Y,Z,E]`
Example::
from FunTuple import FunTuple_Particles as Funtuple
import FunTuple.functorcollections as FC
from DecayTreeFitter import DecayTreeFitter
DTF = DecayTreeFitter(
name = 'DTF',
input_particles = get_particles(...),
input_pvs = get_pvs()
)
variable_decay_origin = FC.DecayTreeFitterResults(DTF, decay_origin=True, with_lifetime=True, with_kinematics=True)
variable_kinematics = FC.DecayTreeFitterResults(DTF, decay_origin=False, with_lifetime=False, with_kinematics=True)
variables = {
'B': variable_decay_origin,
'phi_1020': variable_kinematics,
'Kplus': variable_kinematics,
'Kminus': variable_kinematics,
}
tuple = Funtuple(name="MyTuple",
fields=...,
variables=variables,
...
)
"""
suffix = f"[{prefix}_IDX]" if DTF.FitAllPVs else ""
output_collection = FunctorCollection()
if decay_origin:
output_collection += FunctorCollection(
{
f"{prefix}_CHI2{suffix}": DTF.CHI2,
f"{prefix}_NDOF{suffix}": DTF.NDOF,
f"{prefix}_CHI2DOF{suffix}": DTF.CHI2DOF,
f"{prefix}_NITER{suffix}": DTF.NITER,
f"{prefix}_MASS{suffix}": DTF.MASS,
f"{prefix}_MASSERR{suffix}": DTF.MASSERR,
f"{prefix}_P{suffix}": DTF.P,
f"{prefix}_PERR{suffix}": DTF.PERR,
}
)
if DTF.HasPVConstraint:
output_collection += FunctorCollection(
{
f"{prefix}_PV_KEY{suffix}": F.VALUE_OR(-1)
@ DTF(F.VALUE_OR(-1) @ F.OBJECT_KEY @ F.ENDVERTEX),
f"{prefix}_PV_X{suffix}": DTF(F.END_VX),
f"{prefix}_PV_Y{suffix}": DTF(F.END_VY),
f"{prefix}_PV_Z{suffix}": DTF(F.END_VZ),
}
)
if with_lifetime:
output_collection += FunctorCollection(
{
f"{prefix}_CTAU{suffix}": DTF.CTAU,
f"{prefix}_CTAUERR{suffix}": DTF.CTAUERR,
f"{prefix}_FD{suffix}": DTF.FD,
f"{prefix}_FDERR{suffix}": DTF.FDERR,
}
)
if with_kinematics:
output_collection += FunctorCollection(
{
f"{prefix}_PX{suffix}": DTF(F.PX),
f"{prefix}_PY{suffix}": DTF(F.PY),
f"{prefix}_PZ{suffix}": DTF(F.PZ),
f"{prefix}_PE{suffix}": DTF(F.ENERGY),
}
)
return output_collection
[docs]
def MCPromptDecay(
*,
mctruth_alg: Optional[DaVinciMCTools.MCTruthAndBkgCat] = None,
max_lifetime: float = 1e-7 * SystemOfUnits.ns,
store_longlived_particle_id: bool = True,
) -> FunctorCollection:
"""
Candidate-level collection of functors to determine if a decay is prompt based on the true lifetime of its ancestors.
If a decay does not have any long-lived ancestors, it is classified as prompt.
This algorithm is based on the Runs 1 and 2
MCTupleToolPrompt tool (https://gitlab.cern.ch/lhcb/Analysis/-/blob/run2-patches/Phys/DecayTreeTupleMC/src/MCTupleToolPrompt.cpp).
If no argument is provided, i.e. mctruth_alg==None, this collection creates functors that apply to MCParticles;
otherwise, the functors created are assumed to apply to Particles,
which will find the associated MCParticle automatically using the input truth matching.
Args:
mctruth_alg (DaVinciMCTools.MCTruthAndBkgCat, optional): An instance of the `MCTruthAndBkgCat` helper class
that is used for retrieving MC truth information of a reconstructed particle. Defaults to None.
max_lifetime (float, optional): maximum lifetime of short-lived particles (ns). Defaults to 1e-7 ns.
store_longlived_particle_id (bool, optional): create functors to store the particle ID and container key of the first long-lived ancestor. Defaults to True.
Output variables:
- `MC_ISPROMPT`
- `MC_LONGLIVED_ID`
- `MC_LONGLIVED_KEY`
Example::
from FunTuple import FunTuple_Particles as Funtuple
import FunTuple.functorcollections as FC
from DaVinciMCTools import MCTruthAndBkgCat
# Load particles onto TES location for a given event cycle
input_data = get_particles("/Event/HLT2/<linename>/Particles")
# Define an algorithm that builds one-to-one relation table b/w Reco Particle -> Truth MC Particle.
# Here "options" is the DV options object (see DaVinciExamples or DaVinciTutorials)
mctruth_alg = MCTruthAndBkgCat(input_data)
variables = FC.MCPromptDecay(mctruth_alg)
tuple = Funtuple(name="MyTuple",
fields=...,
variables=variables,
inputs=input_data)
"""
if mctruth_alg is None:
MCTRUTH = lambda func: func
else:
MCTRUTH = mctruth_alg
collection = FunctorCollection(
{
"MC_ISPROMPT": F.VALUE_OR(False) @ MCTRUTH(F.MC_ISPROMPT(max_lifetime)),
}
)
if store_longlived_particle_id:
collection += FunctorCollection(
{
"MC_LONGLIVED_ID": F.VALUE_OR(0)
@ MCTRUTH(
F.VALUE_OR(0)
@ F.PARTICLE_ID
@ F.MC_FIRST_LONGLIVED_ANCESTOR(max_lifetime)
),
"MC_LONGLIVED_KEY": F.VALUE_OR(-1)
@ MCTRUTH(
F.VALUE_OR(-1)
@ F.OBJECT_KEY
@ F.MC_FIRST_LONGLIVED_ANCESTOR(max_lifetime)
),
}
)
return collection
[docs]
def MCHierarchy(
*, mctruth_alg: Optional[DaVinciMCTools.MCTruthAndBkgCat] = None
) -> FunctorCollection:
"""
Candidate-level collection of functors yielding the hierarchy/relationships among MC particles.
If no argument is provided, i.e. `mctruth_alg==None`, this collection creates functors that apply to MCParticles;
otherwise, the functors created are assumed to apply to Particles,
which will find the associated MCParticle automatically using the input truth matching.
Args:
mctruth_alg (DaVinciMCTools.MCTruthAndBkgCat, optional): An instance of the `MCTruthAndBkgCat` helper class
that is used for retrieving MC truth information of a reconstructed particle. Defaults to None.
Output variables:
- `TRUEID`
- `MC_MOTHER_ID`
- `MC_MOTHER_KEY`
- `MC_GD_MOTHER_ID`
- `MC_GD_MOTHER_KEY`
- `MC_GD_GD_MOTHER_ID`
- `MC_GD_GD_MOTHER_KEY`
Example::
from FunTuple import FunTuple_Particles as Funtuple
import FunTuple.functorcollections as FC
from DaVinciMCTools import MCTruthAndBkgCat
# Load particles onto TES location for a given event cycle
input_data = get_particles("/Event/HLT2/<linename>/Particles")
# Define an algorithm that builds one-to-one relation table b/w Reco Particle -> Truth MC Particle.
# Here "options" is the DaVinci options object (see DaVinciExamples or DaVinciTutorials package)
mctruth_alg = MCTruthAndBkgCat(input_data)
variables = FC.MCHierarchy(mctruth_alg)
tuple = Funtuple(name="MyTuple",
fields=...,
variables=variables,
inputs=input_data)
"""
if mctruth_alg is None:
MCTRUTH = lambda func: func
else:
MCTRUTH = mctruth_alg
MCMOTHER_ID = lambda n: F.VALUE_OR(0) @ MCTRUTH(F.MC_MOTHER(n, F.PARTICLE_ID))
MCMOTHER_KEY = lambda n: F.VALUE_OR(-1) @ MCTRUTH(F.MC_MOTHER(n, F.OBJECT_KEY))
MCHierarchy = {
"TRUEID": F.VALUE_OR(0) @ MCTRUTH(F.PARTICLE_ID),
"MC_MOTHER_ID": MCMOTHER_ID(1),
"MC_MOTHER_KEY": MCMOTHER_KEY(1),
"MC_GD_MOTHER_ID": MCMOTHER_ID(2),
"MC_GD_MOTHER_KEY": MCMOTHER_KEY(2),
"MC_GD_GD_MOTHER_ID": MCMOTHER_ID(3),
"MC_GD_GD_MOTHER_KEY": MCMOTHER_KEY(3),
}
return FunctorCollection(MCHierarchy)
[docs]
def MCVertexInfo(
*, mctruth_alg: Optional[DaVinciMCTools.MCTruthAndBkgCat] = None
) -> FunctorCollection:
"""
Candidate-level collection of functors on vertices of the related MC particle.
Args:
mctruth_alg (DaVinciMCTools.MCTruthAndBkgCat, optional): An instance of the `MCTruthAndBkgCat` helper class
that is used for retrieving monte-carlo truth information of a reconstructed particle. Defaults to None.
If no argument is provided, i.e. `mctruth_alg==None`, this collection creates functors that apply to MCParticles;
otherwise, the functors created are assumed to apply to Particles,
which will find the associated MCParticle automatically using the input truth matching.
Output variables:
- `TRUEORIGINVERTEX_[X,Y,Z]`
- `TRUEENDVERTEX_[X,Y,Z]`
Example::
from FunTuple import FunTuple_Particles as Funtuple
import FunTuple.functorcollections as FC
from DaVinciMCTools import MCTruthAndBkgCat
# Load particles onto TES location for a given event cycle
input_data = get_particles("/Event/HLT2/<linename>/Particles")
# Define an algorithm that builds one-to-one relation table b/w Reco Particle -> Truth MC Particle.
# Here "options" is the DV options object (see DaVinciExamples or DaVinciTutorials)
mctruth_alg = MCTruthAndBkgCat(input_data)
variables = FC.MCVertexInfo(mctruth_alg)
tuple = Funtuple(name="MyTuple",
fields=...,
variables=variables,
...
)
"""
if mctruth_alg is None:
MCTRUTH = lambda func: func
else:
MCTRUTH = mctruth_alg
functor_dictionary = {
"TRUEORIGINVERTEX_X": MCTRUTH(F.ORIGIN_VX),
"TRUEORIGINVERTEX_Y": MCTRUTH(F.ORIGIN_VY),
"TRUEORIGINVERTEX_Z": MCTRUTH(F.ORIGIN_VZ),
"TRUEENDVERTEX_X": MCTRUTH(F.END_VX),
"TRUEENDVERTEX_Y": MCTRUTH(F.END_VY),
"TRUEENDVERTEX_Z": MCTRUTH(F.END_VZ),
}
return FunctorCollection(functor_dictionary)
def MCPrimaryVertexInfo(
*, mctruth_alg: Optional[DaVinciMCTools.MCTruthAndBkgCat] = None
) -> FunctorCollection:
"""
Candidate-level collection of functors yielding the X, Y and Z positions of the primary vertex of the
(related) MC particle.
Args:
mctruth_alg (DaVinciMCTools.MCTruthAndBkgCat, optional): An instance of the `MCTruthAndBkgCat` helper class
that is used for retrieving monte-carlo truth information of a reconstructed particle. Defaults to None.
)
If none, no call is made to the truth-matching. This is suitable for use only for tuples which
directly run on MCParticles (MCTuples).
Output variables:
- `TRUEPRIMARYVERTEX_[X,Y,Z]`
"""
if mctruth_alg is None:
MCTRUTH = lambda func: func
else:
MCTRUTH = mctruth_alg
mc_primary_vertex_info_functors = {
"TRUEPRIMARYVERTEX_X": MCTRUTH(F.MC_PV_VX),
"TRUEPRIMARYVERTEX_Y": MCTRUTH(F.MC_PV_VY),
"TRUEPRIMARYVERTEX_Z": MCTRUTH(F.MC_PV_VZ),
}
return FunctorCollection(mc_primary_vertex_info_functors)
[docs]
def MCPrimaries(*, mc_header: DataHandle) -> FunctorCollection:
"""
Event-level collection of functors to store the properties (X,Y,Z,T) of all MC primary vertices.
Args:
mc_header (DataHandle): The TES location of the `MCHeader` object,
obtainded via the `PyConf.reading.get_mc_header` function.
Output variables:
- `MCPVX[MCPV_IDX]` : `F.MAP(F.VALUE_OR(F.NaN) @ F.X_COORDINATE @ F.POSITION)`
- `MCPVY[MCPV_IDX]` : `F.MAP(F.VALUE_OR(F.NaN) @ F.Y_COORDINATE @ F.POSITION)`
- `MCPVZ[MCPV_IDX]` : `F.MAP(F.VALUE_OR(F.NaN) @ F.Z_COORDINATE @ F.POSITION)`
- `MCPVT[MCPV_IDX]` : `F.MAP(F.VALUE_OR(-1) @ F.MC_VTX_TIME)`
- `MCPV_SIZE` : `F.SIZE_OF @ F.MC_ALLPVS(mc_header)`
Example::
from FunTuple import FunTuple_Particles as Funtuple
import FunTuple.functorcollections as FC
mc_header = get_mc_header()
evt_variables = FC.MCPrimaries(mc_header)
tuple = Funtuple(name="MyTuple",
fields=...,
variables=...,
event_variables=evt_variables,
...
)
"""
mcprimaries = {
"MCPVX[MCPV_IDX]": F.MAP(F.VALUE_OR(F.NaN) @ F.X_COORDINATE @ F.POSITION)
@ F.MC_ALLPVS(mc_header),
"MCPVY[MCPV_IDX]": F.MAP(F.VALUE_OR(F.NaN) @ F.Y_COORDINATE @ F.POSITION)
@ F.MC_ALLPVS(mc_header),
"MCPVZ[MCPV_IDX]": F.MAP(F.VALUE_OR(F.NaN) @ F.Z_COORDINATE @ F.POSITION)
@ F.MC_ALLPVS(mc_header),
"MCPVT[MCPV_IDX]": F.MAP(F.VALUE_OR(-1) @ F.MC_VTX_TIME)
@ F.MC_ALLPVS(mc_header),
"MCPV_SIZE": F.SIZE_OF @ F.MC_ALLPVS(mc_header),
}
return FunctorCollection(mcprimaries)
[docs]
def MCReconstructible(
*,
mcreconstructible_alg: DaVinciMCTools.MCReconstructible,
extra_info: bool = False,
mctruth_alg: Optional[DaVinciMCTools.MCTruthAndBkgCat] = None,
) -> FunctorCollection:
"""
Candidate-level collection of functors to store reconstrutible information for the MC particle.
A stable charged or neutral MCParticle can be reconstructible in one of the following categories:
-1 = No MC classification possible
0 = Outside detector acceptance
1 = In acceptance but not reconstructible
2 = Reconstructible as a Long charged track
3 = Reconstructible as a Downstream charged track
4 = Reconstructible as an Upstream charged track
5 = Reconstructible as a T charged track
6 = Reconstructible as a VELO charged track
50 = Reconstructible as a Neutral particle
Note that the support for stable neutral MCParticle is not present yet (requires separate functors).
An MC decay to be reconstructible implies that all final state MCParticles are reconstructible.
For more info on definition (Run1/2) see: https://lhcb-comp.web.cern.ch/analysis/davinci/v8/recrecdefinition.htm
Args:
mcreconstructible_alg (DaVinciMCTools.MCReconstructible): An instance of the `MCReconstructible` helper class
that builds the necessary functors.
extra_info (bool, optional): If False, returns a functor to ntuple reconstructible category for
charged MCParticle. If True, returns extra information e.g. hasUT, hasVelo, etc. Defaults to False.
mctruth_alg (DaVinciMCTools.MCTruthAndBkgCat, optional): An instance of the `MCTruthAndBkgCat` helper class
that is used for retrieving monte-carlo truth information of a reconstructed particle. Defaults to None.
Output variables:
- `MC_RECONSTRUCTIBLE`
If `extra_info`:
- `MC_HASVELO`
- `MC_HASVELO_AND_T`
- `MC_HAST[1,2,3][ ,X,S]`
- `MC_HASUT[1,2]`
- `MC_ACCVELO`
- `MC_ACCVELO_AND_T`
- `MC_ACCT[1,2,3][ ,X,S]`
- `MC_ACCUT[ ,1,2]`
Example::
from FunTuple import FunTuple_MCParticles as Funtuple
import FunTuple.functorcollections as FC
from DaVinciMCTools import MCReconstructible as MCRectible
# Load MCParticles onto TES location for a given event cycle
input_data = get_particles("/Event/HLT2/MC/Particles")
# Define the helper class and get functor collection
reconstructible_alg = MCRectible()
variables = FC.MCReconstructible(reconstructible_alg)
tuple = Funtuple(name="MyTuple",
fields=...,
variables=variables,
...
)
"""
if mctruth_alg is None:
MCTRUTH = lambda func: func
else:
MCTRUTH = mctruth_alg
# See above reconstructible categories or functor "MC_ChargeReconstructible" in "Simulation.h" in Rec.
func_dict = {"MC_RECONSTRUCTIBLE": MCTRUTH(mcreconstructible_alg.Reconstructible)}
if extra_info:
func_dict.update(
{
# Has sub-detector hits
# https://gitlab.cern.ch/lhcb/LHCb/-/blob/master/Event/MCEvent/include/Event/MCTrackInfo.h
"MC_HASVELO": MCTRUTH(mcreconstructible_alg.HasVelo),
"MC_HASVELO_AND_T": MCTRUTH(mcreconstructible_alg.HasVeloAndT),
"MC_HAST": MCTRUTH(mcreconstructible_alg.HasT),
"MC_HAST1": MCTRUTH(mcreconstructible_alg.HasT1),
"MC_HAST2": MCTRUTH(mcreconstructible_alg.HasT2),
"MC_HAST3": MCTRUTH(mcreconstructible_alg.HasT3),
"MC_HAST1X": MCTRUTH(mcreconstructible_alg.HasT1X),
"MC_HAST2X": MCTRUTH(mcreconstructible_alg.HasT2X),
"MC_HAST3X": MCTRUTH(mcreconstructible_alg.HasT3X),
"MC_HAST1S": MCTRUTH(mcreconstructible_alg.HasT1S),
"MC_HAST2S": MCTRUTH(mcreconstructible_alg.HasT2S),
"MC_HAST3S": MCTRUTH(mcreconstructible_alg.HasT3S),
"MC_HASUT": MCTRUTH(mcreconstructible_alg.HasUT),
"MC_HASUT1": MCTRUTH(mcreconstructible_alg.HasUT1),
"MC_HASUT2": MCTRUTH(mcreconstructible_alg.HasUT2),
# Within sub-detector acceptance
# https://gitlab.cern.ch/lhcb/LHCb/-/blob/master/Event/MCEvent/include/Event/MCTrackInfo.h
"MC_ACCVELO": MCTRUTH(mcreconstructible_alg.AccVelo),
"MC_ACCVELO_AND_T": MCTRUTH(mcreconstructible_alg.AccVeloAndT),
"MC_ACCT": MCTRUTH(mcreconstructible_alg.AccT),
"MC_ACCT1": MCTRUTH(mcreconstructible_alg.AccT1),
"MC_ACCT2": MCTRUTH(mcreconstructible_alg.AccT2),
"MC_ACCT3": MCTRUTH(mcreconstructible_alg.AccT3),
"MC_ACCT1X": MCTRUTH(mcreconstructible_alg.AccT1X),
"MC_ACCT2X": MCTRUTH(mcreconstructible_alg.AccT2X),
"MC_ACCT3X": MCTRUTH(mcreconstructible_alg.AccT3X),
"MC_ACCT1S": MCTRUTH(mcreconstructible_alg.AccT1S),
"MC_ACCT2S": MCTRUTH(mcreconstructible_alg.AccT2S),
"MC_ACCT3S": MCTRUTH(mcreconstructible_alg.AccT3S),
"MC_ACCUT": MCTRUTH(mcreconstructible_alg.AccUT),
"MC_ACCUT1": MCTRUTH(mcreconstructible_alg.AccUT1),
"MC_ACCUT2": MCTRUTH(mcreconstructible_alg.AccUT2),
##These should be updated
# "MC_HASVELOR": MCTRUTH(mcreconstructible_alg.HasVeloR),
# "MC_HASVELOPhi": MCTRUTH(mcreconstructible_alg.HasVeloPhi),
# "MC_ACCVELOR": MCTRUTH(mcreconstructible_alg.AccVeloR),
# "MC_ACCVELOPhi": MCTRUTH(mcreconstructible_alg.AccVeloPhi),
}
)
return FunctorCollection(func_dict)
[docs]
def MCReconstructed(
*, mcreconstructed_alg: DaVinciMCTools.MCReconstructed, extra_info: bool = False
) -> FunctorCollection:
"""
Candidate-level collection of functors to store reconstrutible information for the MC particle.
A stable charged or neutral MCParticle can be reconstructed in one of the following categories:
-1 No MC classification possible (e.g. NO MC)
0 Not reconstructed
1 Reconstructed as a Long charged track
2 Reconstructed as a Downstream charged track
3 Reconstructed as an Upstream charged track
4 Reconstructed as a T charged track
5 Reconstructed as a VELO charged track
6 Reconstructed as a 2D Velo track
7 Reconstructed as a Muon track
50 Reconstructed as a Neutral particle
51 Reconstructed as a merged pi0
52 Reconstructed as a type that doesn't fit other categories
An MC decay to be reconstructed implies that all final state MCParticles are reconstructed.
For more info on definition (Run1/2) see: https://lhcb-comp.web.cern.ch/analysis/davinci/v8/recrecdefinition.htm
(since the method tuple->farray() only support vector<double> currently, all array will convert to array of floats)
Args:
mcreconstructed_alg (DaVinciMCTools.MCReconstructed): An instance of the `MCReconstructed` helper class
that first maps MCParticle to reconstructed track and builds the necessary functors.
extra_info (bool, optional): If False, returns a functor to ntuple reconstructed category for MCParticle.
If True, returns extra information e.g. TrackType, hasUT, hasVelo, TrackChi2, GhostProb, etc.
Defaults to False.
Example::
from FunTuple import FunTuple_MCParticles as Funtuple
import FunTuple.functorcollections as FC
from DaVinciMCTools import MCReconstructed as MCRected
#load particles onto TES location for a given event cycle
input_data = get_particles("/Event/HLT2/MC/Particles")
#map MCParticle -> reconstructed track
mcreconstructed_alg = MCRected(input_data)
#get the variables
variables = FC.MCReconstructed(mcreconstructed_alg)
tuple = Funtuple(name="MyTuple",
fields=...,
variables=variables,
...
)
"""
suffix = ""
if not mcreconstructed_alg.use_best_mcmatch:
suffix = "[TRACK_INDX]"
func_dict = {
f"RECONSTRUCTED{suffix}": mcreconstructed_alg.Reconstructed,
}
if extra_info:
func_dict.update(
{
# For various track properties (e.g track type) see:
# https://gitlab.cern.ch/lhcb/LHCb/-/blob/master/Event/TrackEvent/include/Event/TrackEnums.h
f"TRACK_TYPE{suffix}": mcreconstructed_alg.TrackType,
f"TRACK_FLAG{suffix}": mcreconstructed_alg.Flag,
f"TRACK_HISTORY{suffix}": mcreconstructed_alg.History,
# Common track properties
f"TRACK_REFPOINT_X{suffix}": mcreconstructed_alg.ReferencePoint_X,
f"TRACK_REFPOINT_Y{suffix}": mcreconstructed_alg.ReferencePoint_Y,
f"TRACK_REFPOINT_Z{suffix}": mcreconstructed_alg.ReferencePoint_Z,
f"TRACK_QOVERP{suffix}": mcreconstructed_alg.QOVERP,
f"TRACK_CHI2{suffix}": mcreconstructed_alg.Chi2,
f"TRACK_CHI2DOF{suffix}": mcreconstructed_alg.Chi2DOF,
f"TRACK_NDOF{suffix}": mcreconstructed_alg.nDoF,
f"TRACK_GHOSTPROB{suffix}": mcreconstructed_alg.GhostProbability,
f"TRACK_HAST{suffix}": mcreconstructed_alg.HasT,
f"TRACK_HASUT{suffix}": mcreconstructed_alg.HasUT,
f"TRACK_HASVELO{suffix}": mcreconstructed_alg.HasVelo,
f"TRACK_NHITS{suffix}": mcreconstructed_alg.nHits,
f"TRACK_NVPHITS{suffix}": mcreconstructed_alg.nVPHits,
f"TRACK_NUTHITS{suffix}": mcreconstructed_alg.nUTHits,
f"TRACK_NFTHITS{suffix}": mcreconstructed_alg.nFTHits,
# PID properties
f"TRACK_PID_MU{suffix}": mcreconstructed_alg.PIDmu,
f"TRACK_PID_PI{suffix}": mcreconstructed_alg.PIDpi,
f"TRACK_PID_K{suffix}": mcreconstructed_alg.PIDk,
f"TRACK_PID_P{suffix}": mcreconstructed_alg.PIDp,
f"TRACK_PID_E{suffix}": mcreconstructed_alg.PIDe,
}
)
if not mcreconstructed_alg.use_best_mcmatch:
func_dict[f"TRACK_MCASSOCWEIGHTS{suffix}"] = mcreconstructed_alg.MCAssoc_Weights
# Note no suffix needed since we are returning simply a number representing
# the number of reconstructed tracks associated to an MC particle
# (in technical terms the size of object "range" from relations table)
func_dict["N_RECO_TRACKS"] = mcreconstructed_alg.nRecoTracks
return FunctorCollection(func_dict)
[docs]
def ParticleIsolation(
*,
isolation_alg: VertexAndConeIsolation,
array_indx_name: str,
sumcone_invalid_value=F.NaN,
) -> FunctorCollection:
"""
Candidate-level collection of functors on track or neutral isolation, using information from relations between sets of particles determined by a given criterion.
It accesses to `VertexAndConeIsolation` class information.
Note::
Variables will have the following naming convention:
`HEAD_<isolation_alg.name>_<var_name>`
Args:
isolation_alg (VertexAndConeIsolation): An instance of the `VertexAndConeIsolation` helper
class that is used to retrieve isolation variables.
array_indx_name (string): string that helps in naming the default `indx` variable.
This appears due to the functor arrays `_DPHI`, `_DETA`.
sumcone_invalid_value (optional):
Value that allows user flexibility in defining the invalid values related to the WeightedRelTable being empty.
Defaults to `F.NaN`.
Output variables:
- `_CMULT`
- `_CP[,T,X,Y,Z]`
- `_P[,T,X,Y,Z]ASY`
- `_DETA`
- `_DPHI`
Example::
import Functors as F
from FunTuple import FunTuple_Particles as Funtuple
from IsolationTools import VertexAndConeIsolation
from FunTuple.functorcollections import ParticleIsolation
from PyConf.reading import get_particles
branches = {"B": "[B0 -> ([J/psi(1S) -> tau+ mu-]CC)(K*(892)0 -> K+ pi-)]CC"}
b2ksttaumu_data = get_particles(f"/Event/HLT2/{line}/Particles")
b_cciso_data = get_particles(f"/Event/HLT2/{line}/B_{tes_long_track_iso}/Particles")
cone_isolation_b = VertexAndConeIsolation(
name = 'TrackIsolation',
reference_particles = b2ksttaumu_data,
related_particles = b_cciso_data,
cut = (F.DR<1.))
variables = FC.ParticleIsolation(isolation_alg=cone_isolation_b,
array_indx_name='indx')
tuple = Funtuple(name="MyTuple",
fields=branches,
variables=variables,
...)
"""
prefix = "HEAD"
if isolation_alg.name:
prefix += f"_{isolation_alg.name}"
ParticleIsolationVariables = {
f"{prefix}_CMULT": isolation_alg.MULT,
f"{prefix}_CP": isolation_alg.CP,
f"{prefix}_CPT": isolation_alg.CPT,
f"{prefix}_CPX": isolation_alg.CPX,
f"{prefix}_CPY": isolation_alg.CPY,
f"{prefix}_CPZ": isolation_alg.CPZ,
f"{prefix}_PASY": isolation_alg.PASY,
f"{prefix}_PTASY": isolation_alg.PTASY,
f"{prefix}_PXASY": isolation_alg.PXASY,
f"{prefix}_PYASY": isolation_alg.PYASY,
f"{prefix}_PZASY": isolation_alg.PZASY,
f"{prefix}_DETA[{array_indx_name}]": isolation_alg.DETA,
f"{prefix}_DPHI[{array_indx_name}]": isolation_alg.DPHI,
}
return FunctorCollection(ParticleIsolationVariables)
[docs]
def ConeIsolation(
*,
charged_cone_isolation_alg: VertexAndConeIsolation,
neutral_cone_isolation_alg: VertexAndConeIsolation,
array_indx_name: str,
sumcone_invalid_value=F.NaN,
) -> FunctorCollection:
"""
Candidate-level collection of functors on charged and neutral isolation, using information from relations between sets of particles determined by a given criterion.
It accesses to `VertexAndConeIsolation` class information.
Note:
Variables will have the following naming convention:
`HEAD_CC_<charged_cone_isolation_alg.name>_<var_name>` for charged isolation
`HEAD_NC_<neutral_cone_isolation_alg.name>_<var_name>` for neutral isolation
Args:
charge_cone_isolation_alg (VertexAndConeIsolation): An instance of the `VertexAndConeIsolation` helper class that is used to retrieve isolation variables for charge cone.
neutral_cone_isolation_alg (VertexAndConeIsolation): An instance of the `VertexAndConeIsolation` helper class that is used to retrieve isolation variables for neutral cone.
array_indx_name (string): string that helps in naming the default `indx` variable.
This appears due to the functor arrays _DPHI, _DETA.
sumcone_invalid_value (optional):
Value that allows user flexibility in defining the invalid values related to the WeightedRelTable being empty.
Defaults to `F.NaN`.
Output variables:
These are for both the neutral (`NC`) and charged (`CC`) cones:
- `_Max_P` : `charged_cone_isolation_alg.MAXCP`
- `_Max_PT` : `charged_cone_isolation_alg.MAXCPT`
- `_Min_P` : `charged_cone_isolation_alg.MINCP`
- `_Min_PT` : `charged_cone_isolation_alg.MINCPT`
Example::
import Functors as F
from FunTuple import FunTuple_Particles as Funtuple
from FunTuple.functorcollections import ConeIsolation
from IsolationTools import VertexAndConeIsolation
from PyConf.reading import get_particles
from GaudiKernel.SystemOfUnits import GeV
branches = {"B": "[B0 -> ([J/psi(1S) -> tau+ mu-]CC)(K*(892)0 -> K+ pi-)]CC"}
b2ksttaumu_data = get_particles(f"/Event/HLT2/{line}/Particles")
b_cciso_data = get_particles(f"/Event/HLT2/{line}/B_{tes_long_track_iso}/Particles")
b_nciso_data = get_particles(f"/Event/HLT2/{line}/B_{tes_neutrals_iso}/Particles")
charged_cone_isolation_b = VertexAndConeIsolation(
name = 'ConeIsolation05',
reference_particles = b2ksttaumu_data,
related_particles = b_cciso_data,
cut = (F.SQRT@F.DR2<0.5))
neutral_cone_isolation_b = VertexAndConeIsolation(
name = 'ConeIsolation10',
reference_particles = b2ksttaumu_data,
related_particles = b_nciso_data,
cut = (F.SQRT@F.DR2<1.))
variables = FC.ConeIsolation(charged_cone_isolation_alg=charged_cone_isolation_b,
neutral_cone_isolation_alg=neutral_cone_isolation_b,
array_indx_name='indx')
tuple = Funtuple(name="MyTuple",
fields=branches,
variables=variables,
...)
"""
cc_str = "CC"
nc_str = "NC"
if charged_cone_isolation_alg:
cc_str += f"_{charged_cone_isolation_alg.name}"
if neutral_cone_isolation_alg:
nc_str += f"_{neutral_cone_isolation_alg.name}"
charged_cone_isolation_alg.name = cc_str
neutral_cone_isolation_alg.name = nc_str
cc_prefix = "HEAD_" + cc_str
nc_prefix = "HEAD_" + nc_str
# Invoke twice the ParticleIsolation functorcollection, once for neutral and once for charged particles, then add to the functor dictionary
ConeIsolationVariables = {
f"{cc_prefix}_Max_P": charged_cone_isolation_alg.MAXCP,
f"{cc_prefix}_Max_PT": charged_cone_isolation_alg.MAXCPT,
f"{cc_prefix}_Min_P": charged_cone_isolation_alg.MINCP,
f"{cc_prefix}_Min_PT": charged_cone_isolation_alg.MINCPT,
f"{nc_prefix}_Max_P": neutral_cone_isolation_alg.MAXCP,
f"{nc_prefix}_Max_PT": neutral_cone_isolation_alg.MAXCPT,
f"{nc_prefix}_Min_P": neutral_cone_isolation_alg.MINCP,
f"{nc_prefix}_Min_PT": neutral_cone_isolation_alg.MINCPT,
}
ConeIsolationVariables.update(
ParticleIsolation(
isolation_alg=charged_cone_isolation_alg,
sumcone_invalid_value=sumcone_invalid_value,
array_indx_name=array_indx_name,
).functor_dict
)
ConeIsolationVariables.update(
ParticleIsolation(
isolation_alg=neutral_cone_isolation_alg,
sumcone_invalid_value=sumcone_invalid_value,
array_indx_name=array_indx_name,
).functor_dict
)
return FunctorCollection(ConeIsolationVariables)
[docs]
def VertexIsolation(
*,
isolation_alg: VertexAndConeIsolation,
) -> FunctorCollection:
"""
Candidate-level collection of functors on vertex isolation, using information from relations between sets of particles determined by a given criterion.
It accesses to `VertexAndConeIsolation` class information.
Note:
Variables will have the following naming convention:
`VTXISO_<isolation_alg.name>_<var_name>`
Args:
isolation_alg (VertexAndConeIsolation): An instance of the `VertexAndConeIsolation` helper
class that is used to retrieve isolation variables.
name (string, optional): string that helps in distinguishing variables
according to some criterion (for example if the isolation criterion are applied to charged or neutral particles).
Defaults to "".
Output variables:
- `_NParts` : `isolation_alg.MULT` - Number of particles in CHI2 window
- `_Smallest_DELTACHI2` : `isolation_alg.Smallest_DELTACHI2` - smallest delta chi2 when adding one track
- `_Smallest_CHI2` : `isolation_alg.Smallest_CHI2` - smallest chi2 when adding one track
- `_Smallest_DELTACHI2_MASS` : `isolation_alg.Smallest_Mass_DELTACHI2` - mass of the candidate with the smallest delta chi2 when adding one track
Example::
import Functors as F
from FunTuple import FunTuple_Particles as Funtuple
from FunTuple.functorcollections import VertexIsolation
from IsolationTools import VertexAndConeIsolation
from PyConf.reading import get_particles
branches = {"B": "[B0 -> ([J/psi(1S) -> tau+ mu-]CC)(K*(892)0 -> K+ pi-)]CC"}
b2ksttaumu_data = get_particles(f"/Event/HLT2/{line}/Particles")
b_extra_one_track_cand = get_particles(f"/Event/HLT2/{line}/B_{tes_long_one_track_iso}/Particles")
b_extra_two_tracks_cand = get_particles(f"/Event/HLT2/{line}/B_{tes_long_two_tracks_iso}/Particles")
vertex_isolation_b_one_track = VertexAndConeIsolation(
name = 'OneTrack',
reference_particles = b2ksttaumu_data,
related_particles = b_extra_one_track_cand,
cut = (F.CHI2@F.F.FORWARDARG1<9.))
vertex_isolation_b_two_tracks = VertexAndConeIsolation(
name = 'TwoTracks',
reference_particles = b2ksttaumu_data,
related_particles = b_extra_two_tracks_cand,
cut = (F.CHI2@F.F.FORWARDARG1<15.))
variables_ot = FC.VertexVariables(isolation_alg=vertex_isolation_b_one_track)
variables_tt = FC.VertexVariables(isolation_alg=vertex_isolation_b_two_tracks)
tuple = Funtuple(name="MyTuple",
fields=branches,
variables=variables_ot+variables_tt,
...)
"""
prefix = "VTXISO"
if isolation_alg.name:
prefix += f"_{isolation_alg.name}"
VertexIsolationVariables = {
# Number of particles in CHI2 window
f"{prefix}_NParts": isolation_alg.MULT,
# smallest delta chi2 when adding one track
f"{prefix}_Smallest_DELTACHI2": isolation_alg.Smallest_DELTACHI2,
# smallest chi2 when adding one track
f"{prefix}_Smallest_CHI2": isolation_alg.Smallest_CHI2,
# mass of the candidate with the smallest delta chi2 when adding one track
f"{prefix}_Smallest_DELTACHI2_MASS": isolation_alg.Smallest_Mass_DELTACHI2,
}
return FunctorCollection(VertexIsolationVariables)
[docs]
def NeutralCaloInfo(extra_info: bool = False) -> FunctorCollection:
"""
Candidate-level collection of functors on neutral calorimeter hypotheses information.
Useful for example for neutral particle reconstruction efficiency or neutral PID studies.
Args:
extra_info (bool, optional): If True, adds Neutral CellIDs bitwise information, row and column.
Output variables:
- `CaloTrackMatchChi2` : 2D chi2 for Track-CaloCluster matching
- `CaloNeutralShowerShape` : 2nd order moment of the cluster
- `CaloClusterMass` : CALO cluster Mass obtained from MergedPi0Alg
- `CaloNeutralEcalEnergy` : The ECAL cluster energy associated to the neutral CaloHypo
- `CaloNeutral1To9EnergyRatio` : Fraction of highest 1x1 cell to 3x3 cells forming the cluster for the Neutral CaloHypo
- `CaloNeutral4To9EnergyRatio` : Fraction of highest 2x2 cell to 3x3 cells forming the cluster for the Neutral CaloHypo
- `CaloNeutralHcal2EcalEnergyRatio` : The Hcal/Ecal energy ratio associated to the neutral calorimeter hypotheses
- `CaloNumSaturatedCells` : Number of saturated cells
- `CaloNeutralArea` : Area code of Neutral CellID
If `extra_info` is set to `True`, it adds also:
- `CaloNeutralID` : Bitwise information (32 bits) of all Neutral seed CellIDs
- `CaloNeutraRow` : Row code of Neutral CellID
- `CaloNeutraCol` : Column code of Neutral CellID
"""
_NeutralCaloInfo = {
"CaloTrackMatchChi2": F.CLUSTERMATCH_CHI2,
"CaloNeutralShowerShape": F.CALO_NEUTRAL_SHOWER_SHAPE,
"CaloClusterMass": F.CALO_CLUSTER_MASS,
"CaloNeutralEcalEnergy": F.CALO_NEUTRAL_ECAL_ENERGY,
"CaloNeutral1To9EnergyRatio": F.CALO_NEUTRAL_1TO9_ENERGY_RATIO,
"CaloNeutral4To9EnergyRatio": F.CALO_NEUTRAL_4TO9_ENERGY_RATIO,
"CaloNeutralHcal2EcalEnergyRatio": F.CALO_NEUTRAL_HCAL2ECAL_ENERGY_RATIO,
"CaloNumSaturatedCells": F.CALO_NUM_SATURATED_CELLS,
"CaloNeutralArea": F.CALOCELLID_AREA @ F._CALO_NEUTRAL_ID,
}
if extra_info:
_NeutralCaloInfo.update(
{
"CaloNeutralID": F.CALO_NEUTRAL_ID,
"CaloNeutralRow": F.CALOCELLID_ROW @ F._CALO_NEUTRAL_ID,
"CaloNeutralCol": F.CALOCELLID_COLUMN @ F._CALO_NEUTRAL_ID,
}
)
return FunctorCollection(_NeutralCaloInfo)
[docs]
def ChargedCaloInfo(extra_info: bool = False) -> FunctorCollection:
"""
Candidate-level collection of functors on charged calorimeter hypotheses information.
Useful for example for charged particle reconstruction efficiency or charged PID studies.
Args:
extra_info (bool, optional): If True, adds extra information, including CaloCluster and CaloHypo (photon and electron) CellIDs related ones.
Output variables:
- `INECAL` : In Ecal acceptance
- `INHCAL` : In Hcal acceptance
- `INBREM` : In Brem acceptance: the track extrapolation from upstream of the magnet is in Ecal acceptance.
- `HASBREM` : Has non-zero brem momentum-recovery energy available
- `HASBREMADDED` : Has non-zero brem momentum-recovery energy added by BremAdder
- `BREMPIDE` : Brem-based DLL for electron-ID
- `ECALPIDE` : Ecal-based DLL for electron-ID
- `ECALPIDMU` : Ecal-based DLL for muon-ID
- `HCALPIDE` : Hcal-based DLL for electron-ID
- `HCALPIDMU` : Hcal-based DLL for muon-ID
- `CLUSTERAREA` : Area code of CaloCluster CellID
- `ELECTRONAREA` : Area code of CaloHypo (electron) CellID
- `BREMHYPOAREA` : Area code of CaloHypo (photon) CellID
If `extra_info` is set to `True`, it adds also:
- `CLUSTERID` : All significant bits representation of CellID (32bits), i.e. CellID.all(), associated to CaloCluster of track match
- `CLUSTERROW` : Row code of CaloCluster CellID
- `CLUSTERCOL` : Column code of CaloCluster CellID
- `CLUSTERMATCH_CHI2` : 2D chi2 for Track/CaloCluster matching (neutral/charged)
- `ELECTRONSHOWEREOP` : Ratio of electron energy/momentum with track-based cell selection
- `ELECTRONSHOWERDLL` : Summed per-cell E/p DLL (electron versus pion) with track-based cell selection and energy estimation
- `ELECTRONMATCH_CHI2` : 3D chi2 for Track/CaloHypo(e) matching (charged)
- `ELECTRONENERGY` : Cluster energy associated to CaloHypo (charged)
- `ELECTRONID` : All significant bits representation of CellID (32bits), i.e. CellID.all(), associated to CaloHypo seed (electron hypo).
- `ELECTRONROW` : Row code of CaloHypo (electron) CellID
- `ELECTRONCOL` : Column code of CaloHypo (electron) CellID
- `BREMENERGY` : Brem momentum-recovery energy
- `BREMBENDCORR` : Correction factor accounting for bending biases in track due to brem
- `BREMHYPOID` : All significant bits representation of CellID (32bits), i.e. CellID.all(), for CaloHypo (photon) associated to track for brem recovery
- `BREMHYPOROW` : Row code of CaloHypo (photon) CellID
- `BREMHYPOCOL` : Column code of CaloHypo (photon) CellID
- `BREMHYPOMATCH_CHI2` : 2D chi2 of CaloHypo (photon) associated to track for brem recovery
- `BREMHYPOENERGY` : Energy of CaloHypo (photon) associated to track for brem recovery
- `BREMHYPODELTAX` : Test statistic of being first-state like of CaloHypo (photon) for brem recovery
- `BREMTRACKBASEDENERGY` : Track-based brem energy determination
- `HCALEOP` : Hcal energy deposit over momentum (track)
"""
_ChargedCaloInfo = {
"INECAL": F.INECAL,
"INHCAL": F.INHCAL,
"INBREM": F.INBREM,
"HASBREM": F.HASBREM,
"HASBREMADDED": F.HASBREMADDED,
"BREMENERGY": F.BREMENERGY,
"BREMPIDE": F.BREMPIDE,
"ECALPIDE": F.ECALPIDE,
"ECALPIDMU": F.ECALPIDMU,
"HCALPIDE": F.HCALPIDE,
"HCALPIDMU": F.HCALPIDMU,
"CLUSTERAREA": F.CALOCELLID_AREA @ F._CLUSTERID,
"ELECTRONAREA": F.CALOCELLID_AREA @ F._ELECTRONID,
"BREMHYPOAREA": F.CALOCELLID_AREA @ F._BREMHYPOID,
}
if extra_info:
_ChargedCaloInfo.update(
{
"CLUSTERID": F.CLUSTERID,
"CLUSTERROW": F.CALOCELLID_ROW @ F._CLUSTERID,
"CLUSTERCOL": F.CALOCELLID_COLUMN @ F._CLUSTERID,
"CLUSTERMATCH_CHI2": F.CLUSTERMATCH_CHI2,
"ELECTRONSHOWEREOP": F.ELECTRONSHOWEREOP,
"ELECTRONSHOWERDLL": F.ELECTRONSHOWERDLL,
"ELECTRONMATCH_CHI2": F.ELECTRONMATCH_CHI2,
"ELECTRONENERGY": F.ELECTRONENERGY,
"ELECTRONID": F.ELECTRONID,
"ELECTRONROW": F.CALOCELLID_ROW @ F._ELECTRONID,
"ELECTRONCOL": F.CALOCELLID_COLUMN @ F._ELECTRONID,
"BREMBENDCORR": F.BREMBENDCORR,
"BREMHYPOID": F.BREMHYPOID,
"BREMHYPOROW": F.CALOCELLID_ROW @ F._BREMHYPOID,
"BREMHYPOCOL": F.CALOCELLID_COLUMN @ F._BREMHYPOID,
"BREMHYPOMATCH_CHI2": F.BREMHYPOMATCH_CHI2,
"BREMHYPOENERGY": F.BREMHYPOENERGY,
"BREMHYPODELTAX": F.BREMHYPODELTAX,
"BREMTRACKBASEDENERGY": F.BREMTRACKBASEDENERGY,
"HCALEOP": F.HCALEOP,
}
)
return FunctorCollection(_ChargedCaloInfo)
[docs]
def ParticleID(extra_info: bool = False) -> FunctorCollection:
"""
Candidate-level collection of functors on global particle identification variables.
Args:
extra_info (bool, optional) : If True, store DLLs in addition. Defaults to False.
Output variables:
- `PARTICLE_ID` : `F.PARTICLE_ID`
- `PROBNN_[D,E,GHOST,K,MU,P,PI]` : `F.PROBNN_[D,E,GHOST,K,MU,P,PI]`
"""
pid_vars = {
"PARTICLE_ID": F.PARTICLE_ID,
"PROBNN_D": F.PROBNN_D,
"PROBNN_E": F.PROBNN_E,
"PROBNN_GHOST": F.PROBNN_GHOST,
"PROBNN_K": F.PROBNN_K,
"PROBNN_MU": F.PROBNN_MU,
"PROBNN_P": F.PROBNN_P,
"PROBNN_PI": F.PROBNN_PI,
}
if extra_info:
pid_vars.update(
{
"PID_E": F.PID_E,
"PID_K": F.PID_K,
"PID_MU": F.PID_MU,
"PID_P": F.PID_P,
"PID_PI": F.PID_PI,
}
)
return FunctorCollection(pid_vars)
def LHCInfo(
extra_info: bool = False,
) -> FunctorCollection:
"""
Event-level collection providing the LHC conditions fill number, beam energy and LHCb clock phase.
Args:
extra_info (bool, optional) : If True, stores LHCEnergy and LHCbClockPhase in addition to the LHC FillNumber.
Defaults to False.
Output variables:
- `FillNumber` : `F.VALUE_OR(-1) @ F.FILL_NUMBER()`
If `extra_info`:
- `LHCEnergy` : `F.VALUE_OR(-1) @ F.LHC_ENERGY()`
- `LHCbClockPhase`: `F.VALUE_OR(-1) @ F.LHCB_CLOCKPHASE()`
Example::
import FunTuple.functorcollections as FC
from DaVinci import options
variables = FC.LHCInfo()
# ...
tuple = FuntupleEvent(name="MyTuple",
fields=...,
variables=variables,
...)
"""
lhc = {
"FillNumber": F.VALUE_OR(-1) @ F.FILL_NUMBER(),
}
if extra_info:
lhc.update(
{
"LHCEnergy": F.VALUE_OR(-1) @ F.LHC_ENERGY(),
"LHCbClockPhase": F.VALUE_OR(-1) @ F.LHCB_CLOCKPHASE(),
}
)
return FunctorCollection(lhc)
[docs]
def RecSummary(
detector_info: bool = False,
) -> FunctorCollection:
"""
Event-level collection providing access to the RecSummary record.
Provides event multiplicities.
The list of quantities is at https://gitlab.cern.ch/lhcb/LHCb/-/blob/master/Event/RecEvent/include/Event/RecSummary.h.
Args:
detector_info (bool, optional) : If True, stores hit multiplicities in subdetectors.
Defaults to False.
Output variables:
- `nPVs`
- `n[Long,Downstream,Upstream,Velo,T,Back]Tracks`
- `[e,h]CalTot`
- `nRich[1,2]Hits`
- `n[VP,UT,FT,Ecal]Clusters`
Example::
import FunTuple.functorcollections as FC
from DaVinci import options
variables = FC.RecSummary()
# ...
tuple = FuntupleEvent(name="MyTuple",
fields=...,
variables=variables,
...)
"""
reco_values = [
"nPVs", # 0, // Number of reconstructed Primary Vertices
"nLongTracks", # 10, // Number of reconstructed Long Tracks
"nDownstreamTracks", # 11, // Number of reconstructed Downstream Tracks
"nUpstreamTracks", # 12, // Number of reconstructed Upstream Tracks
"nVeloTracks", # 13, // Number of reconstructed VELO Tracks
"nTTracks", # 14, // Number of reconstructed T-station Tracks
"nBackTracks", # 15, // Number of reconstructed VELO backwards Tracks
# "nTracks", # 16, // Number of reconstructed Tracks
# "nGhosts", # 17, // Number of identified ghost Tracks
"eCalTot", # 71, // Total energy deposit in ecal
"hCalTot", # 72, // Total energy deposit in hcal
]
det_values = [
"nRich1Hits", # 20, // Number of hits in RICH1
"nRich2Hits", # 21, // Number of hits in RICH2
# "nVeloClusters", # 30, // Number of Velo clusters
"nVPClusters", # 31, // Number of VP clusters
# "nITClusters", # 40, // Number of IT clusters
# "nTTClusters", # 50, // Number of TT clusters
"nUTClusters", # 51, // Number of UT clusters
# "nOTClusters", # 60, // Number of Outer Tracker clusters
"nFTClusters", # 41, // Number of FT clusters
# "nSPDhits", # 70, // Number of SPD hits
"nEcalClusters", # 73, // Number of Ecal clusters
# these are empty.
# "nMuonCoordsS0", # 80, // Number of Coords in Muon Station 0
# "nMuonCoordsS1", # 91, // Number of Coords in Muon Station 1
# "nMuonCoordsS2", # 92, // Number of Coords in Muon Station 2
# "nMuonCoordsS3", # 93, // Number of Coords in Muon Station 3
# "nMuonCoordsS4", # 94, // Number of Coords in Muon Station 4
# "nMuonTracks", # 95, // Number of Muon Tracks
]
rec_summary = get_rec_summary()
rec_collection = {}
for t in reco_values:
rec_collection[t] = F.VALUE_OR(-1) @ F.RECSUMMARY_INFO(rec_summary, t)
if detector_info:
for t in det_values:
rec_collection[t] = F.VALUE_OR(-1) @ F.RECSUMMARY_INFO(rec_summary, t)
return FunctorCollection(rec_collection)
[docs]
def FlavourTaggingResults(
all_taggers: DataHandle,
tagger_name_list: list = [
"Run2_SSPion",
"Run2_SSKaon",
"Run2_SSProton",
"Run2_OSKaon",
"Run2_OSElectron",
"Run2_OSMuon",
"Run2_OSVertexCharge",
],
prefix: str = "",
) -> FunctorCollection:
"""
Candidate-level collection of functors that access FlavourTagging results.
Args:
all_taggers: a FlavourTags object that holds the information of all relevant taggers
tagger_name_list (list, optional): a list of all the names (as specified in ``LHCb::Tagger``)
of the taggers that are requested as branches. Default is all Run2 taggers.
prefix (str, optional): additional prefix to output functors. Defaults to "".
Output variables:
- `_Dec` : `F.TaggingDecision` for each tagger specified
- `_Omega` : `F.TaggingMistag` for each tagger specified
- `_MVA` : `F.TaggingMVAOutput` for each tagger specified
Example::
from FunTuple import FunTuple_Particles as Funtuple
import FunTuple.functorcollections as FC
from Hlt2Conf.flavourTagging import run2_all_taggers
line_data = get_particles(f"/Event/HLT2/{line_name}/Particles")
all_tagging = run2_all_taggers(line_data)
variables = {"B": FC.FlavourTaggingResults(all_tagging) }
tuple = Funtuple(
name="Tuple",
tuple_name="DecayTree",
fields=fields,
variables=variables,
inputs=line_data,
)
"""
ft_results = {}
for tagger_name in tagger_name_list:
ft_results[f"{prefix}{tagger_name}_Dec"] = F.TaggingDecision(
all_taggers.OutputFlavourTags, tagger_name
)
ft_results[f"{prefix}{tagger_name}_Omega"] = F.TaggingMistag(
all_taggers.OutputFlavourTags, tagger_name
)
ft_results[f"{prefix}{tagger_name}_MVA"] = F.TaggingMVAOutput(
all_taggers.OutputFlavourTags, tagger_name
)
return FunctorCollection(ft_results)
def SMOGInfo() -> FunctorCollection:
"""
Event-level collection providing the SMOG conditions injection mode, injected gas and stable injection flag.
Output variables:
- `SMOGInjectionMode`
- `SMOGInjectedGas`
- `SMOGInjectionStable`
Example::
import FunTuple.functorcollections as FC
from DaVinci import options
variables = FC.SMOGInfo()
# ...
tuple = FuntupleEvent(name="MyTuple",
fields=...,
variables=variables,
...)
"""
SMOG = {
"SMOGInjectionMode": F.VALUE_OR(-1) @ F.SMOG_INJECTION_MODE(),
"SMOGInjectedGas": F.VALUE_OR(-1) @ F.SMOG_INJECTED_GAS(),
"SMOGInjectionStable": F.VALUE_OR(-1) @ F.SMOG_STABLE_INJECTION(),
}
return FunctorCollection(SMOG)
def AllPrimaryVertexInfo(
pvs: DataHandle, extra_info: bool = False
) -> FunctorCollection:
"""
Event-level information about all of the reconstructed primary vertices.
Should be used as part of the event variables in FunTuple.
Output variables:
- `ALLPV_[X,Y,Z]` : `F.ALLPV[X,Y,Z](pvs)`
- `ALLPV_[X,Y,Z]ERR`
"""
All_PV_information = {
"ALLPV_X": F.ALLPVX(pvs),
"ALLPV_Y": F.ALLPVY(pvs),
"ALLPV_Z": F.ALLPVZ(pvs),
"ALLPV_nPVs": F.SIZE(pvs),
}
if extra_info:
All_PV_information["ALLPV_XERR"] = F.MAP(
F.SQRT @ F.CALL(0, 0) @ F.POS_COV_MATRIX
) @ F.TES(pvs)
All_PV_information["ALLPV_YERR"] = F.MAP(
F.SQRT @ F.CALL(1, 1) @ F.POS_COV_MATRIX
) @ F.TES(pvs)
All_PV_information["ALLPV_ZERR"] = F.MAP(
F.SQRT @ F.CALL(2, 2) @ F.POS_COV_MATRIX
) @ F.TES(pvs)
return FunctorCollection(All_PV_information)