import os
from typing import List
from typing import Optional
from pydantic import BaseModel
from pydantic import Field
from pydantic import model_validator
from . import t4_camera
[docs]
class RunConfig(BaseModel, extra="forbid"):
input_root_directory: str = Field(
...,
description="Root directory containing raw or processed run data.",
)
run_ids: List[int] = Field(
...,
description="List of run IDs to process.",
)
input_basename: str = Field(
...,
description="Template of the t4r or bin file base names with run and suffix placeholders.",
)
input_suffixes: List[str] = Field(
["BOT", "TOP"],
description="Suffixes .",
)
output_directory: str = Field(
...,
description="Base name for generated output files.",
)
from_t4r: bool = Field(
True,
description="If True, input data is read from T4R formatted files.",
)
exclude_pileup: bool = Field(
False,
description="If True, events flagged as pile-up are excluded from analysis.",
)
filtering: bool = Field(
False,
description="If True, 'blob' events are filtered out before histogramming.",
)
filter_radius: float = Field(
50,
description="Radial filter size in pixels.",
gt=0,
)
filter_x_bin: int = Field(
5,
description="Histogram bin size in X direction (pixels).",
gt=0,
)
filter_y_bin: int = Field(
5,
description="Histogram bin size in Y direction (pixels).",
gt=0,
)
filter_toa_bin: float = Field(
0.5,
description="Time-of-arrival (TOA) histogram bin width in nanoseconds.",
gt=0,
)
rel_toa_min: Optional[float] = Field(
None,
description="Minimum TOA cut in nanoseconds. If None, no lower bound is applied.",
)
rel_toa_max: Optional[float] = Field(
None,
description="Maximum TOA cut in nanoseconds. If None, no upper bound is applied.",
)
frequency: float = Field(
0.355e6,
description=("Synchrotron revolution frequency in Hz."),
gt=0,
)
@property
def input_basenames(self) -> List[str]:
return [
self.input_basename.format(run_id=i, suffix=suffix)
for i in self.run_ids
for suffix in self.input_suffixes
]
@property
def input_basename_groups(self) -> List[List[str]]:
return [
[
self.input_basename.format(run_id=i, suffix=suffix)
for suffix in self.input_suffixes
]
for i in self.run_ids
]
@property
def input_t4r_paths(self) -> List[str]:
return [
os.path.join(self.input_root_directory, "PROCESSED_DATA", f"{basename}.t4r")
for basename in self.input_basenames
]
@property
def input_bin_paths(self) -> List[str]:
return [
os.path.join(self.input_root_directory, "RAW_DATA", f"{basename}.bin")
for basename in self.input_basenames
]
@property
def input_t4r_path_groups(self) -> List[str]:
return [
[
os.path.join(
self.input_root_directory, "PROCESSED_DATA", f"{basename}.t4r"
)
for basename in names
]
for names in self.input_basename_groups
]
@property
def input_bin_path_groups(self) -> List[str]:
return [
[
os.path.join(self.input_root_directory, "RAW_DATA", f"{basename}.bin")
for basename in names
]
for names in self.input_basename_groups
]
@property
def period(self) -> float:
"""
Bunch period in nanoseconds.
"""
return 1e9 / self.frequency
@property
def revolution(self) -> float:
"""
Revolution time in microseconds.
"""
return self.period * 1e-3 # ns -> µs
@property
def rel_toa_bin_size(self) -> int:
"""
TOA bin size based on the TOA resolution.
"""
return t4_camera.toa_resolution / 8
[docs]
@model_validator(mode="after")
def validate_toa_range(self):
if self.rel_toa_min is None:
self.rel_toa_min = 0
if self.rel_toa_max is None:
self.rel_toa_max = self.period
if self.rel_toa_min >= self.rel_toa_max:
raise ValueError("toa_min must be smaller than toa_max")
return self
[docs]
def get_output_path(self, *args) -> str:
"""
Return output path in existing parent directory.
"""
filename = os.path.join(self.output_directory, *args)
dirname = os.path.dirname(filename)
if dirname:
os.makedirs(dirname, exist_ok=True)
return filename