###############################################################################
# (c) Copyright 2019-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. #
###############################################################################
"""Define the HLT2 track reconstruction outputs for use by lines."""
from GaudiKernel.SystemOfUnits import mm
from functools import partial
from typing import Any, Callable
from PyConf import configurable
from PyConf.application import default_raw_banks
from PyConf.dataflow import DataHandle
from PyConf.utilities import ConfigurationError, DISABLE_TOOL
from PyConf.packing import persistable_location, persistable_locations
from PyConf.Algorithms import (
PrForwardTrackingVelo, PrForwardTracking, PrVeloUT, PrHybridSeeding,
PrMatchNN, PrLongLivedTracking, TrackBestTrackCreator, PrResidualPrUTHits,
PrResidualSeedingLong, PrResidualVeloTracks, PrResidualSciFiHits,
fromPrForwardTracksV1Tracks, fromPrForwardTracksFromVeloUTV1Tracks,
fromPrMatchTracksV1Tracks, fromPrSeedingTracksV1Tracks,
fromPrDownstreamTracksV1Tracks, fromPrVeloTracksV1Tracks,
TrackContainerSplitter, TrackCloneKiller, TrackContainersMerger,
TrackEventFitter, TracksToSelection, TracksEmptyProducer,
TrackSelectionMerger, TrackSelectionToContainer,
LHCb__Converters__Track__SOA__fromV1Track as TrackSOAFromV1,
LHCb__Converters__Track__SOA__fromSharedV1Track as TrackSOAFromSharedV1,
PrKalmanFilter, PrKalmanFilter_noUT, PrKalmanFilter_Seed,
PrKalmanFilter_Velo, PrKalmanFilter_Downstream, PrKalmanFilter_Upstream,
VeloHeavyFlavourTrackFinder, PrStoreSciFiHits, PrStoreUTHit,
PrStoreUTHitEmptyProducer)
from PyConf.Tools import (
PrAddUTHitsTool,
PrIgnoreUTHitsTool,
UpgradeGhostId,
UpgradeGhostId_noUT,
TrackMasterExtrapolator,
TrackLinearExtrapolator,
GhostProbabilityTool_Long_noUT,
GhostProbabilityTool_Ttrack,
GhostProbabilityTool_Velo,
SimplifiedMaterialLocator,
TrackMasterFitter,
MeasurementProvider,
VPMeasurementProvider,
UTMeasurementProvider,
FTMeasurementProvider,
)
from RecoConf.core_algorithms import make_unique_id_generator
from RecoConf.legacy_rec_hlt1_tracking import (
make_FTRawBankDecoder_clusters, make_PrStorePrUTHits_hits,
make_VeloClusterTrackingSIMD_hits, make_legacy_rec_hlt1_tracks,
make_legacy_rec_hlt1_fitted_tracks, all_velo_track_types,
make_VPClus_location_and_offsets)
import Functors as F
def make_VPClus_hits():
return make_VPClus_location_and_offsets()["Location"]
[docs]@configurable
def make_PrStoreUTHit_hits(make_raw=default_raw_banks,
isCluster=True,
positionMethod="GeoWeighting"):
"""Decodes UT hits from raw data.
Args:
make_raw (DataHandle): RawEventLocation for VeloClusterTrackingSIMD, defaults to `default_raw_event <PyConf.application.default_raw_event>`.
positionMethod (str): available options "MaxAdc", "AdcWeighting", "GeoWeighting", defaults to "GeoWeighting"
Returns:
DataHandle: PrStoreUTHit's UTHitsLocation.
"""
return PrStoreUTHit(
RawBanks=make_raw("UT"),
isCluster=isCluster,
positionMethod=positionMethod).UTHitsLocation
[docs]@configurable
def make_PrStoreUTHit_empty_hits():
"""Creates an empty container of UT hits, used for the no UT scenario.
Returns:
DataHandle: PrStoreUTHitEmptyProducer' Output.
"""
return PrStoreUTHitEmptyProducer().Output
[docs]@configurable
def get_global_materiallocator(materiallocator=SimplifiedMaterialLocator):
"""Returns instance of (Simplified/Detailed)MaterialLocator.
Args:
materiallocator: MaterialLocator tool, defaults to SimplifiedMaterialLocator.
Returns:
Instance of input tool.
"""
return materiallocator()
[docs]@configurable
def make_PrStoreSciFiHits_hits(make_ft_clusters=make_FTRawBankDecoder_clusters,
disabled_layers=[]):
"""Makes SciFi hits from FTLiteClusters taking the detector geometry into accout. This is the SoA
replacement for `make_PrStoreFTHit_hits`.
Args:
make_ft_clusters (DataHandle): maker of FT clusters, defaults to `make_FTRawBankDecoder_clusters`.
Returns:
DataHandle: PrStoreSciFiHits's Output.
"""
my_disabled_layers = [j in disabled_layers
for j in range(12)] # 12 = n Layers in SciFi
return PrStoreSciFiHits(
HitsLocation=make_ft_clusters(),
LayerMasks=tuple(my_disabled_layers)).Output
[docs]@configurable
def get_global_measurement_provider(velo_hits=make_VPClus_hits,
ut_hits=make_PrStoreUTHit_hits,
ft_clusters=make_FTRawBankDecoder_clusters,
vp_provider=VPMeasurementProvider,
ut_provider=UTMeasurementProvider,
ft_provider=FTMeasurementProvider,
muon_provider=DISABLE_TOOL,
ignoreVP=False,
ignoreUT=False,
ignoreFT=False,
ignoreMuon=True):
"""Returns instance of MeasurementProvider given VP, UT and FT provider and cluster/hits.
Args:
velo_clusters (DataHandle): maker of velo clusters, defaults to `make_VPClus_location_and_offsets`.
ut_clusters (DataHandle): maker of UT hits, defaults to `make_PrStoreUTHit_hits`.
ft_clusters (DataHandle): maker of FT clusters, defaults to `make_FTRawBankDecoder_clusters`.
vp_provider: MeasurementProvider tool for VeloPix, defaults to VPMeasurementProvider.
ut_provider: MeasurementProvider tool for UT, defaults to UTMeasurementProvider.
ft_provider: MeasurementProvider tool for SciFi, defaults to FTMeasurementProvider.
muon_provider: MeasurementProvider tool for Muon detector, disabled by default
IngoreMuon: defaults to true, ignores muon measurement provider tool
Returns:
Instance of a global MeasurementProvider.
"""
return MeasurementProvider(
VPProvider=vp_provider(ClusterLocation=velo_hits()),
UTProvider=ut_provider(ClusterLocation=ut_hits()),
FTProvider=ft_provider(ClusterLocation=ft_clusters()),
MuonProvider=muon_provider(),
IgnoreVP=ignoreVP,
IgnoreUT=ignoreUT,
IgnoreFT=ignoreFT,
IgnoreMuon=ignoreMuon)
[docs]@configurable
def get_track_master_fitter(
get_materiallocator=get_global_materiallocator,
get_measurement_provider=get_global_measurement_provider,
MaxUpdateTransports=10):
"""Returns instance of TrackMasterFitter given MaterialLocator and MeasurementProvider.
Args:
get_materiallocator: MaterialLocator tool, defaults to `get_global_materiallocator`.
get_measurement_provider: MeasurementProvider tool, defaults to `get_global_measurement_provider`.
Returns:
Instance of TrackMasterFitter.
"""
materialLocator = get_materiallocator()
return TrackMasterFitter(
MeasProvider=get_measurement_provider(),
MaterialLocator=materialLocator,
Extrapolator=TrackMasterExtrapolator(MaterialLocator=materialLocator),
MaxUpdateTransports=MaxUpdateTransports)
@configurable
def get_track_master_extrapolator(
get_materiallocator=get_global_materiallocator):
return TrackMasterExtrapolator(MaterialLocator=get_materiallocator())
[docs]@configurable
def make_TrackEventFitter_fitted_tracks(tracks,
fitter_tool=get_track_master_fitter):
"""Fits tracks with TrackEventFitter.
Args:
tracks (dict of dicts): input tracks to TrackEventFitter, needs``["Forward"]["v1"]`` tracks, e.g. from make_legacy_rec_hlt1_tracks.
fitter_tool: instance of track fitting tool, defaults to `get_track_master_fitter`.
Returns:
A dict mapping fitted HLT1 forward v1 tracks to ``'v1'``.
"""
return {
"v1":
TrackEventFitter(
TracksInContainer=tracks["Forward"]["v1"],
Fitter=fitter_tool()).TracksOutContainer
}
[docs]@configurable
def make_PrForwardTracking_tracks(
input_tracks,
make_ft_hits=make_PrStoreSciFiHits_hits,
ut_hits_tool=get_global_ut_hits_tool,
add_ut_hits=False,
momentum_window=False,
delta_quality=0.02,
):
""" Makes forward tracks with PrForwardTracking (long tracks from upstream seeds).
Args:
input_tracks (dict): upstream tracks, needs ``'Pr'`` tracks, e.g. from `all_upstream_track_types <RecoConf.legacy_rec_hlt1_tracking.all_velo_track_types>`.
make_ft_hits (DataHandle): maker of FT hits, defaults to `make_PrStoreSciFiHits_hits <RecoConf.legacy_rec_hlt1_tracking.make_PrStoreSciFiHits_hits>`.
ut_hits_tool: tool providing UT hits, defaults to `get_global_ut_hits_tool`.
add_ut_hits: switch for adding UT hits, defaults to `False`.
momentum_window: switch for momentum guided search window, defaults to `False`.
delta_quality: defines the maximal difference in track quality between candidates, defaults to `0.02`.
Returns:
DataHandle: PrForwardTracking's OutputTracks.
Note:
PrForwardTracking's default does not use a momentum estimate, this is overwritten here.
"""
return PrForwardTracking(
SciFiHits=make_ft_hits(),
InputTracks=input_tracks["Pr"],
AddUTHitsToolName=ut_hits_tool(enable=add_ut_hits),
UseMomentumSearchWindow=momentum_window,
DeltaQuality=delta_quality,
).OutputTracks
[docs]@configurable
def make_PrForwardTrackingVelo_tracks(input_tracks,
make_ft_hits=make_PrStoreSciFiHits_hits,
ut_hits_tool=get_global_ut_hits_tool,
add_ut_hits=True):
""" Makes forward tracks for HLT2 (long tracks from Velo seeds) with PrForwardTrackingVelo.
Args:
input_tracks (dict): velo tracks, needs ``'Pr'`` tracks, e.g. from
`all_velo_track_types <RecoConf.legacy_rec_hlt1_tracking.all_velo_track_types>`.
make_ft_hits (DataHandle): maker of FT hits, defaults to
`make_PrStoreSciFiHits_hits <RecoConf.legacy_rec_hlt1_tracking.make_PrStoreSciFiHits_hits>`.
ut_hits_tool: tool providing UT hits, defaults to `get_global_ut_hits_tool <RecoConf.hlt2_tracking.get_global_ut_hits_tool>`.
Returns:
DataHandle: PrForwardTrackingVelo's OutputTracks.
Note:
PrForwardTrackingVelo's defaults have been overridden in this maker with ``UseMomentumEstimate=False``.
"""
return PrForwardTrackingVelo(
InputTracks=input_tracks["Pr"],
SciFiHits=make_ft_hits(),
AddUTHitsToolName=ut_hits_tool(enable=add_ut_hits)).OutputTracks
[docs]def get_PrForwardTracksV1_converter(input_tracks, ancestor_tracks):
""" Makes the conversion from PrForwardTracks to v1 tracks.
Args:
input_tracks (DataHandle): Forward tracks, needs ``'Pr'`` tracks, e.g. from
`make_PrForwardTrackingVelo_tracks <RecoConf.hlt2_tracking.make_PrForwardTrackingVelo_tracks>`.
ancestor_tracks (dict): tracks the input_tracks were made from, e.g. from
`all_velo_track_types <RecoConf.legacy_rec_hlt1_tracking.all_velo_track_types>`.
Returns:
DataHandle: fromPrForwardTracksV1Tracks's OutputTracksLocation
"""
return fromPrForwardTracksV1Tracks(
InputTracksLocation=input_tracks,
VeloTracksLocation=ancestor_tracks["v1"]).OutputTracksLocation
[docs]def get_PrForwardTracksFromVeloUTV1_converter(input_tracks, ancestor_tracks):
""" Makes the conversion from PrForwardTracks to v1 tracks.
Args:
input_tracks (DataHandle): forward tracks, needs ``'Pr'`` tracks, e.g. from
`make_PrForwardTracking_tracks <RecoConf.hlt2_tracking.make_PrForwardTracking_tracks>`.
ancestor_tracks (dict): tracks the input_tracks were made from, e.g. from
`all_upstream_track_types <RecoConf.legacy_rec_hlt1_tracking.all_upstream_track_types>`.
Returns:
DataHandle: fromPrForwardTracksFromVeloUTV1Tracks's OutputTracksLocation
"""
return fromPrForwardTracksFromVeloUTV1Tracks(
InputTracksLocation=input_tracks,
UpstreamTracksLocation=ancestor_tracks["v1"]).OutputTracksLocation
[docs]@configurable
def make_PrVeloUTFilter_tracks(velo_tracks,
make_ut_hits=make_PrStorePrUTHits_hits):
""" Makes upstream tracks by filtering velo tracks by checking for UT hits.
Args:
velo_tracks (dict): velo tracks, needs ``'Pr'`` tracks, e.g. from `all_velo_track_types <RecoConf.legacy_rec_hlt1_tracking.all_velo_track_types>`.
make_ut_hits (DataHandle): UT hit maker, defaults to `make_PrStoreUTHit_hits <RecoConf.legacy_rec_hlt1_tracking.make_PrStoreUTHit_hits>`.
Returns:
DataHandle: PrVeloUT's OutputTracksName.
Note: The UT hits are not added to the track, momenta are only added if the fit of the track candidate succeeds.
Can be useful as input to `<RecoConf.hlt2_tracking.make_PrForwardTracking_tracks>`
"""
return PrVeloUT(
InputTracksName=velo_tracks["Pr"],
UTHits=make_ut_hits(),
MinPT=30,
MinMomentum=1200,
MinPTFinal=50,
MinMomentumFinal=1500,
LD3HitsMin=-1.0,
LD4HitsMin=-1.0,
FilterMode=True).OutputTracksName
[docs]@configurable
def all_hlt2_forward_track_types(
input_tracks,
make_forward_tracks=make_PrForwardTrackingVelo_tracks,
converter=get_PrForwardTracksV1_converter):
"""Helper function to get all types of HLT2 forward tracks.
Args:
input_tracks (dict): velo tracks, needs ``'v2'`` tracks, e.g. from `all_velo_track_types <RecoConf.legacy_rec_hlt1_tracking.all_velo_track_types>`.
make_forward_tracks (DataHandle): maker of forward tracks, defaults to `make_PrForwardTrackingVelo_tracks`.
Returns:
A dict mapping Pr, v1, v2 and v2Zip HLT2 forward tracks to ``'Pr'``, ``'v1'``, ``'v2'`` and ``'v2Zip'`` respectively.
"""
forward_tracks_pr = make_forward_tracks(input_tracks=input_tracks)
forward_tracks_v1 = converter(forward_tracks_pr, input_tracks)
return {
"Pr": forward_tracks_pr,
"v1": forward_tracks_v1,
}
[docs]@configurable
def make_PrHybridSeeding_tracks(make_ft_hits=make_PrStoreSciFiHits_hits):
"""Makes seed tracks with PrHybridSeeding [1]_.
Args:
make_ft_hits (DataHandle): maker of FT hits, defaults to `make_PrStoreSciFiHits_hits <RecoConf.legacy_rec_hlt1_tracking.make_PrStoreSciFiHits_hits>`.
Returns:
A dict mapping v1 and v2 SciFi seeding tracks to ``'v1'`` and ``'v2'`` respectively.
.. [1] https://cds.cern.ch/record/2027531/
"""
scifi_tracks_pr = PrHybridSeeding(FTHitsLocation=make_ft_hits()).OutputName
scifi_tracks_v1 = fromPrSeedingTracksV1Tracks(
InputTracksLocation=scifi_tracks_pr).OutputTracksLocation
return {"v1": scifi_tracks_v1, "Pr": scifi_tracks_pr}
@configurable
def make_seeding_tracks(make_seed_tracks=make_PrHybridSeeding_tracks):
return make_seed_tracks()
[docs]@configurable
def make_PrMatchNN_tracks(velo_tracks,
scifi_tracks,
ut_hits_tool=get_global_ut_hits_tool,
add_ut_hits=True):
"""Makes long tracks from SciFi seed tracks, velo tracks and UT hits using PrMatchNN.
Args:
velo_tracks (dict): velo tracks, needs ``'v2'`` tracks, e.g. from `all_velo_track_types <RecoConf.legacy_rec_hlt1_tracking.all_velo_track_types>`.
scifi_tracks (dict): SciFi seeding tracks, needs ``'v2'`` tracks, e.g. from `make_PrHybridSeeding_tracks`.
ut_hits_tool: tool providing UT hits, defaults to `get_global_ut_hits_tool <RecoConf.hlt2_tracking.get_global_ut_hits_tool>`.
Returns:
A dict mapping v1 and v2 long tracks from track-matching to ``'v1'`` and ``'v2'`` respectively.
"""
match_tracks_pr = PrMatchNN(
VeloInput=velo_tracks["Pr"],
SeedInput=scifi_tracks["Pr"],
AddUTHitsToolName=ut_hits_tool(enable=add_ut_hits)).MatchOutput
match_tracks_v1 = fromPrMatchTracksV1Tracks(
InputTracksLocation=match_tracks_pr,
VeloTracksLocation=velo_tracks["v1"],
SeedTracksLocation=scifi_tracks["v1"]).OutputTracksLocation
return {"v1": match_tracks_v1, "Pr": match_tracks_pr}
[docs]@configurable
def make_PrLongLivedTracking_tracks(scifi_tracks,
make_ut_hits=make_PrStorePrUTHits_hits):
"""Makes downstream tracks from SciFi seed tracks and UT hits using PrLongLivedTracking.
Args:
scifi_tracks (dict): SciFi seeding tracks, needs ``'Pr'`` tracks, e.g. from `make_PrHybridSeeding_tracks`.
make_ut_hits: algorithm putting UT hits into storage
Returns:
A dict mapping v1 and v2 downstream tracks to ``'v1'`` and ``'v2'`` respectively.
"""
downstream_tracks = PrLongLivedTracking(
InputLocation=scifi_tracks["Pr"], UTHits=make_ut_hits()).OutputLocation
downstream_tracks_v1 = fromPrDownstreamTracksV1Tracks(
InputTracksLocation=downstream_tracks,
SeedTracksLocation=scifi_tracks["v1"]).OutputTracksLocation
return {"Pr": downstream_tracks, "v1": downstream_tracks_v1}
[docs]@configurable
def make_PrKalmanFilter_tracks(
input_tracks: DataHandle,
hits_vp: DataHandle,
hits_ut: DataHandle,
hits_ft: DataHandle,
max_chi2ndof: float = 2.8,
max_chi2ndof_pre_outlier_removal: float = 20,
min_chi2_outlier: float = 9,
reference_extrapolator: Any = get_track_master_extrapolator,
name: str = None,
) -> DataHandle:
"""Configure the PrKalmanFilter to fit Long Tracks.
Args:
input_tracks (DataHandle): Location of PrLongTracks.
hits_vp (DataHandle): Location of Velo Hits.
hits_ut (DataHandle): Location of UT Hits.
hits_ft (DataHandle): Location of SciFi Hits.
max_chi2ndof (float, optional): Maximum chi2/ndof of the fitted track. Defaults to 2.8.
max_chi2ndof_pre_outlier_removal (float, optional): Maximum chi2/ndof of the fitted track before outlier removal. Defaults to 20.
min_chi2_outlier (float, optional): Minimum chi2 of node to be counted as outlier. Defaults to 9.
reference_extrapolator (Any, optional): Defaults to get_track_master_extrapolator.
name (str, optional): Defaults to None.
Returns:
DataHandle: Location of fitted tracks.
"""
return PrKalmanFilter(
Input=input_tracks,
HitsVP=hits_vp,
HitsUT=hits_ut,
HitsFT=hits_ft,
MaxChi2=max_chi2ndof,
MaxChi2PreOutlierRemoval=max_chi2ndof_pre_outlier_removal,
MinChi2Outlier=min_chi2_outlier,
ReferenceExtrapolator=reference_extrapolator(),
name=name,
InputUniqueIDGenerator=make_unique_id_generator(),
).OutputTracks
[docs]@configurable
def make_PrKalmanFilter_noUT_tracks(
input_tracks: DataHandle,
hits_vp: DataHandle,
hits_ft: DataHandle,
max_chi2ndof: float = 2.8,
max_chi2ndof_pre_outlier_removal: float = 20,
min_chi2_outlier: float = 9,
reference_extrapolator: Any = get_track_master_extrapolator,
name: str = None,
) -> DataHandle:
"""Configure the PrKalmanFilter to fit Long Tracks without UT hits.
Args:
input_tracks (DataHandle): Location of PrLongTracks.
hits_vp (DataHandle): Location of Velo Hits.
hits_ft (DataHandle): Location of SciFi Hits.
max_chi2ndof (float, optional): Maximum chi2/ndof of the fitted track. Defaults to 2.8.
max_chi2ndof_pre_outlier_removal (float, optional): Maximum chi2/ndof of the fitted track before outlier removal. Defaults to 20.
min_chi2_outlier (float, optional): Minimum chi2 of node to be counted as outlier. Defaults to 9.
reference_extrapolator (Any, optional): Defaults to get_track_master_extrapolator.
name (str, optional): Defaults to None.
Returns:
DataHandle: Location of fitted tracks.
"""
return PrKalmanFilter_noUT(
Input=input_tracks,
HitsVP=hits_vp,
HitsFT=hits_ft,
MaxChi2=max_chi2ndof,
MaxChi2PreOutlierRemoval=max_chi2ndof_pre_outlier_removal,
MinChi2Outlier=min_chi2_outlier,
ReferenceExtrapolator=reference_extrapolator(),
name=name,
InputUniqueIDGenerator=make_unique_id_generator(),
).OutputTracks
[docs]@configurable
def make_PrKalmanFilter_Downstream_tracks(
input_tracks: DataHandle,
hits_ut: DataHandle,
hits_ft: DataHandle,
max_chi2ndof: float = 2.8,
max_chi2ndof_pre_outlier_removal: float = 20,
min_chi2_outlier: float = 9,
reference_extrapolator: Any = get_track_master_extrapolator,
name: str = None,
) -> DataHandle:
"""Configure the PrKalmanFilter to fit Downstream Tracks.
Args:
input_tracks (DataHandle): Location of PrDownstreamTracks.
hits_ut (DataHandle): Location of UT Hits.
hits_ft (DataHandle): Location of SciFi Hits.
max_chi2ndof (float, optional): Maximum chi2/ndof of the fitted track. Defaults to 2.8.
max_chi2ndof_pre_outlier_removal (float, optional): Maximum chi2/ndof of the fitted track before outlier removal. Defaults to 20.
min_chi2_outlier (float, optional): Minimum chi2 of node to be counted as outlier. Defaults to 9.
reference_extrapolator (Any, optional): Defaults to get_track_master_extrapolator.
name (str, optional): Defaults to None.
Returns:
DataHandle: Location of fitted tracks.
"""
return PrKalmanFilter_Downstream(
Input=input_tracks,
HitsUT=hits_ut,
HitsFT=hits_ft,
MaxChi2=max_chi2ndof,
MaxChi2PreOutlierRemoval=max_chi2ndof_pre_outlier_removal,
MinChi2Outlier=min_chi2_outlier,
ReferenceExtrapolator=reference_extrapolator(),
name=name,
InputUniqueIDGenerator=make_unique_id_generator(),
).OutputTracks
[docs]@configurable
def make_PrKalmanFilter_Upstream_tracks(
input_tracks: DataHandle,
hits_vp: DataHandle,
hits_ut: DataHandle,
max_chi2ndof: float = 2.8,
max_chi2ndof_pre_outlier_removal: float = 20,
min_chi2_outlier: float = 9,
reference_extrapolator: Any = get_track_master_extrapolator,
name: str = None,
) -> DataHandle:
"""Configure the PrKalmanFilter to fit Upstream Tracks.
Args:
input_tracks (DataHandle): Location of PrUpstreamTracks.
hits_vp (DataHandle): Location of Velo Hits.
hits_ut (DataHandle): Location of UT Hits.
max_chi2ndof (float, optional): Maximum chi2/ndof of the fitted track. Defaults to 2.8.
max_chi2ndof_pre_outlier_removal (float, optional): Maximum chi2/ndof of the fitted track before outlier removal. Defaults to 20.
min_chi2_outlier (float, optional): Minimum chi2 of node to be counted as outlier. Defaults to 9.
reference_extrapolator (Any, optional): Defaults to get_track_master_extrapolator.
name (str, optional): Defaults to None.
Returns:
DataHandle: Location of fitted tracks.
"""
return PrKalmanFilter_Upstream(
Input=input_tracks,
MaxChi2=max_chi2ndof,
MaxChi2PreOutlierRemoval=max_chi2ndof_pre_outlier_removal,
HitsUT=hits_ut,
HitsVP=hits_vp,
ReferenceExtrapolator=reference_extrapolator(),
name=name,
InputUniqueIDGenerator=make_unique_id_generator()).OutputTracks
[docs]@configurable
def make_PrKalmanFilter_Velo_tracks(
input_tracks: DataHandle,
hits_vp: DataHandle,
max_chi2ndof: float = 2.8,
max_chi2ndof_pre_outlier_removal: float = 20,
min_chi2_outlier: float = 9,
reference_extrapolator: Any = TrackLinearExtrapolator,
name: str = None,
) -> DataHandle:
"""Configure the PrKalmanFilter to fit Velo Tracks.
Args:
input_tracks (DataHandle): Location of PrVeloTracks.
hits_vp (DataHandle): Location of Velo Hits.
max_chi2ndof (float, optional): Maximum chi2/ndof of the fitted track. Defaults to 2.8.
max_chi2ndof_pre_outlier_removal (float, optional): Maximum chi2/ndof of the fitted track before outlier removal. Defaults to 20.
min_chi2_outlier (float, optional): Minimum chi2 of node to be counted as outlier. Defaults to 9.
reference_extrapolator (Any, optional): Defaults to TrackLinearExtrapolator.
name (str, optional): Defaults to None.
Returns:
DataHandle: Location of fitted tracks.
"""
return PrKalmanFilter_Velo(
Input=input_tracks,
HitsVP=hits_vp,
MaxChi2=max_chi2ndof,
MaxChi2PreOutlierRemoval=max_chi2ndof_pre_outlier_removal,
MinChi2Outlier=min_chi2_outlier,
ReferenceExtrapolator=reference_extrapolator(),
name=name,
InputUniqueIDGenerator=make_unique_id_generator(),
).OutputTracks
[docs]@configurable
def make_PrKalmanFilter_Seed_tracks(
input_tracks: DataHandle,
hits_ft: DataHandle,
max_chi2ndof: float = 2.8,
max_chi2ndof_pre_outlier_removal: float = 20,
min_chi2_outlier: float = 9,
reference_extrapolator: Any = get_track_master_extrapolator,
name: str = None,
) -> DataHandle:
"""Configure the PrKalmanFilter to fit Seed Tracks.
Args:
input_tracks (DataHandle): Location of PrSeedTracks.
hits_ft (DataHandle): Location of SciFi Hits.
max_chi2ndof (float, optional): Maximum chi2/ndof of the fitted track. Defaults to 2.8.
max_chi2ndof_pre_outlier_removal (float, optional): Maximum chi2/ndof of the fitted track before outlier removal. Defaults to 20.
min_chi2_outlier (float, optional): Minimum chi2 of node to be counted as outlier. Defaults to 9.
reference_extrapolator (Any, optional): Defaults to get_track_master_extrapolator.
name (str, optional): Defaults to None.
Returns:
DataHandle: Location of fitted tracks.
"""
return PrKalmanFilter_Seed(
Input=input_tracks,
HitsFT=hits_ft,
MaxChi2=max_chi2ndof,
MaxChi2PreOutlierRemoval=max_chi2ndof_pre_outlier_removal,
MinChi2Outlier=min_chi2_outlier,
ReferenceExtrapolator=reference_extrapolator(),
name=name,
InputUniqueIDGenerator=make_unique_id_generator(),
).OutputTracks
@configurable
def get_GhostProbabilityTool(track_type='',
weights_location='',
without_UT=False):
if not without_UT:
return get_UpgradeGhostId_tool
else:
if track_type == 'Long':
return partial(
GhostProbabilityTool_Long_noUT,
WeightsFileName=
'paramfile://data/GhostProbability/hlt2_ghostprob_long_noUT.json'
if not weights_location else weights_location,
TrackType='Long')
elif track_type == 'Ttrack':
return partial(
GhostProbabilityTool_Ttrack,
WeightsFileName=
'paramfile://data/GhostProbability/hlt2_ghostprob_ttrack_noUT.json'
if not weights_location else weights_location,
TrackType='Ttrack')
elif track_type == 'Velo':
return partial(
GhostProbabilityTool_Velo,
WeightsFileName=
'paramfile://data/GhostProbability/hlt2_ghostprob_velo_noUT.json'
if not weights_location else weights_location,
TrackType='Velo')
else:
return get_UpgradeGhostId_tool_no_UT
[docs]def get_default_hlt2_tracks():
"""Function to get default set of tracks reconstructed in HLT2, which are later used as input to `make_TrackBestTrackCreator_tracks`.
Returns:
A dict mapping all types of velo, upstream, HLT1 forward fitted, HLT2 forward, SciFi seeding, downstream and matched long tracks to ``'Velo'``, ``'Upstream'``, ``'ForwardFastFitted'``, ``'Forward'``, ``'Seed'``, ``'Downstream'`` and ``'Match'``` respectively.
"""
hlt1_tracks = make_legacy_rec_hlt1_tracks()
fitted_hlt1_tracks = make_legacy_rec_hlt1_fitted_tracks(hlt1_tracks)
hlt2_forward_tracks = all_hlt2_forward_track_types(hlt1_tracks["Velo"])
scifi_tracks = make_seeding_tracks()
downstream_tracks = make_PrLongLivedTracking_tracks(scifi_tracks)
match_tracks = make_PrMatchNN_tracks(hlt1_tracks["Velo"], scifi_tracks)
return {
"Velo": hlt1_tracks["Velo"],
"Upstream": hlt1_tracks["Upstream"],
"ForwardFastFitted": fitted_hlt1_tracks,
"Forward": hlt2_forward_tracks,
"Seed": scifi_tracks,
"Downstream": downstream_tracks,
"Match": match_tracks
}
[docs]def get_default_hlt2_tracks_without_UT():
"""Function to get default set of tracks reconstructed without UT in HLT2, which are later used as input to `make_TrackBestTrackCreator_tracks`.
Returns:
A dict mapping all types of velo, HLT2 forward, SciFi seeding and matched long tracks to ``'Velo'``, ``'Forward'``, ``'Seed'`` and ``'Match'``` respectively.
"""
velo_tracks = all_velo_track_types()
make_forward = partial(
make_PrForwardTrackingVelo_tracks,
ut_hits_tool=get_ignore_ut_hits_tool,
add_ut_hits=False)
hlt2_forward_tracks = all_hlt2_forward_track_types(
velo_tracks, make_forward_tracks=make_forward)
scifi_tracks = make_seeding_tracks()
match_tracks = make_PrMatchNN_tracks(
velo_tracks,
scifi_tracks,
ut_hits_tool=get_ignore_ut_hits_tool,
add_ut_hits=False)
return {
"Velo": velo_tracks,
"Forward": hlt2_forward_tracks,
"Seed": scifi_tracks,
"Match": match_tracks
}
[docs]def get_default_track_list_for_light_reco(skip_UT=False):
"""Function to set the default order of tracks which are used as input for the track fit.
Returns:
A list of strings.
"""
if (skip_UT):
return ["Forward", "Match"]
else:
return ["Forward", "Match", "Downstream", "Upstream"]
[docs]def get_default_track_list_for_TrackBestTrackCreator(skip_UT=False):
"""Function to set the default order of tracks which are used as input to `make_TrackBestTrackCreator_tracks`.
Returns:
A list of strings.
"""
#TODO: the list has a random order, pls fix
if (skip_UT):
return ["Velo", "Match", "Seed", "Forward"]
else:
return [
"Velo", "ForwardFastFitted", "Upstream", "Match", "Seed",
"Forward", "Downstream"
]
[docs]def get_default_out_track_types_for_light_reco(skip_UT=False):
"""Returns default output track types of light reco sequence.
Returns:
Dict with lists of strings.
"""
return {
"Best": ["BestLong"] if skip_UT else ["BestLong", "BestDownstream"],
"Unfitted4Calo": ["SeedDecloned"]
}
[docs]def get_default_tracks_for_calo(tracks, light_reco=False):
"""Returns default tracks used as input for calo.
Returns:
A list of strings.
"""
if light_reco:
track_dict = get_default_out_track_types_for_light_reco()
track_list = []
for key in ["Best", "Unfitted4Calo"]:
track_list += track_dict[key]
trackSelections = [tracks[key]["v1"] for key in track_list]
return TrackSelectionMerger(
InputLocations=trackSelections).OutputLocation
else:
return tracks["Best"]["v1"]
[docs]@configurable
def kill_clones(inputTracks,
referenceTracks=None,
SkipSameContainerTracks=True,
KeepUnFitted=True,
UseUnFittedRef=False):
"""Perform the clone killing with respect to `referenceTracks` containers without the fit.
"""
if referenceTracks is not None:
clone_killer = partial(
TrackCloneKiller,
SkipSameContainerTracks=SkipSameContainerTracks,
KeepUnFitted=KeepUnFitted,
UseUnFittedRef=UseUnFittedRef)
if len(referenceTracks) == 1:
decloned = clone_killer(
TracksInContainer=inputTracks,
TracksRefContainer=referenceTracks[0]).TracksOutContainer
else:
selections = []
for reference in referenceTracks:
selections += [
TracksToSelection(InputLocation=reference).OutputLocation
]
merged_selections = TrackSelectionMerger(
InputLocations=selections).OutputLocation
# to be fixed:
merged_tracks = TrackSelectionToContainer(
InputLocation=merged_selections).OutputLocation
decloned = clone_killer(
TracksInContainer=inputTracks,
TracksRefContainer=merged_tracks).TracksOutContainer
else:
decloned = inputTracks
return decloned
[docs]@configurable
def fit_and_select(inputTracks,
referenceTracks=None,
get_fitter_tool=get_track_master_fitter,
get_ghost_tool=get_GhostProbabilityTool(),
do_not_refit=False,
nameSuffix=""):
"""Perform the fit and selection of tracks:
1. Clone killing with respect to `referenceTracks` containers of (preferably) fitted tracks
2. Fit decloned container using `TrackEventFitter`
3. Select tracks using `TrackBestTrackCreator` with no-fit mode
"""
decloned = kill_clones(
inputTracks=inputTracks, referenceTracks=referenceTracks)
if do_not_refit:
fitted = decloned
else:
fitted = TrackEventFitter(
name="TrackEventFitter" + nameSuffix,
TracksInContainer=decloned,
Fitter=get_fitter_tool(MaxUpdateTransports=0)).TracksOutContainer
best = TrackBestTrackCreator(
name="BestTrackCreator" + nameSuffix,
TracksInContainers=[fitted],
GhostIdTool=get_ghost_tool(),
DoNotRefit=True,
AddGhostProb=True,
FitTracks=False).TracksOutContainer
return best
[docs]@configurable
def make_TrackBestTrackCreator_tracks(
tracks: Any,
track_version: str = "v1",
get_tracklist:
Callable = get_default_track_list_for_TrackBestTrackCreator,
get_fitter_tool: Callable = get_track_master_fitter,
get_ghost_tool: Callable = get_GhostProbabilityTool(),
do_not_refit: bool = False,
add_ghost_prob: bool = True,
fit_tracks: bool = True,
max_chi2ndof: float = 3,
max_ghost_prob: float = 99999,
name: str = None,
skip_UT: bool = False) -> dict[str, DataHandle]:
"""Persists best quality tracks, calls track fitters, kills clones and adds neural-net response for fake-track (a.k.a. ghost) rejection.
Args:
tracks (dict or list): Reconstructed tracks, e.g. from `get_default_hlt2_tracks`.
track_version (str, optional): Track version of input and output tracks. Defaults to "v1".
get_tracklist (Callable, optional): Sets the list of `tracks` which is used as input to TrackBestTrackCreator's ``TracksInContainers``. Be aware that changing the order of tracks in that list has an impact on the output and performance of TrackBestTrackCreator.. Defaults to get_default_track_list_for_TrackBestTrackCreator.
get_fitter_tool (Callable, optional): Track fitting tool. Defaults to `get_track_master_fitter <RecoConf.legacy_rec_hlt1_tracking.get_track_master_fitter>`.
get_ghost_tool (Callable, optional): GhostId tool, adding a neural-net response that has been trained to reject fake (a.k.a. ghost) tracks. Defaults to get_GhostProbabilityTool.
do_not_refit (bool, optional): Whether or not to not refit input tracks. Defaults to False.
add_ghost_prob (bool, optional): Add ghost probability information to fitted tracks. Defaults to True.
fit_tracks (bool, optional): Defaults to True.
max_chi2ndof (float, optional): Maximum chi2/ndof of best tracks. Defaults to 3.
name (str, optional): Defaults to None.
skip_UT (bool, optional): Defaults to False.
Raises:
ConfigurationError: Raises error if input is not a list or a dict.
Returns:
dict[str, DataHandle]: Dictionary with "Best" key.
"""
outTracks = {}
if isinstance(tracks, dict):
track_list = [
tracks[track_type][track_version]
for track_type in get_tracklist(skip_UT=skip_UT)
]
elif isinstance(tracks, list):
track_list = tracks
else:
raise ConfigurationError(
"make_TrackBestTrackCreator_tracks expects either list or dict.")
tbtc = partial(
TrackBestTrackCreator,
name=name,
TracksInContainers=track_list,
GhostIdTool=get_ghost_tool(),
DoNotRefit=do_not_refit,
AddGhostProb=add_ghost_prob,
FitTracks=fit_tracks,
MaxChi2DoF=max_chi2ndof,
MaxGhostProb=max_ghost_prob,
)
if fit_tracks:
if skip_UT:
with get_global_measurement_provider.bind(
ignoreUT=True, ut_hits=make_PrStoreUTHit_empty_hits):
outTracks["Best"] = tbtc(
Fitter=get_fitter_tool()).TracksOutContainer
else:
outTracks["Best"] = tbtc(
Fitter=get_fitter_tool()).TracksOutContainer
else:
# do not configure the fitter if not needed, dd4hep is not happy otherwise (UT dependency)
outTracks["Best"] = tbtc().TracksOutContainer
return outTracks
[docs]@configurable
def make_light_reco_best_tracks(
tracks,
track_version,
fit_preselection=F.ALL,
get_tracklist=get_default_track_list_for_light_reco,
do_not_refit=False,
skip_UT=False):
"""Persists best quality tracks, calls track fitters, kills clones and adds neural-net response for fake-track (a.k.a. ghost) rejection.
Rejects too soft tracks prior to the fit.
Returns:
outTracks: dictionaty of track DataHandles.
"""
track_list = get_tracklist(skip_UT=skip_UT)
input_tracks = {
track_type: tracks[track_type][track_version]
for track_type in track_list
}
dictTracks = {}
selected_tracks = {}
if fit_preselection is not F.ALL:
for trType in track_list:
splitter = TrackContainerSplitter(
name="TrackContainerSplitter" + trType,
TracksInContainer=input_tracks[trType],
Code=fit_preselection)
selected_tracks[trType] = splitter.PassedContainer
dictTracks["Soft" + trType] = splitter.RejectedContainer
tracks_to_merge = [dictTracks["SoftForward"], dictTracks["SoftMatch"]]
dictTracks["SoftLong"] = TrackContainersMerger(
name="TrackContainersMergerSoftLong",
InputLocations=tracks_to_merge).OutputLocation
else:
selected_tracks = {
trType: input_tracks[trType]
for trType in track_list
}
with get_global_measurement_provider.bind(ignoreUT=skip_UT):
dictTracks["BestForward"] = fit_and_select(
inputTracks=selected_tracks["Forward"],
do_not_refit=do_not_refit,
get_ghost_tool=get_GhostProbabilityTool(
track_type='Long', without_UT=skip_UT),
nameSuffix="Forward")
dictTracks["BestMatch"] = fit_and_select(
inputTracks=selected_tracks["Match"],
referenceTracks=[dictTracks["BestForward"]],
do_not_refit=do_not_refit,
get_ghost_tool=get_GhostProbabilityTool(
track_type='Long', without_UT=skip_UT),
nameSuffix="Match")
# all of these gives a very tiny more ~0.01% long tracks wrt to number of long tracks from TrackBestTrackCreator(Match, Forward)
tracks_to_merge = [dictTracks["BestMatch"], dictTracks["BestForward"]]
dictTracks["BestLong"] = TrackContainersMerger(
name="TrackContainersMergerLong",
InputLocations=tracks_to_merge).OutputLocation
if not skip_UT:
dictTracks["BestDownstream"] = fit_and_select(
inputTracks=selected_tracks["Downstream"],
referenceTracks=[dictTracks["BestLong"]],
do_not_refit=do_not_refit,
get_ghost_tool=get_GhostProbabilityTool(
track_type='Downstream'),
nameSuffix="Downstream")
dictTracks["BestUpstream"] = fit_and_select(
inputTracks=selected_tracks["Upstream"],
referenceTracks=[dictTracks["BestLong"]],
do_not_refit=do_not_refit,
get_ghost_tool=get_GhostProbabilityTool(track_type='Upstream'),
nameSuffix="Upstream")
# this gives ~0.5% less downstream tracks wrt to number of downstream tracks from TrackBestTrackCreator(Long, Downstream)
seed_reference_keys = ["BestLong"]
if not skip_UT:
seed_reference_keys += ["BestDownstream"]
dictTracks["SeedDecloned"] = kill_clones(
inputTracks=tracks["Seed"][track_version],
referenceTracks=[dictTracks[key] for key in seed_reference_keys])
dictTracks["BestSeed"] = fit_and_select(
inputTracks=tracks['Seed']['v1'],
do_not_refit=do_not_refit,
get_ghost_tool=get_GhostProbabilityTool(
track_type='Ttrack', without_UT=skip_UT),
nameSuffix="Seed")
splitter = TrackContainerSplitter(
name="TrackContainerSplitterVeloBackward",
TracksInContainer=tracks['Velo']['v1'],
Code=F.TRACKISVELOBACKWARD)
velo_backward = splitter.PassedContainer
velo_forward = splitter.RejectedContainer
dictTracks["BestVelo"] = fit_and_select(
inputTracks=velo_forward,
do_not_refit=do_not_refit,
get_ghost_tool=get_GhostProbabilityTool(
track_type='Velo', without_UT=skip_UT),
nameSuffix="Velo")
dictTracks["BestVeloBackward"] = fit_and_select(
inputTracks=velo_backward,
do_not_refit=do_not_refit,
get_ghost_tool=get_GhostProbabilityTool(
track_type='VeloBackward', without_UT=skip_UT),
nameSuffix="VeloBackward")
out_track_dict = get_default_out_track_types_for_light_reco(
skip_UT=skip_UT)
#add the upstream tracks "by hand", temporary solution
if not skip_UT:
out_track_dict["Best"].append("BestUpstream")
outTrackTypes = out_track_dict["Best"] + out_track_dict["Unfitted4Calo"]
outTrackTypes += ["BestVelo", "BestVeloBackward",
"BestSeed"] # needed for monitoring
outTracks = {
track_type: dictTracks[track_type]
for track_type in outTrackTypes
}
return outTracks
[docs]@configurable
def make_pr_kf_light_reco_best_tracks(tracks,
fast_reco,
fit_forward_first=True):
"""
Preselect forward,match, and downstream tracks
Fit forward tracks -> TrackBestTrackCreator (TBTC) for ghost rejection
Clone kill match Tracks with respect to forward -> Fit -> TBTC for ghost rejection
Merge match and forward -> BestLong
Clone kill downstream with respect to BestLong -> Fit -> TBTC for ghost rejection
Returns:
DataHandles: dictionary of BestLong, BestDownstream, BestUpstream, SeedDecloned, BestSeed, BestVelo, BestVeloBackward
"""
from PyConf.Algorithms import (PrCloneKillerLong, PrCloneKillerDown,
PrCloneKillerUp, PrCloneKillerSeed,
PrCloneKillerSeedDown)
vp_hits = make_VeloClusterTrackingSIMD_hits()
ut_hits = make_PrStorePrUTHits_hits()
ft_hits = make_PrStoreSciFiHits_hits()
if fit_forward_first:
first, name_1 = tracks['Forward']['Pr'], "Forward"
second, name_2 = tracks['Match']['Pr'], "Match"
first_ut_hits = tracks['UTHitsRes'] if fast_reco else ut_hits
first_ft_hits = tracks['FTHitsRes'] if fast_reco else ft_hits
second_ut_hits = ut_hits
second_ft_hits = ft_hits
else:
first, name_1 = tracks['Match']['Pr'], "Match"
second, name_2 = tracks['Forward']['Pr'], "Forward"
second_ut_hits = tracks['UTHitsRes'] if fast_reco else ut_hits
second_ft_hits = tracks['FTHitsRes'] if fast_reco else ft_hits
first_ut_hits = ut_hits
first_ft_hits = ft_hits
fitted_1 = make_PrKalmanFilter_tracks(
name="PrKalmanFilter" + name_1 + "_{hash}",
input_tracks=first,
hits_vp=vp_hits,
hits_ut=first_ut_hits,
hits_ft=first_ft_hits)
best_1 = make_TrackBestTrackCreator_tracks(
tracks=[fitted_1],
name="TBTC_" + name_1 + "_{hash}",
get_ghost_tool=get_GhostProbabilityTool(track_type='Long'),
do_not_refit=True,
fit_tracks=False)["Best"]
decloned_2 = PrCloneKillerLong(
name="CloneKiller" + name_2 + "_{hash}",
TracksInContainer=second,
TracksRefContainer=best_1,
).TracksOutContainer
fitted_2 = make_PrKalmanFilter_tracks(
name="PrKalmanFilter" + name_2 + "_{hash}",
input_tracks=decloned_2,
hits_vp=vp_hits,
hits_ut=second_ut_hits,
hits_ft=second_ft_hits)
best_2 = make_TrackBestTrackCreator_tracks(
tracks=[fitted_2],
name="TBTC" + name_2 + "_{hash}", # FIXME TBTC -> TBTC_
get_ghost_tool=get_GhostProbabilityTool(track_type='Long'),
do_not_refit=True,
fit_tracks=False)["Best"]
best_long = TrackContainersMerger(
InputLocations=[best_1, best_2]).OutputLocation
decloned_down = PrCloneKillerDown(
TracksInContainer=tracks['Downstream']['Pr'],
TracksRefContainer=best_long,
).TracksOutContainer
fitted_down = make_PrKalmanFilter_Downstream_tracks(
input_tracks=decloned_down,
hits_ut=tracks['UTHitsRes'] if fast_reco else ut_hits,
hits_ft=ft_hits)
best_down = make_TrackBestTrackCreator_tracks(
tracks=[fitted_down],
name="TBTC_down_{hash}",
get_ghost_tool=get_GhostProbabilityTool(track_type='Downstream'),
do_not_refit=True,
fit_tracks=False)["Best"]
seed_vs_bestlong = PrCloneKillerSeed(
TracksInContainer=tracks["Seed"]["Pr"],
TracksRefContainer=best_long,
).TracksOutContainer
seed_decloned_pr = PrCloneKillerSeedDown(
TracksInContainer=seed_vs_bestlong,
TracksRefContainer=best_down,
).TracksOutContainer
seed_decloned = fromPrSeedingTracksV1Tracks(
InputTracksLocation=seed_decloned_pr).OutputTracksLocation
# these are used by the monitoring/mcchecking
seed_fitted = make_PrKalmanFilter_Seed_tracks(
input_tracks=tracks['Seed']['Pr'], hits_ft=ft_hits)
best_seed = make_TrackBestTrackCreator_tracks(
tracks=[seed_fitted],
name="TBTC_Seed_{hash}",
do_not_refit=True,
get_ghost_tool=get_GhostProbabilityTool(track_type='Ttrack'),
fit_tracks=False)["Best"]
declonned_up = PrCloneKillerUp(
TracksInContainer=tracks['Upstream']['Pr'],
TracksRefContainer=best_long,
).TracksOutContainer
fitted_up = make_PrKalmanFilter_Upstream_tracks(
input_tracks=declonned_up, hits_ut=ut_hits, hits_vp=vp_hits)
best_up = make_TrackBestTrackCreator_tracks(
tracks=[fitted_up],
name="TBTC_up_{hash}",
get_ghost_tool=get_GhostProbabilityTool(track_type='Upstream'),
do_not_refit=True,
fit_tracks=False)["Best"]
velo_fitted = make_PrKalmanFilter_Velo_tracks(
input_tracks=tracks['Velo']['Pr'],
hits_vp=vp_hits,
)
best_velo = make_TrackBestTrackCreator_tracks(
tracks=[velo_fitted],
name="TBTC_Velo_{hash}",
get_ghost_tool=get_GhostProbabilityTool(track_type='Velo'),
do_not_refit=True,
fit_tracks=False)["Best"]
velo_backward_fitted = make_PrKalmanFilter_Velo_tracks(
name="PrKalmanFilter_VeloBackward_{hash}",
input_tracks=tracks['Velo']['Pr::backward'],
hits_vp=vp_hits,
)
best_velo_backward = make_TrackBestTrackCreator_tracks(
tracks=[velo_backward_fitted],
name="TBTC_VeloBackward_{hash}",
get_ghost_tool=get_GhostProbabilityTool(track_type='VeloBackward'),
do_not_refit=True,
fit_tracks=False)["Best"]
return {
'BestLong': best_long,
'BestUpstream': best_up,
'BestDownstream': best_down,
'BestVelo': best_velo,
'BestVeloBackward': best_velo_backward,
'BestSeed': best_seed,
'SeedDecloned': seed_decloned,
}
[docs]@configurable
def make_pr_kf_light_reco_best_tracks_without_UT(tracks,
fast_reco,
fit_forward_first=True):
"""
Preselect forward, match, and downstream tracks
Fit forward tracks -> TrackBestTrackCreator (TBTC) for ghost rejection
Clone kill match Tracks with respect to forward -> Fit -> TBTC for ghost rejection
Merge match and forward -> BestLong
Returns:
DataHandles: BestLong
"""
from PyConf.Algorithms import (PrCloneKillerLong, PrCloneKillerSeed)
vp_hits = make_VeloClusterTrackingSIMD_hits()
ft_hits = make_PrStoreSciFiHits_hits()
if fit_forward_first:
first, name_1 = tracks['Forward']['Pr'], "Forward"
second, name_2 = tracks['Match']['Pr'], "Match"
first_ft_hits = tracks['FTHitsRes'] if fast_reco else ft_hits
second_ft_hits = ft_hits
else:
first, name_1 = tracks['Match']['Pr'], "Match"
second, name_2 = tracks['Forward']['Pr'], "Forward"
second_ft_hits = tracks['FTHitsRes'] if fast_reco else ft_hits
first_ft_hits = ft_hits
fitted_1 = make_PrKalmanFilter_noUT_tracks(
name="PrKalmanFilter" + name_1 + "_{hash}",
input_tracks=first,
hits_vp=vp_hits,
hits_ft=first_ft_hits)
best_1 = make_TrackBestTrackCreator_tracks(
tracks=[fitted_1],
name="TBTC_" + name_1 + "_{hash}",
get_ghost_tool=get_GhostProbabilityTool(
track_type='Long', without_UT=True),
do_not_refit=True,
fit_tracks=False,
skip_UT=True)["Best"]
decloned_2 = PrCloneKillerLong(
name="CloneKiller" + name_2 + "_{hash}",
TracksInContainer=second,
TracksRefContainer=best_1,
).TracksOutContainer
fitted_2 = make_PrKalmanFilter_noUT_tracks(
name="PrKalmanFilter" + name_2 + "_{hash}",
input_tracks=decloned_2,
hits_vp=vp_hits,
hits_ft=second_ft_hits)
best_2 = make_TrackBestTrackCreator_tracks(
tracks=[fitted_2],
name="TBTC" + name_2 + "_{hash}",
get_ghost_tool=get_GhostProbabilityTool(
track_type='Long', without_UT=True),
do_not_refit=True,
fit_tracks=False,
skip_UT=True)["Best"]
best_long = TrackContainersMerger(
InputLocations=[best_1, best_2]).OutputLocation
decloned_seed_pr = PrCloneKillerSeed(
TracksInContainer=tracks["Seed"]["Pr"],
TracksRefContainer=best_long,
).TracksOutContainer
decloned_seed = fromPrSeedingTracksV1Tracks(
InputTracksLocation=decloned_seed_pr).OutputTracksLocation
# these are used by the monitoring
seed_fitted = make_PrKalmanFilter_Seed_tracks(
input_tracks=tracks['Seed']['Pr'], hits_ft=ft_hits)
best_seed = make_TrackBestTrackCreator_tracks(
tracks=[seed_fitted],
name="TBTC_Seed_{hash}",
get_ghost_tool=get_GhostProbabilityTool(
track_type='Ttrack', without_UT=True),
do_not_refit=True,
fit_tracks=False,
skip_UT=True)["Best"]
velo_fitted = make_PrKalmanFilter_Velo_tracks(
input_tracks=tracks['Velo']['Pr'],
hits_vp=vp_hits,
)
best_velo = make_TrackBestTrackCreator_tracks(
tracks=[velo_fitted],
name="TBTC_Velo_{hash}",
get_ghost_tool=get_GhostProbabilityTool(
track_type='Velo', without_UT=True),
do_not_refit=True,
fit_tracks=False,
skip_UT=True)["Best"]
velo_backward_fitted = make_PrKalmanFilter_Velo_tracks(
name="PrKalmanFilter_VeloBackward_{hash}",
input_tracks=tracks['Velo']['Pr::backward'],
hits_vp=vp_hits,
)
best_velo_backward = make_TrackBestTrackCreator_tracks(
tracks=[velo_backward_fitted],
name="TBTC_VeloBackward_{hash}",
get_ghost_tool=get_GhostProbabilityTool(
track_type='VeloBackward', without_UT=True),
do_not_refit=True,
fit_tracks=False,
skip_UT=True)["Best"]
return {
'BestLong': best_long,
'BestSeed': best_seed,
'BestVelo': best_velo,
'BestVeloBackward': best_velo_backward,
'SeedDecloned': decloned_seed
}
[docs]@configurable
def make_hlt2_tracks(light_reco=False, fast_reco=False, use_pr_kf=False):
"""Function to get all types of tracks reconstructed in HLT2
Returns:
A dict mapping all types of velo, upstream, HLT1 forward fitted, HLT2 forward, SciFi seeding, downstream, matched long and best tracks to ``'Velo'``, ``'Upstream'``, ``'ForwardFastFitted'``, ``'Forward'``, ``'Seed'``, ``'Downstream'``, ``'Match'`` and ``'Best'`` respectively.
"""
if fast_reco:
track_dict = get_fast_hlt2_tracks()
else:
track_dict = get_default_hlt2_tracks()
track_version = "v1"
if light_reco:
if use_pr_kf:
# it's currently hardcoded but let's make sure that everyone knows we only return v1 for now
assert track_version == 'v1'
track_containers = make_pr_kf_light_reco_best_tracks(
tracks=track_dict, fast_reco=fast_reco)
else:
track_containers = make_light_reco_best_tracks(
tracks=track_dict, track_version=track_version)
else:
track_containers = make_TrackBestTrackCreator_tracks(
tracks=track_dict, track_version=track_version)
for trType in track_containers.keys():
track_dict[trType] = {track_version: track_containers[trType]}
return track_dict
[docs]@configurable
def make_hlt2_tracks_without_UT(light_reco=True,
fast_reco=False,
use_pr_kf=True):
"""Function to get all types of tracks reconstructed in HLT2 without the UT
Returns:
A dict mapping all types of velo, HLT2 forward, SciFi seeding, matched long and best tracks to ``'Velo'``, ``'Forward'``, ``'Seed'``, ``'Match'`` and ``'Best'`` respectively.
"""
if fast_reco:
track_dict = get_fast_hlt2_tracks_without_UT()
else:
track_dict = get_default_hlt2_tracks_without_UT()
track_version = "v1"
if light_reco:
if use_pr_kf:
# it's currently hardcoded but let's make sure that everyone knows we only return v1 for now
assert track_version == 'v1'
# if both, matching and forward run on all hits, fitting matching first gives higher hit efficiency
track_containers = make_pr_kf_light_reco_best_tracks_without_UT(
tracks=track_dict,
fast_reco=fast_reco,
fit_forward_first=fast_reco)
else:
track_containers = make_light_reco_best_tracks(
tracks=track_dict, track_version=track_version, skip_UT=True)
elif use_pr_kf:
raise RuntimeError(
"sequence without UT that uses PrKalmanFilter only implemented for light reco configuration"
)
else:
track_containers = make_TrackBestTrackCreator_tracks(
tracks=track_dict,
track_version=track_version,
skip_UT=True,
get_ghost_tool=get_GhostProbabilityTool(without_UT=True))
for trType in track_containers.keys():
track_dict[trType] = {track_version: track_containers[trType]}
return track_dict
[docs]def make_ReduceVeloTracks_fromLong(input_tracks, velotracks):
"""
Function to remove Velo tracks used by PrMatchNN or PrForwardTracking algorithm and
create a new velo track container for the residual tracks, which are later used as
input to other algorithms.
"""
velo_tracks = PrResidualVeloTracks(
TracksLocation=input_tracks["Pr"],
VeloTrackLocation=velotracks["Pr"]).VeloTrackOutput
v1_tracks = fromPrVeloTracksV1Tracks(
InputTracksLocation=velo_tracks).OutputTracksLocation
return {"Pr": velo_tracks, "v1": v1_tracks}
[docs]def make_ReduceSeedTracks_fromMatch(match_tracks, seed_tracks):
"""
Function to remove Seed tracks used by PrMatchNN algorithm and create a new SeedTracks container for the residual tracks, which are later used as input to other algorithm PrLongLivedTracking
"""
residual_seed_pr = PrResidualSeedingLong(
MatchTracksLocation=match_tracks["Pr"],
SeedTracksLocation=seed_tracks["Pr"]).SeedTracksOutput
residual_seed_v1 = fromPrSeedingTracksV1Tracks(
InputTracksLocation=residual_seed_pr).OutputTracksLocation
return {"Pr": residual_seed_pr, "v1": residual_seed_v1}
[docs]@configurable
def make_ReduceSciFiHits_fromLong(input_tracks,
make_ft_hits=make_PrStoreSciFiHits_hits):
"""
Function to remove SciFi hits used by PrMatchNN or PrFowardTracking algorithm
and create a new SciFiHits container for the residual SciFi hits,
which are later used as input to other algorithms.
"""
return PrResidualSciFiHits(
TracksLocation=input_tracks["Pr"],
SciFiHitsLocation=make_ft_hits()).SciFiHitsOutput
[docs]@configurable
def make_ReducePrUTHits_fromLong(input_tracks,
make_ut_hits=make_PrStorePrUTHits_hits):
"""
Function to remove PrUTHits used by Long tracks and create a new PrUTHits container for the residual PrUTHits, which are later used as input to PrLongLivedTracking or PrForwardTracking
"""
return PrResidualPrUTHits(
TracksLocation=input_tracks["Pr"],
PrUTHitsLocation=make_ut_hits()).PrUTHitsOutput
[docs]def get_fast_hlt2_tracks():
"""Function to get fast set of tracks reconstructed in HLT2
The PrForwardTracking uses the residual VeloTracks and SciFi hits from ``'Match'`` tracks, the PrLonglivedTracking uses the residual SeedTracks and UTHits from ``'Match'`` tracks. This is expected to a significant speedup for track reconstruction with a moderate loss of efficiency.
Returns:
A dict mapping all types of velo, upstream, HLT1 forward fitted, HLT2 forward, SciFi seeding, downstream and matched long tracks to ``'Velo'``, ``'Upstream'``, ``'ForwardFastFitted'``, ``'Forward'``, ``'Seed'``, ``'Downstream'`` and ``'Match'``` respectively.
"""
hlt1_tracks = make_legacy_rec_hlt1_tracks()
fitted_hlt1_tracks = make_legacy_rec_hlt1_fitted_tracks(hlt1_tracks)
scifi_tracks = make_PrHybridSeeding_tracks()
match_tracks = make_PrMatchNN_tracks(hlt1_tracks["Velo"], scifi_tracks)
residual_velo = make_ReduceVeloTracks_fromLong(match_tracks,
hlt1_tracks["Velo"])
make_residual_fthits = partial(make_ReduceSciFiHits_fromLong, match_tracks)
make_residual_uthits = partial(make_ReducePrUTHits_fromLong, match_tracks)
get_residual_ut_hits_tool = partial(
get_global_ut_hits_tool, make_ut_hits=make_residual_uthits)
make_residual_forward = partial(
make_PrForwardTrackingVelo_tracks,
make_ft_hits=make_residual_fthits,
ut_hits_tool=get_residual_ut_hits_tool)
hlt2_forward_tracks = all_hlt2_forward_track_types(
residual_velo, make_forward_tracks=make_residual_forward)
scifi_tracks_reduce = make_ReduceSeedTracks_fromMatch(
match_tracks, scifi_tracks)
downstream_tracks = make_PrLongLivedTracking_tracks(
scifi_tracks_reduce, make_ut_hits=make_residual_uthits)
return {
"Velo": hlt1_tracks["Velo"],
"Upstream": hlt1_tracks["Upstream"],
"ForwardFastFitted": fitted_hlt1_tracks,
"Forward": hlt2_forward_tracks,
"Seed": scifi_tracks,
"Downstream": downstream_tracks,
"Match": match_tracks,
"UTHitsRes": make_residual_uthits(),
"FTHitsRes": make_residual_fthits(),
}
[docs]def get_fast_hlt2_tracks_without_UT():
"""Function to get fast set of tracks reconstructed without UT in HLT2
The PrForwardTracking uses the residual VeloTracks and SciFi hits from ``'Match'`` tracks. This is expected to a significant speedup for track reconstruction with a moderate loss of efficiency.
Returns:
A dict mapping all types of velo, HLT2 forward, SciFi seeding and matched long tracks to ``'Velo'``, ``'Forward'``, ``'Seed'`` and ``'Match'``` respectively.
"""
velo_tracks = all_velo_track_types()
scifi_tracks = make_PrHybridSeeding_tracks()
match_tracks = make_PrMatchNN_tracks(
velo_tracks,
scifi_tracks,
ut_hits_tool=get_ignore_ut_hits_tool,
add_ut_hits=False)
residual_velo = make_ReduceVeloTracks_fromLong(match_tracks, velo_tracks)
make_residual_fthits = partial(make_ReduceSciFiHits_fromLong, match_tracks)
make_residual_forward = partial(
make_PrForwardTrackingVelo_tracks,
make_ft_hits=make_residual_fthits,
ut_hits_tool=get_ignore_ut_hits_tool,
add_ut_hits=False)
hlt2_forward_tracks = all_hlt2_forward_track_types(
residual_velo, make_forward_tracks=make_residual_forward)
return {
"Velo": velo_tracks,
"Forward": hlt2_forward_tracks,
"Seed": scifi_tracks,
"Match": match_tracks,
"FTHitsRes": make_residual_fthits(),
}
[docs]def get_persistable_tracks_per_type(hlt2_tracks):
"""Return dictionary of tracks that can be configured to be persisted per track type
generally meant for persistreco
"""
persisted_tracks_config = {
'Long': {
'keys': {'BestLong'},
'location': 'LongTracks'
},
'Downstream': {
'keys': {'BestDownstream'},
'location': 'DownstreamTracks'
},
'Upstream': {
'keys': {'BestUpstream'},
'location': 'UpstreamTracks'
},
'Ttrack': {
'keys': {'SeedDecloned'},
'location': 'Ttracks'
},
'Velo': {
'keys': {'Velo'},
'location': 'VeloTracks'
},
'FittedVelo': {
'keys': {'BestVelo', 'BestVeloBackward'},
'location': 'FittedVeloTracks',
},
}
tracks = {}
persistable_locs = persistable_locations()
available_track_types = set(hlt2_tracks.keys())
for track_type, config in persisted_tracks_config.items():
should_persist = config['location'] in persistable_locs
if config['keys'] <= available_track_types:
# TODO preferably this should be an algorithm that makes SharedObjectsContainers
# of them (but needs this type to be persistable) or set at moment of producing
in_tracks = [hlt2_tracks[key]['v1'] for key in config['keys']]
tracks[track_type] = TrackContainersMerger(
name=f"Persistable{track_type}TracksContainer" + "_{hash}",
InputLocations=in_tracks,
outputs={
'OutputLocation':
persistable_location(config['location'])
if should_persist else None
}).OutputLocation
elif should_persist:
# in case one should generally persist these, but don't exist (e.g. no UT)
tracks[track_type] = TracksEmptyProducer(
name=f"Persistable{track_type}EmptyTracksContainer" +
"_{hash}",
allow_duplicate_instances_with_distinct_names=True,
outputs={
'Output': persistable_location(config['location'])
}).Output
return tracks
def convert_tracks_to_v3_from_v1(tracks_v1,
track_types=None,
shared_container=False):
tracks = dict()
trackrels = dict()
Converter = TrackSOAFromSharedV1 if shared_container else TrackSOAFromV1
relevant_track_types = track_types if track_types is not None else list(
tracks_v1.keys())
for tracktype in relevant_track_types:
trackconverter = Converter(
InputTracks=tracks_v1[tracktype]
if type(tracks_v1) is dict else tracks_v1,
InputUniqueIDGenerator=make_unique_id_generator(),
RestrictToType=tracktype)
tracks[tracktype] = trackconverter.OutputTracks
trackrels[tracktype] = trackconverter.Relations
return tracks, trackrels
@configurable
def make_velo_heavyflavour_tracks(composites,
pvs,
max_distance_from_pvsv=0.5 * mm,
revert_charge=False):
vp_hits = make_VeloClusterTrackingSIMD_hits()
btracking_alg = VeloHeavyFlavourTrackFinder(
Composites=composites,
PVs=pvs,
Hits=vp_hits,
MaxDistanceFromPVSV=max_distance_from_pvsv,
RevertCharge=revert_charge)
return {"VeloHeavyFlavourTrackFinder": btracking_alg, "VPHits": vp_hits}