###############################################################################
# (c) Copyright 2021 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. #
###############################################################################
"""Wrappers around ThOr selection framework configurables."""
from PyConf import configurable
from PyConf.Algorithms import (
ChargedBasicsFilter, ParticleRangeFilter, PrFilter__PV, TwoBodyCombiner,
ThreeBodyCombiner, FourBodyCombiner, FiveBodyCombiner, SixBodyCombiner,
ParticleContainerMerger, MCParticleRangeFilter)
from Moore.persistence.particle_moving import is_particle_producer
from RecoConf.reconstruction_objects import make_pvs
#: Tag to select v1 Particle usage
PARTICLE_V1 = "v1"
#: Tag to select v2 Particle usage
PARTICLE_V2 = "v2"
#: Filter algorithms to be used depending on the input Particle API
PARTICLE_FILTERS = {
PARTICLE_V1: ParticleRangeFilter,
PARTICLE_V2: ChargedBasicsFilter,
}
#: N-body combiner algorithms to be used depending on the input Particle API
PARTICLE_COMBINERS = {
PARTICLE_V1: {
2: TwoBodyCombiner,
3: ThreeBodyCombiner,
4: FourBodyCombiner,
5: FiveBodyCombiner,
6: SixBodyCombiner,
},
}
@configurable
def thor_backend(particle_api=PARTICLE_V1):
"""Return the C++ Particle API to be used by selection algorithms.
ThOr functors can operate on both 'v1' Particle objects (used in Runs 1
and 2) and on 'v2' Particle objects (used in Run 3) however the
algorithms which use the functors are specific on the input Particle
type. Helper functions in the `algorithms_thor` module use this function
to determine what algorithms to instantiate.
Raises:
ValueError: If an invalid API value is given.
"""
if particle_api not in {PARTICLE_V1, PARTICLE_V2}:
raise ValueError("Invalid Particle API {!r}".format(particle_api))
return particle_api
def _validate_type(data_handle):
"""Validate that the data's type matches the current Particle API value.
Raises:
TypeError: If `data_handle.type` does not implement the current API.
"""
t = data_handle.type
producer = data_handle.producer
valid_v1 = thor_backend() == PARTICLE_V1 and is_particle_producer(producer)
valid_v2 = thor_backend() == PARTICLE_V2 and t in {"LHCb::v2::Particle"}
if not (valid_v1 or valid_v2):
raise TypeError(
f"Data handle {data_handle} of type {t!r} is not appropriate for the current ThOr backend {thor_backend()!r}"
)
[docs]def ParticleFilter(Input, Cut, **kwargs):
"""Return the output of a particle filter.
Arguments:
Input (DataHandle): Particle container to filter.
Cut: ThOr functor expression.
**kwargs: Passed to the algorithm instantiation.
It can be used to set a name to the algorithm during development, but not in production code (see Moore#378 and Moore#380)
"""
_validate_type(Input)
cls = PARTICLE_FILTERS[thor_backend()]
return cls(Input=Input, Cut=Cut, **kwargs).Output
def PVFilter(Input, Cut, **kwargs):
"""Return the output of a PV filter.
Arguments:
Input (DataHandle): PV container to filter.
Cut: ThOr functor expression.
**kwargs: Passed to the algorithm instantiation.
It can be used to set a name to the algorithm during development, but not in production code (see Moore#378 and Moore#380)
"""
return PrFilter__PV(Input=Input, Cut=Cut, **kwargs).Output
[docs]def ParticleCombiner(Inputs, **kwargs):
"""Return the output of an N-body particle combiner.
Arguments:
Inputs (list of DataHandle): The length ``N`` of this list is used to
instantiate the appropriate ``NBodyParticleCombiner``.
**kwargs: Passed to the algorithm instantiation.
"""
for inp in Inputs:
_validate_type(inp)
# Dispatch to the combiner for the given multiplicity
combiners = PARTICLE_COMBINERS[thor_backend()]
combiner = combiners[len(Inputs)]
input_kwargs = {
f"Input{idx}": data
for idx, data in enumerate(Inputs, start=1)
}
if "PrimaryVertices" not in kwargs:
kwargs["PrimaryVertices"] = make_pvs()
return combiner(**input_kwargs, **kwargs).OutputParticles
def MCParticleFilter(Input, Cut, **kwargs):
"""Return the output of an MCparticle filter.
Arguments:
Input (DataHandle/KeyedContainer): MCParticle container to filter
Cut: ThOr functor expression.
**kwargs: Passed to the algorithm instantiation.
It can be used to set a name to the algorithm during development, but not in production code (see Moore#378 and Moore#380)
"""
return MCParticleRangeFilter(Input=Input, Cut=Cut, **kwargs).Output
# moved from Hlt2Conf.algorithms.
def ParticleContainersMerger(inputs, name='ParticleContainersMerger_{hash}'):
#"""Merge input Particle containers to a single output container."""
return ParticleContainerMerger(
InputContainers=inputs, name=name).OutputContainer