from typing import List
import pandas as pd
import basf2
import modularAnalysis as ma
import vertex as vx
import flavorTagger as ft
from vibe.core.utils.misc import fancy_validation_mode_header
from vibe.core.validation_mode import ValidationModeBaseClass
from vibe.core.helper.histogram_tools import HistVariable, Histogram, HistComponent
from vibe.core.helper.root_helper import makeROOTCompatible
__all__ = [
"BtoJpsiKstarValidationMode",
]
[docs]
@fancy_validation_mode_header
class BtoJpsiKstarValidationMode(ValidationModeBaseClass):
name = "BtoJpsiKstar"
latex_str = r"$B \rightarrow J/\psi K^{*0}$"
[docs]
def create_basf2_path(self):
main_path = basf2.Path()
# fsp selection
goodTrack = "abs(dz) < 2 and dr < 0.5"
muons = ("mu-:rec", f"{goodTrack} and p > 0.7 and muonID > 0.9")
pions = ("pi-:rec", f"{goodTrack} and pionID > 0.5")
kaons = ("K-:rec", f"{goodTrack} and kaonID > 0.5")
ma.fillParticleLists(
decayStringsWithCuts=[muons, pions, kaons],
path=main_path,
)
# J/psi reconstruction
ma.reconstructDecay(
decayString="J/psi:mu_rec -> mu-:rec mu+:rec",
cut="2.9869 < M < 3.2069",
path=main_path,
)
# Kstar reconstruction
ma.reconstructDecay(
decayString="K*0:rec -> pi-:rec K+:rec",
cut="0.79555 < M < 0.99555",
path=main_path,
)
# B meson reconstruction
ma.reconstructDecay(
decayString="B0:rec -> J/psi:mu_rec K*0:rec",
cut="",
path=main_path,
)
# MC matching + vertex + kinematic cuts
ma.matchMCTruth(
"B0:rec",
path=main_path,
)
vx.treeFit(
list_name="B0:rec",
conf_level=0,
updateAllDaughters=True,
ipConstraint=True,
massConstraint=[443],
path=main_path,
)
ma.applyCuts("B0:rec", "Mbc>5.27 and abs(deltaE)<0.1", path=main_path)
# select signal + build ROE
cleanMask = ("cleanMask", "nCDCHits > 0 and useCMSFrame(p)<=3.2", "p >= 0.05 and useCMSFrame(p)<=3.2")
ma.buildRestOfEvent("B0:rec", path=main_path)
ma.appendROEMasks("B0:rec", mask_tuples=[cleanMask], path=main_path)
ma.buildContinuumSuppression("B0:rec", roe_mask="cleanMask", path=main_path)
ma.buildRestOfEvent("B0:rec", path=main_path)
tagVMaskCut = "dr < 5 and abs(dz) < 10 and nCDCHits > 0 and nSVDHits > 0 and p >= 0.05"
ma.buildRestOfEvent("B0:rec", path=main_path)
ma.appendROEMask(
list_name="B0:rec", mask_name="tagVMask", trackSelection=tagVMaskCut, eclClusterSelection="", path=main_path
)
vx.TagV(
list_name="B0:rec",
MCassociation="breco",
constraintType="tube",
confidenceLevel=-2,
maskName="tagVMask",
fitAlgorithm="Rave",
path=main_path,
)
ft.flavorTagger(particleLists=["B0:rec"], useGNN=True, path=main_path)
ma.fillParticleList(decayString="pi+:goodtracks", cut="pt> 0.1", path=main_path)
ma.fillParticleList(decayString="gamma:mdst", cut="E > 0.1", path=main_path)
ma.buildEventShape(
inputListNames=["pi+:goodtracks", "gamma:mdst"],
allMoments=True,
foxWolfram=True,
harmonicMoments=True,
cleoCones=True,
thrust=True,
collisionAxis=True,
jets=True,
sphericity=True,
checkForDuplicates=False,
path=main_path,
)
var = [
"isSignal",
"isSignalAcceptBremsPhotons",
"deltaE",
"Mbc",
"DeltaT",
"DeltaTErr",
"mcDeltaT",
"mcDeltaTau",
"beamE",
"Ecms",
"beamPx",
"beamPy",
"beamPz",
"cosAngleBetweenMomentumAndBoostVector",
"DeltaT3D",
"chiProb",
"TagVNDF",
"TagVpVal",
"qrMC",
"mcFlavorOfOtherB",
"qrGNN",
"FBDT_qrCombined",
"TagVz",
"mcTagVz",
"z",
"mcDecayVertexZ",
]
self.variables_to_validation_ntuple(
decay_str="B0:rec",
variables=var,
path=main_path,
)
return main_path
@property
def analysis_validation_histograms(self) -> List[Histogram]:
return [
Histogram(
name="qrGNN",
title="",
hist_variable=HistVariable(
df_label=makeROOTCompatible(variable="qrGNN"),
label=r"$qrGNN$ (cleaned ROE)",
unit=r"unitless",
bins=50,
scope=(-1, 1),
),
hist_components=[
HistComponent(
label="All",
),
],
),
Histogram(
name="tagvzres",
title=r"$B^0 \rightarrow J/\psi K^{*0}$",
hist_variable=HistVariable(
df_label=makeROOTCompatible(variable="tagvzres"),
label=r"tag vertex z residual",
unit=r"$\mu$m",
bins=100,
scope=(-300, 300),
),
hist_components=[
HistComponent(
label="All",
additional_cut_str="isSignalAcceptBremsPhotons==1",
),
],
),
Histogram(
name="zres",
title=r"$B^0 \rightarrow J/\psi K^{*0}$",
hist_variable=HistVariable(
df_label=makeROOTCompatible(variable="zres"),
label=r"vertex z residual",
unit=r"$\mu$m",
bins=100,
scope=(-300, 300),
),
hist_components=[
HistComponent(
label="All",
additional_cut_str="isSignalAcceptBremsPhotons==1",
),
],
),
Histogram(
name="deltatres",
title=r"$B^0 \rightarrow J/\psi K^{*0}$",
hist_variable=HistVariable(
df_label=makeROOTCompatible(variable="deltatres"),
label=r"$\Delta t$ residual",
unit=r"ps",
bins=100,
scope=(-10, 10),
),
hist_components=[
HistComponent(
label="All",
additional_cut_str="isSignalAcceptBremsPhotons==1",
),
],
),
Histogram(
name="effective_tagging_efficiency",
title=r"$B^0 \rightarrow J/\psi K^{*0}$",
hist_variable=HistVariable(
df_label=makeROOTCompatible(variable="effective_tagging_efficiency"),
label=r"effective tagging efficiency",
unit=r"%",
bins=100,
scope=(0, 50),
),
hist_components=[
HistComponent(
label="All",
),
],
),
Histogram(
name="effective_tagging_efficiency error",
title=r"$B^0 \rightarrow J/\psi K^{*0}$",
hist_variable=HistVariable(
df_label=makeROOTCompatible(variable="effective_tagging_efficiency_error"),
label=r"effective tagging efficiency error",
unit=r"%",
bins=100,
scope=(0, 10),
),
hist_components=[
HistComponent(
label="All",
),
],
),
Histogram(
name="deltaTErr",
title=r"$B^0 \rightarrow J/\psi K^{*0}$",
hist_variable=HistVariable(
df_label=makeROOTCompatible(variable="DeltaTErr"),
label=r"Delta T error",
unit=r"ps",
bins=100,
scope=(0, 2.5),
),
hist_components=[
HistComponent(
label="All",
),
],
),
]
[docs]
def offline_df_manipulation(self, df: pd.DataFrame) -> pd.DataFrame:
df = df.query("TagVpVal > 0 and chiProb > 0 and TagVNDF>0.5") # apply vertex quality cuts
df["tagvzres"] = 1e4 * (df["TagVz"] - df["mcTagVz"])
df["zres"] = 1e4 * (df["z"] - df["mcDecayVertexZ"])
df["deltatres"] = df["DeltaT"] - df["mcDeltaT"]
effeff, effeff_err = getEffEff(df)
df["effective_tagging_efficiency"] = [100 * effeff] * df.shape[0]
df["effective_tagging_efficiency_error"] = [100 * effeff_err] * df.shape[0]
return df
def getEffEff(df):
import numpy as np
qrbins = [0.0, 0.1, 0.25, 0.45, 0.6, 0.725, 0.875, 1.0]
cut = "abs(qrMC)==1"
total = len(df.query(cut))
qr_var = "FBDT_qrCombined"
q_true = "qrMC"
effeff = 0
effeff_err = 0
for r_i in range(len(qrbins) - 1):
qr_cut = f"{cut} & {qrbins[r_i]}<abs({qr_var}) & abs({qr_var})<{qrbins[r_i+1]}"
wrong = len(df.query(f"{qr_cut} & {qr_var}*{q_true} < 0"))
correct = len(df.query(f"{qr_cut} & {qr_var}*{q_true} > 0"))
dilution = 1 - 2 * wrong / (wrong + correct)
effeff += (wrong + correct) / total * dilution * dilution
dilution2_err = 2 * dilution * np.sqrt((wrong * correct) / (wrong + correct) ** 3)
# correlation is ignored here
effeff_err += (wrong + correct) / total * dilution2_err + np.sqrt(
(wrong + correct) * (total - (wrong + correct)) / (total) ** 3
) * dilution * dilution
return effeff, effeff_err
