"""Pydantic models validating the ID13 YAML templates."""
import difflib
from pathlib import Path
from typing import Any, Dict, List, Optional, Union
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
PYFAI_CONFIG_SUFFIXES = (".poni", ".json")
BLISS_DATASET_SUFFIXES = (".h5", ".edf", ".tiff", ".tif")
# Additional pyFAI worker options accepted next to ``filename``/``tag``.
# TODO: replace the placeholder entries below with the authoritative list.
ALLOWED_PYFAI_OPTIONS = frozenset(
{
"npt_rad",
"npt_azim",
"azimuth_range",
"radial_range",
}
)
[docs]
def suggest_pyfai_option(name: str, cutoff: float = 0.6) -> Optional[str]:
"""Return the closest allowed pyFAI option to ``name``, or ``None``.
Typo helper: e.g. ``"mpt_rad"`` -> ``"npt_rad"``.
"""
matches = difflib.get_close_matches(
name, sorted(ALLOWED_PYFAI_OPTIONS), n=1, cutoff=cutoff
)
return matches[0] if matches else None
[docs]
class PyFAIConfiguration(BaseModel):
"""A single pyFAI configuration entry of an ID13 template.
``filename`` points to an existing ``.poni``/``.json`` configuration file.
Any extra key (``tag`` aside, e.g. ``nbpt_rad``, ``azimuth_range``, ...) is
kept and forwarded to the pyFAI worker as an integration option.
"""
model_config = ConfigDict(extra="allow")
filename: Path
tag: Optional[str] = None
@model_validator(mode="before")
@classmethod
def _coerce_bare_path(cls, data: Any) -> Any:
# Accept a bare path string/Path instead of a mapping
if isinstance(data, (str, Path)):
return {"filename": str(data)}
return data
@field_validator("filename")
@classmethod
def _check_filename(cls, value: Path) -> Path:
if value.suffix not in PYFAI_CONFIG_SUFFIXES:
raise ValueError(
f"pyFAI configuration file must be one of {PYFAI_CONFIG_SUFFIXES}, "
f"got: {value}"
)
if not value.exists():
raise ValueError(f"pyFAI configuration file does not exist: {value}")
if not value.is_file():
raise ValueError(f"pyFAI configuration path is not a file: {value}")
return value
@model_validator(mode="after")
def _reject_unknown_options(self) -> "PyFAIConfiguration":
unknown = set(self.__pydantic_extra__ or {}) - ALLOWED_PYFAI_OPTIONS
if unknown:
hints = []
for key in sorted(unknown):
suggestion = suggest_pyfai_option(key)
hints.append(
f"{key!r} (did you mean {suggestion!r}?)"
if suggestion
else repr(key)
)
raise ValueError(
"unknown pyFAI option(s): "
+ ", ".join(hints)
+ "\n"
+ f"Allowed parameters: {sorted(ALLOWED_PYFAI_OPTIONS)}"
)
return self
[docs]
class SlurmJobParameters(BaseModel):
"""SLURM job parameters forwarded to the ewoksjob SLURM backend.
The fields below are the ones used across ID13 pipelines; any other valid
SLURM key is still accepted (``extra="allow"``).
"""
model_config = ConfigDict(extra="allow")
name: Optional[str] = None
partition: Optional[str] = None
time_limit: Optional[str] = None
tasks_per_node: Optional[int] = None
cpus_per_task: Optional[int] = None
memory_per_cpu: Optional[str] = None
tres_per_job: Optional[str] = None
constraints: Optional[str] = None
[docs]
class SubmitParameters(BaseModel):
"""Job submission parameters of an ID13 template."""
model_config = ConfigDict(extra="forbid")
submit: bool = True
slurm_job_parameters: SlurmJobParameters = Field(default_factory=SlurmJobParameters)
queue: Optional[str] = None
local_execution: bool = False
[docs]
class IntegrateOptions(BaseModel):
"""Options of the ``integrate`` block.
``prioritize_nonnative_h5items`` and ``eiger_format`` are the documented
switches; any other key is forwarded as-is to the IntegrateBlissScan task
(``extra="allow"``).
"""
model_config = ConfigDict(extra="allow")
prioritize_nonnative_h5items: bool = False
eiger_format: bool = False
class _TemplateBase(BaseModel):
"""Fields shared by every ID13 pyFAI-based YAML template.
The common structure (input BLISS files, pyFAI configurations, scan
selection and submission settings) lives here so the integration and
reprocess templates validate their shared part identically. Unknown
top-level keys are rejected (``extra="forbid"``) to catch typos early.
"""
model_config = ConfigDict(extra="forbid")
bliss_filenames: List[Path]
pyfai_configurations: List[PyFAIConfiguration]
scans: Optional[List[Union[int, str]]] = None
tag: Optional[str] = None
output_root_folder: Optional[Path] = None
save_separate_files: bool = False
integrate: IntegrateOptions = Field(default_factory=IntegrateOptions)
submit_parameters: SubmitParameters = Field(default_factory=SubmitParameters)
@field_validator("bliss_filenames")
@classmethod
def _check_bliss_filenames(cls, value: List[Path]) -> List[Path]:
if not value:
raise ValueError("bliss_filenames must not be empty")
for path in value:
if path.suffix != ".h5":
raise ValueError(f"bliss_filenames must be .h5 files, got: {path}")
if not path.exists():
raise ValueError(
f"bliss file does not exist, maybe a typo in the path?: {path}"
)
if not path.is_file():
raise ValueError(f"bliss path is not a file: {path}")
return value
[docs]
class PyFAIIntegrationModel(_TemplateBase):
"""Full validated content of a pyFAI integration YAML template."""
[docs]
class DiffMapOptions(BaseModel):
"""Options of the ``diffmap`` block (DiffractionMap/kmap reconstruction).
``do_diffmap`` toggles the task; ``normalization_counter`` and
``dark_value`` tune the monitor normalization. Any other key is forwarded
as-is to the diffmap task (``extra="allow"``).
"""
model_config = ConfigDict(extra="allow")
do_diffmap: bool = True
normalization_counter: Optional[str] = None
dark_value: float = 0.0
[docs]
class AverageOptions(BaseModel):
"""Options of the ``average`` block (azimuthal average + reference peaks).
``reference`` is the name of the standard used to overlay Bragg peaks. Any
other key is forwarded as-is to the average task (``extra="allow"``).
"""
model_config = ConfigDict(extra="allow")
do_average: bool = True
reference: Optional[str] = None
[docs]
class EDFOptions(BaseModel):
"""Options of the ``edf`` block (EDF stack export for XRDUA)."""
model_config = ConfigDict(extra="allow")
do_stackedf: bool = True
[docs]
class BackgroundOptions(BaseModel):
"""Options of the ``background`` block (automatic background removal)."""
model_config = ConfigDict(extra="allow")
do_background_removal: bool = False
slurm_job_parameters: SlurmJobParameters = Field(default_factory=SlurmJobParameters)
[docs]
class CNMFOptions(BaseModel):
"""Options of the ``cnmf`` block (constrained NMF phase decomposition).
When ``do_matrix_factorization`` is enabled, ``references_directory`` must
point to an existing folder holding the ``.cif`` reference files.
``radial_limits`` is the ``[min, max]`` 2theta range applied to the neural
network tasks. Any other key is forwarded as-is (``extra="allow"``).
"""
model_config = ConfigDict(extra="allow")
do_matrix_factorization: bool = False
references_directory: Optional[str] = None
radial_limits: Optional[List[float]] = None
nb_components: Optional[int] = None
nb_residues: Optional[int] = None
slurm_job_parameters: SlurmJobParameters = Field(default_factory=SlurmJobParameters)
@field_validator("radial_limits")
@classmethod
def _check_radial_limits(
cls, value: Optional[List[float]]
) -> Optional[List[float]]:
if value is None:
return value
if len(value) != 2:
raise ValueError(
f"radial_limits must contain exactly two values [min, max], got: {value}"
)
low, high = value
if low >= high:
raise ValueError(
f"radial_limits must be increasing, i.e. [min, max] with min < max, "
f"got: {value}"
)
return value
@model_validator(mode="after")
def _check_references_directory(self) -> "CNMFOptions":
if not self.do_matrix_factorization:
return self
directory = self.references_directory
if not directory:
raise ValueError(
"references_directory is required when do_matrix_factorization is True"
)
path = Path(directory)
if not path.exists():
raise ValueError(
f"references_directory does not exist, maybe a typo in the path?: "
f"{directory}"
)
if not path.is_dir():
raise ValueError(f"references_directory is not a directory: {directory}")
return self
[docs]
class InferenceOptions(BaseModel):
"""Options of the ``inference`` block (neural-network phase inference)."""
model_config = ConfigDict(extra="allow")
do_phase_inference: bool = False
slurm_job_parameters: SlurmJobParameters = Field(default_factory=SlurmJobParameters)
[docs]
class ReprocessModel(_TemplateBase):
"""Full validated content of a reprocess YAML template.
On top of the shared integration fields it validates the per-task blocks of
the full reprocessing pipeline: diffraction map, azimuthal average, EDF
export, background removal, matrix factorization and phase inference.
"""
diffmap: DiffMapOptions = Field(default_factory=DiffMapOptions)
average: AverageOptions = Field(default_factory=AverageOptions)
edf: EDFOptions = Field(default_factory=EDFOptions)
background: BackgroundOptions = Field(default_factory=BackgroundOptions)
cnmf: CNMFOptions = Field(default_factory=CNMFOptions)
inference: InferenceOptions = Field(default_factory=InferenceOptions)
def _validate_existing_h5_files(paths: List[Path], label: str) -> List[Path]:
"""Validate a non-empty list of existing ``.h5`` files (shared validator).
A directory entry is expanded in place into the ``.h5`` files it directly
contains (``Path(entry).glob("*.h5")``), so a template may point either at
individual master files or at a folder holding them.
These scripts reprocess already-integrated PROCESSED_DATA files, so any
entry pointing at a RAW_DATA file is rejected: it is almost always a
copy-paste mistake and would silently produce garbage.
"""
if not paths:
raise ValueError(f"{label} must not be empty")
resolved: List[Path] = []
for path in paths:
if path.is_dir():
h5_files = sorted(path.glob("*.h5"))
if not h5_files:
raise ValueError(f"{label} directory contains no .h5 files: {path}")
resolved.extend(h5_files)
continue
if path.suffix != ".h5":
raise ValueError(f"{label} must be .h5 files, got: {path}")
if not path.exists():
raise ValueError(
f"{label} file does not exist, maybe a typo in the path?: {path}"
)
if not path.is_file():
raise ValueError(f"{label} path is not a file: {path}")
resolved.append(path)
for path in resolved:
if "RAW_DATA" in str(path):
raise ValueError(
f"{label} must point to PROCESSED_DATA files, but this is a "
f"RAW_DATA file: {path}"
)
return resolved
class _ScatteringNNOptions(BaseModel):
"""Common options of the scattering neural-network blocks.
``radial_limits`` (the ``[min, max]`` 2theta
window) are shared by the ``background``, ``cnmf`` and ``inference`` blocks
of the standalone scripts. Any other key is forwarded as-is to the task
(``extra="allow"``).
"""
model_config = ConfigDict(extra="allow")
radial_limits: List[float]
@field_validator("radial_limits")
@classmethod
def _check_radial_limits(cls, value: List[float]) -> List[float]:
if len(value) != 2:
raise ValueError(
f"radial_limits must contain exactly two values [min, max], got: {value}"
)
low, high = value
if low >= high:
raise ValueError(
f"radial_limits must be increasing, i.e. [min, max] with min < max, "
f"got: {value}"
)
return value
class _CifReferenceOptions(_ScatteringNNOptions):
"""Scattering options requiring a directory of ``.cif`` reference files.
Both the standalone CNMF and inference scripts always run their neural
network, so ``references_directory`` must point to an existing folder.
"""
references_directory: str
@field_validator("references_directory")
@classmethod
def _check_references_directory(cls, value: str) -> str:
if not value:
raise ValueError("references_directory must not be empty")
path = Path(value)
if not path.exists():
raise ValueError(
f"references_directory does not exist, maybe a typo in the path?: "
f"{value}"
)
if not path.is_dir():
raise ValueError(f"references_directory is not a directory: {value}")
return value
[docs]
class BackgroundScriptOptions(_ScatteringNNOptions):
"""Options of the ``background`` block of the standalone background script."""
[docs]
class CnmfScriptOptions(_CifReferenceOptions):
"""Options of the ``cnmf`` block of the standalone CNMF script."""
nb_components: Optional[int] = None
nb_residues: int = 1
[docs]
class InferenceScriptOptions(_CifReferenceOptions):
"""Options of the ``inference`` block of the standalone inference script."""
class _ProcessedFilesModel(BaseModel):
"""Base for scripts that reprocess already-integrated PROCESSED_DATA files.
Unknown top-level keys are rejected (``extra="forbid"``) to catch typos
early. Each ``processed_filenames`` entry is either an existing ``.h5``
file or a directory, in which case it is expanded into the ``.h5`` files
it directly contains.
"""
model_config = ConfigDict(extra="forbid")
processed_filenames: List[Path]
submit_parameters: SubmitParameters = Field(default_factory=SubmitParameters)
@field_validator("processed_filenames")
@classmethod
def _check_processed_filenames(cls, value: List[Path]) -> List[Path]:
return _validate_existing_h5_files(value, "processed_filenames")
[docs]
class BackgroundModel(_ProcessedFilesModel):
"""Full validated content of a standalone background-removal YAML template."""
nxprocess_path_diffmap: str
nxprocess_path_filtered: str
background: BackgroundScriptOptions
[docs]
class CnmfModel(_ProcessedFilesModel):
"""Full validated content of a standalone CNMF YAML template.
The diffmap and filtered NXprocess paths point to existing groups, while the
CNMF NXprocess is the one to be produced.
"""
nxprocess_path_filtered: str
nxprocess_path_cnmf: str
nxprocess_path_diffmap: Optional[str] = None
cnmf: CnmfScriptOptions
[docs]
class InferenceModel(_ProcessedFilesModel):
"""Full validated content of a standalone phase-inference YAML template."""
nxprocess_path_filtered: str
nxprocess_path_inference: str
inference: InferenceScriptOptions
[docs]
class RecalibModel(BaseModel):
"""Full validated content of a pyFAI recalibration YAML template.
``filename_pyfaiconfig`` is the starting geometry (``.poni``/``.json``) and
``filename_bliss_dataset`` the acquisition of a known calibrant. The per-task
blocks (``pyfai_config``, ``pyfai_calib``, ``save_pyfai_config``) are
forwarded as-is to their tasks. Unknown top-level keys are rejected
(``extra="forbid"``). The ``calibrant`` name is checked at run time against
the pyFAI calibrant registry.
"""
model_config = ConfigDict(extra="forbid")
filename_pyfaiconfig: Path
filename_bliss_dataset: Path
filename_pyfaiconfig_output: Optional[str] = None
scan_nb: int = 1
calibrant: str = "alpha_Al2O3"
detector_name: str = "eiger"
pyfai_config: Dict[str, Any] = Field(default_factory=dict)
pyfai_calib: Dict[str, Any] = Field(default_factory=dict)
save_pyfai_config: Dict[str, Any] = Field(default_factory=dict)
submit_parameters: SubmitParameters = Field(default_factory=SubmitParameters)
@field_validator("filename_pyfaiconfig")
@classmethod
def _check_pyfaiconfig(cls, value: Path) -> Path:
if value.suffix not in PYFAI_CONFIG_SUFFIXES:
raise ValueError(
f"filename_pyfaiconfig must be one of {PYFAI_CONFIG_SUFFIXES}, "
f"got: {value}"
)
if not value.exists():
raise ValueError(
f"filename_pyfaiconfig does not exist, maybe a typo in the path?: "
f"{value}"
)
if not value.is_file():
raise ValueError(f"filename_pyfaiconfig is not a file: {value}")
return value
@field_validator("filename_bliss_dataset")
@classmethod
def _check_bliss_dataset(cls, value: Path) -> Path:
if value.suffix not in BLISS_DATASET_SUFFIXES:
raise ValueError(
f"filename_bliss_dataset must be one of {BLISS_DATASET_SUFFIXES}, "
f"got: {value}"
)
if not value.exists():
raise ValueError(
f"filename_bliss_dataset does not exist, maybe a typo in the path?: "
f"{value}"
)
if not value.is_file():
raise ValueError(f"filename_bliss_dataset is not a file: {value}")
return value