Source code for RecoConf.legacy_rec_hlt1_tracking

###############################################################################
# (c) Copyright 2019 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.                                       #
###############################################################################
import logging
from PyConf import configurable
from PyConf.application import default_raw_banks, make_odin
from PyConf.packing import persistable_location

from PyConf.Algorithms import (
    fromPrUpstreamTracksV1Tracks, fromPrForwardTracksFromVeloUTV1Tracks,
    fromV3TrackV1Track, fromPrVeloTracksV1TracksMerger, FTRawBankDecoder,
    PrVPHitsToVPLightClusters, VeloClusterTrackingSIMD, VPRetinaSPmixer,
    VPRetinaClusterCreator, VPRetinaClusterDecoder,
    VeloRetinaClusterTrackingSIMD, TrackBeamLineVertexFinderSoA, PrVeloUT,
    SciFiTrackForwardingStoreHit, SciFiTrackForwarding, VeloKalman,
    MakePVRelations__PrFittedForwardTracks,
    MakeZip__PrFittedForwardTracks__BestVertexRelations, PrStorePrUTHits,
    PatPV3DFuture, PrStorePrUTHitsEmptyProducer, VPClusFull,
    VPRetinaFullClusterDecoder, MakeZip__BestTracks__MuonPIDs__v2)

from RecoConf.decoders import default_VeloCluster_source

from Functors import ALL

log = logging.getLogger(__name__)

from RecoConf.core_algorithms import make_unique_id_generator

from RecoConf.muonid import make_muon_id


def get_default_VeloFullCluster_algorithm():
    def_source = default_VeloCluster_source()
    if def_source == "VPRetinaCluster":
        return VPRetinaFullClusterDecoder
    return VPClusFull


def get_default_VeloClusterTracking_algorithm():
    def_source = default_VeloCluster_source()
    if def_source == "VPRetinaCluster":
        return VeloRetinaClusterTrackingSIMD
    return VeloClusterTrackingSIMD


[docs]@configurable def make_VeloClusterTrackingSIMD(algorithm=None, make_raw=default_raw_banks, masked_sensors=[], MaxScatterSeeding=0.1, MaxScatterForwarding=0.1, MaxScatter3hits=0.02, SkipForward=1): """Simple helper to make sure both, make_VeloClusterTrackingSIMD_tracks and make_VeloClusterTrackingSIMD_hits, access the identically configured version of VeloClusterTrackingSIMD Args: make_raw (DataHandle): RawEventLocation for VeloClusterTrackingSIMD, defaults to `default_raw_banks <PyConf.application.default_raw_banks>`; algorithm: The Velo tracking algorithm to run; masked_sensors: List of unique sensor IDs that will *not* be considered in the pattern reco's *tracks*, but are included in the *clusters* returned. Returns: The Velo tracking algorithm. """ if algorithm is None: algorithm = get_default_VeloClusterTracking_algorithm() if "Retina" in str(algorithm): # For VeloRetinaClusterTrackingSIMD use VPRetinaClusters rawbank in the file bank_type = "VPRetinaCluster" else: # In case raw_event is made from VP or algorithm is VeloClusterTrackingSIMD, use VP rawbank in the event bank_type = "VP" my_SensorMasks = [j in masked_sensors for j in range(208) ] # 208 = LHCb::Pr::Velo::VPInfos::NSensors return algorithm( RawBanks=make_raw(bank_type), SensorMasks=tuple(my_SensorMasks), MaxScatterSeeding=MaxScatterSeeding, MaxScatterForwarding=MaxScatterForwarding, MaxScatter3hits=MaxScatter3hits, SkipForward=SkipForward)
[docs]@configurable def make_VeloClusterTrackingSIMD_hits( make_tracks=make_VeloClusterTrackingSIMD): """Makes velo hits with VeloClusterTrackingSIMD Args: make_tracks (DatHandle): The function that returns the Velo tracking algorithm. Returns: DataHandle: VeloClusterTrackingSIMD's HitsLocation. """ return make_tracks().HitsLocation
[docs]def make_VPClus_location_and_offsets(): """Makes velo clusters with VPClus Returns: A dict mapping VPClus' ClusterLocation and ClusterOffsets DataHandles to ``'Location'`` and ``'Offsets'`` respectively. """ clustering = PrVPHitsToVPLightClusters( HitsLocation=make_VeloClusterTrackingSIMD_hits()) return { "Location": clustering.ClusterLocation, "Offsets": clustering.ClusterOffsets }
@configurable def make_velo_full_clusters(make_raw=default_raw_banks, make_full_cluster=None, detector=None): if make_full_cluster is None: make_full_cluster = get_default_VeloFullCluster_algorithm() expected_detector = { 'VPClusFull': 'VP', 'VPRetinaFullClusterDecoder': 'VPRetinaCluster', 'VPRetinaFullClustering': 'VP' }.get(make_full_cluster.type.getType(), None) if expected_detector: assert not detector or detector == expected_detector, f"explicitly specified detector will not work: {detector} vs {expected_detector} for {make_full_cluster.type.getType()}" if not detector: detector = expected_detector assert detector, "do not know which rawbank type {make_full_cluster.type.getType()} needs -- and it is not explicitly specified" return make_full_cluster(RawBanks=make_raw(detector)).ClusterLocation
[docs]@configurable def make_SPmixed_raw_banks(detector=None, make_raw=default_raw_banks): """Mix SP. Args: make_raw (DataHandle): RawBanks for VPRetinaSPmixer, defaults to `default_raw_banks <PyConf.application.default_raw_banks>`. Returns: """ # Can detector ever be something else? assert not detector or detector == "VP", detector return VPRetinaSPmixer(RawBanks=make_raw("VP")).MixedRawBanks
[docs]@configurable def make_RetinaCluster_raw_bank(detector=None, make_raw=default_raw_banks): """Makes velo clusters with VPRetinaClusterCreator. Args: make_raw (DataHandle): RawEventLocation for VPRetinaClusterCreator, defaults to `default_raw_vent <PyConf.application.default_raw_banks>`. Returns: """ assert not detector or detector == [ "VPRetinaCluster" ] or detector == "VPRetinaCluster", detector return VPRetinaClusterCreator(RawBanks=make_raw("VP")).RetinaRawBanks
[docs]@configurable def make_RetinaClusters(detector=None, make_raw=default_raw_banks): """Decodes RetinaClusters with VPRetinaClusterDecoder. Args: make_raw (DataHandle): RawBankLocation for VPRetinaClusterCreator, defaults to `default_raw_banks <PyConf.application.default_raw_banks>`. Returns: vector<LHCb::VPLightCluster> """ assert not detector or detector == 'VPRetinaCluster', detector return VPRetinaClusterDecoder( RawBanks=make_raw('VPRetinaCluster')).ClusterLocation
[docs]@configurable def make_VeloClusterTrackingSIMD_tracks( make_tracks=make_VeloClusterTrackingSIMD): """Makes velo tracks with VeloClusterTrackingSIMD. Args: make_tracks (DataHandle): The function that returns the Velo tracking algorithm. Returns: A dict mapping forward- and backward-going velo tracks to ``'Pr'`` and ``'Pr::backward'`` respectively. """ tracking = make_tracks() return { "Pr": tracking.TracksLocation, "Pr::backward": tracking.TracksBackwardLocation, }
[docs]@configurable def filter_velo_tracks(make_velo_tracks=make_VeloClusterTrackingSIMD_tracks, fwdTracksFunctor=ALL, bwdTracksFunctor=ALL): """Helper function to filter the Velo tracks Args: make_velo_tracks (DataHandle): velo tracking algorithm, defaults to `make_VeloClusterTrackingSIMD_tracks`. fwdTracksFunctor: Functor to filter the forward Velo tracks bwdTracksFunctor: Functor to filter the backward Velo tracks Returns: A dict mapping forward- and backward-going tracks to ``'Pr'`` and ``'Pr::backward'``, respectively. """ from Hlt1Conf.algorithms import Filter velo_tracks = make_velo_tracks() fiddlyDict = { "PrVeloTracks": velo_tracks['Pr'], "PrVeloBackwardTracks": velo_tracks['Pr::backward'], } forward_going_tracks = Filter(fiddlyDict, fwdTracksFunctor)['PrVeloTracks'] backward_going_tracks = Filter(fiddlyDict, bwdTracksFunctor)['PrVeloBackwardTracks'] return { "Pr": forward_going_tracks, "Pr::backward": backward_going_tracks, }
[docs]@configurable def all_velo_track_types(make_velo_tracks=make_VeloClusterTrackingSIMD_tracks): """Helper function to get all types of velo tracks. Args: make_velo_tracks (DataHandle): velo tracking algorithm, defaults to `make_VeloClusterTrackingSIMD_tracks`. Returns: A dict mapping forward-, backward-going, v1 and v2 velo tracks to ``'Pr'``, ``'Pr::backward'``, ``'v1'`` and ``'v2'`` respectively. """ velo_tracks = make_velo_tracks() velo_tracks_v1 = fromPrVeloTracksV1TracksMerger( InputTracksLocation1=velo_tracks["Pr"], InputTracksLocation2=velo_tracks["Pr::backward"]).OutputTracksLocation return { "Pr": velo_tracks["Pr"], "Pr::backward": velo_tracks["Pr::backward"], "v1": velo_tracks_v1 }
[docs]@configurable def make_TrackBeamLineVertexFinderSoA_pvs(velo_tracks, location, minz=None): """Makes primary vertices from velo tracks using TrackBeamLineVertexFinderSoA. Args: velo_tracks (dict): velo tracks, needs ``'Pr'`` and ``'Pr::backward'`` tracks, e.g. from `all_velo_track_types`. Returns: DataHandle: TrackBeamLineVertexFinderSoA's OutputVertices. """ pvs = { "v3": TrackBeamLineVertexFinderSoA( TracksLocation=velo_tracks["Pr"], TracksBackwardLocation=velo_tracks["Pr::backward"], MinZ=minz if minz is not None else TrackBeamLineVertexFinderSoA.getDefaultProperties()["MinZ"], outputs={ 'OutputVertices': None }, ).OutputVertices } if "v1" in velo_tracks: from PyConf.Algorithms import PVToRecConverterV1 pvs["v1"] = PVToRecConverterV1( InputVertices=pvs["v3"], InputTracks=velo_tracks["v1"], outputs={ 'OutputVertices': location }, ).OutputVertices else: from PyConf.Algorithms import PVToRecConverterV1WithoutTracks pvs["v1"] = PVToRecConverterV1WithoutTracks( InputVertices=pvs["v3"], AddTrackWeights=False, outputs={ 'OutputVertices': location }, ).OutputVertices return pvs
[docs]@configurable def make_PatPV3DFuture_pvs(velo_tracks, location, use_beam_spot_cut=True, velo_open=False, use_3D_seeding=False): """Makes primary vertices from velo tracks using PatPV3DFuture. Args: velo_tracks (dict): velo tracks, needs ``'v2'`` tracks, e.g. from `all_velo_track_types` Returns: DataHandle: PatPV3DFuture's OutputVerticesName Note: PatPV3DFuture's defaults have been overridden in this maker with ``BeamSpotRCut=0.6, UseBeamSpotRCut=True, minClusterMult=4``. """ if not velo_open: if use_3D_seeding: pvs = { "v1": PatPV3DFuture( InputTracks=velo_tracks["v1"], MinTracks=4, BeamSpotRCut=0.3, BeamSpotRHighMultiplicityCut=0.3, minClusterMult=4.0, UseBeamSpotRCut=use_beam_spot_cut, Seeding3D=True, zMaxSpread=0.9, radialDistanceSeeding=0.2, TrackPairMaxDistance=0.2, maxChi2Merge=3.0, factorToIncreaseErrors=10.0, outputs={ 'OutputVerticesName': location }, ).OutputVerticesName } else: pvs = { "v1": PatPV3DFuture( InputTracks=velo_tracks["v1"], MinTracks=4, BeamSpotRCut=0.3, UseBeamSpotRCut=use_beam_spot_cut, minClusterMult=4, BeamSpotRHighMultiplicityCut=0.3, outputs={ 'OutputVerticesName': location }, ).OutputVerticesName } else: if use_3D_seeding: pvs = { "v1": PatPV3DFuture( InputTracks=velo_tracks["v1"], MinTracks=3, BeamSpotRCut=0.3, minClusterMult=2.5, trackMaxChi2Remove=30, trackMaxChi2=20, UseBeamSpotRCut=use_beam_spot_cut, Seeding3D=True, zMaxSpread=3.0, radialDistanceSeeding=1.0, TrackPairMaxDistance=0.4, maxChi2Merge=3.0, TrackPairMaxCos2Theta=0.99, factorToIncreaseErrors=10.0, outputs={ 'OutputVerticesName': location }, ).OutputVerticesName } else: pvs = { "v1": PatPV3DFuture( InputTracks=velo_tracks["v1"], MinTracks=3, BeamSpotRCut=0.3, minClusterMult=2.5, trackMaxChi2Remove=30, trackMaxChi2=20, UseBeamSpotRCut=use_beam_spot_cut, outputs={ 'OutputVerticesName': location }, ).OutputVerticesName } from PyConf.Algorithms import RecV1ToPVConverter pvs["v3"] = RecV1ToPVConverter( InputVertices=pvs["v1"], outputs={ 'OutputVertices': None }, ).OutputVertices return pvs
[docs]@configurable def make_reco_pvs( velo_tracks, location, make_pvs_from_velo_tracks=make_TrackBeamLineVertexFinderSoA_pvs): """Makes PVs from velo tracks given a PV algorithm. Args: velo_tracks (dict): velo tracks, e.g. from `all_velo_track_types`. make_pvs_from_velo_tracks (DataHandle): PV maker consuming ``velo_tracks``, defaults to `make_TrackBeamLineVertexFinderSoA_pvs`. Returns: DataHandle: Output of ``make_pvs_from_velo_tracks`` """ return make_pvs_from_velo_tracks(velo_tracks, location)
[docs]def make_all_pvs(): """Makes PVs from HLT1 inputs, i.e. from `all_velo_track_types` using the PV maker passed to `make_reco_pvs` Returns: dictionary of v1 and v3 PVs """ return make_reco_pvs(all_velo_track_types(), persistable_location('PVs'))
[docs]def make_pvs(): """Makes PVs from HLT1 inputs, i.e. from `all_velo_track_types` using the PV maker passed to `make_reco_pvs` Returns: DataHandle: primary vertices """ return make_all_pvs()["v3"]
def require_pvs(pvs): raise DeprecationWarning( "Function require_pvs has to be imported from the RecoConf.event_filters module." )
[docs]@configurable def make_PrStorePrUTHits_hits(make_raw=default_raw_banks, isCluster=True, positionMethod="GeoWeighting"): """Decodes UT hits from raw data, fills an SOA container. Args: make_raw (DataHandle): RawEventLocation for VeloClusterTrackingSIMD, defaults to `default_raw_event <PyConf.application.default_raw_event>`. isCluster (bool): True to enable clustering, defaults to True. positionMethod (str): available options "MaxAdc", "AdcWeighting", "GeoWeighting", defaults to "GeoWeighting" Returns: DataHandle: PrStorePrUTHits' UTHitsLocation. """ return PrStorePrUTHits( RawBanks=make_raw("UT"), isCluster=isCluster, positionMethod=positionMethod).UTHitsLocation
[docs]@configurable def make_PrStorePrUTHits_empty_hits(): """Creates an empty container of UT hits, used for the no UT scenario. Returns: DataHandle: PrStorePrUTHitsEmptyProducer' Output. """ return PrStorePrUTHitsEmptyProducer().Output
[docs]@configurable def make_PrVeloUT_tracks(velo_tracks, make_ut_hits=make_PrStorePrUTHits_hits): """Makes upstream tracks from velo tracks and UT hits. Args: velo_tracks (dict): velo tracks, needs ``'Pr'`` tracks, e.g. from `all_velo_track_types`. make_ut_hits (DataHandle): UT hit maker, defaults to `make_PrStorePrUTHits_hits`. Returns: DataHandle: PrVeloUT's OutputTracksName. """ return PrVeloUT( InputTracksName=velo_tracks["Pr"], UTHits=make_ut_hits()).OutputTracksName
[docs]@configurable def all_upstream_track_types(velo_tracks, make_velo_ut_tracks=make_PrVeloUT_tracks): """Helper function to get all types of upstream tracks. Args: velo_tracks (dict): velo tracks, needs ``'v2'`` tracks, e.g. from `all_velo_track_types`. make_velo_ut_tracks (DataHandle): upstream track maker, defaults to `make_PrVeloUT_tracks`. Returns: A dict mapping Pr, v1 and v2 upstream tracks to ``'Pr'``, ``'v1'`` and ``'v2'`` respectively. """ upstream_tracks_pr = make_velo_ut_tracks(velo_tracks=velo_tracks) upstream_tracks_v1 = fromPrUpstreamTracksV1Tracks( InputTracksLocation=upstream_tracks_pr, VeloTracksLocation=velo_tracks["v1"]).OutputTracksLocation return {"Pr": upstream_tracks_pr, "v1": upstream_tracks_v1}
[docs]def make_FTRawBankDecoder_clusters(make_raw=default_raw_banks): """Decodes the FT raw bank into FTLiteClusters. DecodingVersion set by `default_ft_decoding_version`. Args: make_raw (DataHandle): RawEventLocation for VeloClusterTrackingSIMD, defaults to `default_raw_banks <PyConf.application.default_raw_banks>`. Returns: DataHandle: FTRawBankDecoder's OutputLocation. """ return FTRawBankDecoder( name="FTRawBankDecoder", Odin=make_odin(), RawBanks=make_raw("FTCluster")).OutputLocation
[docs]@configurable def make_SciFiTrackForwardingStoreHit_hits( make_ft_clusters=make_FTRawBankDecoder_clusters): """Transforms FTLiteClusters into the input format needed by the SciFiTrackForwarding. Args: make_ft_clusters (DataHandle): maker of FT clusters, defaults to `make_FTRawBankDecoder_clusters`. Returns: DataHandle: SciFiTrackForwardingStoreHit's Output. """ return SciFiTrackForwardingStoreHit(HitsLocation=make_ft_clusters()).Output
[docs]@configurable def make_SciFiTrackForwarding_tracks( input_tracks, make_ft_hits=make_SciFiTrackForwardingStoreHit_hits): """Makes forward tracks with SciFiTrackForwarding (long tracks from upstream seeds). Args: input_tracks (dict): upstream tracks, needs ``'Pr'`` tracks, e.g. from `all_upstream_track_types`. make_ft_hits (DataHandle): maker of FT hits, defaults to `make_SciFiTrackForwardingStoreHit_hits`. Returns: DataHandle: SciFiTrackForwarding's Output. """ return SciFiTrackForwarding( HitsLocation=make_ft_hits(), InputTracks=input_tracks["Pr"]).Output
[docs]@configurable def all_legacy_rec_hlt1_forward_track_types( input_tracks, make_forward_tracks=make_SciFiTrackForwarding_tracks): """Helper function to get all types of HLT1 forward tracks. Args: input_tracks (dict): upstream tracks, needs ``'v2'`` tracks, e.g. from `all_upstream_track_types`. make_forward_tracks (DataHandle): maker of forward tracks, defaults to `make_SciFiTrackForwarding_tracks`. Returns: A dict mapping Pr, v1, v2 and v2Zip HLT1 forward tracks to ``'Pr'``, ``'v1'``, ``'v2'`` and ``'v2Zip'`` respectively. """ forward_tracks_pr = make_forward_tracks(input_tracks=input_tracks) forward_tracks_v1 = fromPrForwardTracksFromVeloUTV1Tracks( InputTracksLocation=forward_tracks_pr, UpstreamTracksLocation=input_tracks["v1"]).OutputTracksLocation return { "Pr": forward_tracks_pr, "v1": forward_tracks_v1, }
[docs]def make_legacy_rec_hlt1_tracks(): """Function to get all types of tracks reconstructed in HLT1 Returns: A dict mapping all types of velo, upstream and HLT1 forward tracks to ``'Velo'``, ``'Upstream'`` and ``'Forward'`` respectively. """ velo_tracks = all_velo_track_types() velo_ut_tracks = all_upstream_track_types(velo_tracks) forward_tracks = all_legacy_rec_hlt1_forward_track_types(velo_ut_tracks) return { "Velo": velo_tracks, "Upstream": velo_ut_tracks, "Forward": forward_tracks, }
[docs]@configurable def make_VeloKalman_fitted_tracks( tracks, make_hits=make_VeloClusterTrackingSIMD_hits, make_unique_id_generator=make_unique_id_generator): """Fits tracks with VeloKalman. Args: tracks (dict of dicts): input tracks to VeloKalman. It will consume ``["Velo"]["Pr"]`` and ``["Forward"]["Pr"]`` tracks. make_hits (DataHandle): maker of velo hits, defaults to `make_VeloClusterTrackingSIMD_hits`. make_unique_id_generator (DataHandle): maker of the ID generator object, defaults to `make_unique_id_generator`. Returns: A dict mapping Pr, v1, v2, v1Sel, v2Sel, v2Zip and v2ZipSel fitted forward tracks to ``'Pr'``, ``'v1'``, ``'v2'``, ``'v1Sel'``, ``'v2Sel'``, ``'v2Zip'`` and ``'v2ZipSel'`` respectively. """ fitted_tracks = VeloKalman( HitsLocation=make_hits(), TracksVPLocation=tracks["Velo"]["Pr"], TracksFTLocation=tracks["Forward"]["Pr"], UniqueIDGenerator=make_unique_id_generator()).OutputTracksLocation fitted_tracks_v1 = fromV3TrackV1Track( InputTracks=fitted_tracks).OutputTracks return { "Pr": fitted_tracks, "v1": fitted_tracks_v1, }
[docs]@configurable def make_fitted_forward_tracks_with_pv_relations(tracks, make_pvs=make_pvs): """Zips fitted forward tracks with their best PV. Args: tracks (dict): fitted forward tracks as input to MakeZip__PrFittedForwardTracks__BestVertexRelations, needs ``["Pr"]`` tracks, e.g. from make_VeloKalman_fitted_tracks. make_pvs (DataHandle): maker of PVs, defaults to `make_pvs`. Returns: A dict mapping tracks zipped with PV relations to ``'PrFittedForwardWithPVs'``. """ pv_container = make_pvs() pr_in_tracks = tracks["Pr"] # PrFittedForwardTracks pv_relations = MakePVRelations__PrFittedForwardTracks( Input=pr_in_tracks, InputVertices=pv_container).Output trackrel_zip = MakeZip__PrFittedForwardTracks__BestVertexRelations( Input1=pr_in_tracks, Input2=pv_relations) return { "PrFittedForwardWithPVs": trackrel_zip.Output, }
[docs]@configurable def make_legacy_rec_hlt1_fitted_tracks( tracks, make_forward_fitted_tracks=make_VeloKalman_fitted_tracks): """Helper function to bind to for passing forward fitted tracks. Args: tracks (dict of dicts): input tracks to the track fitter, i.e. to ``make_forward_fitted_tracks``, e.g. from make_legacy_rec_hlt1_tracks. make_forward_fitted_tracks (DataHandle): track fitter, defaults to `make_VeloKalman_fitted_tracks`. Returns: A dict of fitted tracks. The content depends on the input maker """ return make_forward_fitted_tracks(tracks)
def make_tracks_with_muon_id(tracks, muon_ids): return MakeZip__BestTracks__MuonPIDs__v2( Input1=tracks["Pr"], Input2=muon_ids).Output def make_fitted_tracks_with_muon_id(): tracks = make_legacy_rec_hlt1_fitted_tracks(make_legacy_rec_hlt1_tracks()) muon_ids = make_muon_id(make_legacy_rec_hlt1_tracks()["Forward"]) tracks_with_muon_id = make_tracks_with_muon_id(tracks, muon_ids) return {'PrFittedForwardWithMuonID': tracks_with_muon_id}