Source code for analyzer.modules.common.electrons
from analyzer.core.analysis_modules import AnalyzerModule, MetadataExpr
from analyzer.core.columns import Column
from attrs import define, field
import enum
from analyzer.core.adl import ADLBlock, ADLStatement
[docs]
class CutBasedWPs(str, enum.Enum):
[docs]
cut_mapping = dict(fail=0, veto=1, loose=2, medium=3, tight=4)
@define
[docs]
class ElectronMaker(AnalyzerModule):
"""
Select electrons based on kinematics, cut-based ID, and isolation.
This analyzer filters electrons in an event according to minimum
transverse momentum, maximum pseudorapidity, cut-based ID working point,
and maximum mini-isolation.
Parameters
----------
input_col : Column
Column containing the input electron collection.
output_col : Column
Column where the selected electrons will be stored.
working_point : CutBasedWPs
Cut-based ID working point (fail, veto, loose, medium, tight).
min_pt : float, optional
Minimum transverse momentum in GeV, by default 10.
max_abs_eta : float, optional
Maximum absolute pseudorapidity, by default 2.4.
max_mini_iso : float, optional
Maximum mini-isolation, by default 0.1.
"""
[docs]
working_point: CutBasedWPs
[docs]
max_abs_eta: float = 2.4
[docs]
max_mini_iso: float = 0.1
__corrections: dict = field(factory=dict)
[docs]
def lint(self):
from analyzer.core.linting import LintLevel, LintMessage
messages = []
if self.min_pt is not None and self.min_pt < 5:
messages.append(
LintMessage(
level=LintLevel.WARNING,
category="ElectronKinematics",
message=f"Low electron pt cut: {self.min_pt} GeV.",
module_name=self.name(),
)
)
return messages
[docs]
def run(self, columns, params):
electrons = columns[self.input_col]
pass_pt = electrons.pt > self.min_pt
pass_eta = abs(electrons.eta) < self.max_abs_eta
pass_wp = electrons.cutBased >= cut_mapping[self.working_point]
pass_iso = electrons.miniPFRelIso_all < self.max_mini_iso
columns[self.output_col] = electrons[pass_pt & pass_eta & pass_wp & pass_iso]
return columns, []
[docs]
def outputs(self, metadata):
return [self.output_col]
[docs]
def adlExport(self, metadata):
statements = [
ADLStatement("take", self.input_col.adl_name),
ADLStatement("select", f"pt > {self.min_pt}"),
ADLStatement("select", f"abs(eta) < {self.max_abs_eta}"),
]
statements.append(
ADLStatement("select", f"cutBased >= {cut_mapping[self.working_point]}")
)
statements.append(
ADLStatement("select", f"miniPFRelIso_all < {self.max_mini_iso}")
)
return [
ADLBlock(
block_type="object",
name=self.output_col.adl_name,
statements=statements,
)
]