From 26f3a2ce8928d2e4f3f7048b0841bf5dc9dcc882 Mon Sep 17 00:00:00 2001 From: Paul Coe Date: Tue, 11 Aug 2026 14:57:09 +0100 Subject: [PATCH 1/2] Add transmission module for I19 beamline devices * Start with adding JSON validation for system specification JSON dict-like structure expectations * Pydantic BaseModel classes map sub-structures ( sub-dicts ) within the system description structure here adding JSON validation for energy_interval [ the range of x-ray energies (in keV ) over which an absorption curve is valid ] JSON validation for absorption fit curve parameters [ the scaling constant, roll-off vs energy and an optional residuals polynomial ] --- .../beamlines/i19/transmission/__init__.py | 0 .../transmission/spec_from_json/__init__.py | 8 + .../spec_from_json/energy_interval_spec.py | 33 +++ .../fitted_absorption_curve_spec.py | 45 ++++ .../material_absorption_spectrum_spec.py | 57 ++++ .../beamlines/i19/transmission/__init__.py | 0 .../transmission/spec_from_json/__init__.py | 8 + .../test_energy_interval_spec.py | 255 ++++++++++++++++++ .../test_fitted_absorption_curve_spec.py | 182 +++++++++++++ .../test_material_absorption_spectrum_spec.py | 107 ++++++++ .../spec_from_json/utility_constants.py | 54 ++++ 11 files changed, 749 insertions(+) create mode 100644 src/dodal/devices/beamlines/i19/transmission/__init__.py create mode 100644 src/dodal/devices/beamlines/i19/transmission/spec_from_json/__init__.py create mode 100644 src/dodal/devices/beamlines/i19/transmission/spec_from_json/energy_interval_spec.py create mode 100644 src/dodal/devices/beamlines/i19/transmission/spec_from_json/fitted_absorption_curve_spec.py create mode 100644 src/dodal/devices/beamlines/i19/transmission/spec_from_json/material_absorption_spectrum_spec.py create mode 100644 tests/devices/beamlines/i19/transmission/__init__.py create mode 100644 tests/devices/beamlines/i19/transmission/spec_from_json/__init__.py create mode 100644 tests/devices/beamlines/i19/transmission/spec_from_json/test_energy_interval_spec.py create mode 100644 tests/devices/beamlines/i19/transmission/spec_from_json/test_fitted_absorption_curve_spec.py create mode 100644 tests/devices/beamlines/i19/transmission/spec_from_json/test_material_absorption_spectrum_spec.py create mode 100644 tests/devices/beamlines/i19/transmission/spec_from_json/utility_constants.py diff --git a/src/dodal/devices/beamlines/i19/transmission/__init__.py b/src/dodal/devices/beamlines/i19/transmission/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/src/dodal/devices/beamlines/i19/transmission/spec_from_json/__init__.py b/src/dodal/devices/beamlines/i19/transmission/spec_from_json/__init__.py new file mode 100644 index 00000000000..1f9a48725c1 --- /dev/null +++ b/src/dodal/devices/beamlines/i19/transmission/spec_from_json/__init__.py @@ -0,0 +1,8 @@ +"""Utility library that captures system specification state from JSON. + +Beamline scientists have the freedom to edit the JSON files. + +The latest state of JSON at system start up should convert to instances of the Spec(ification) classes, +( which are pydantic (v2) BaseModel classes ). +For these validated BaseModels the absorber instrumentation classes are built, that the transmission system logic will govern. +""" diff --git a/src/dodal/devices/beamlines/i19/transmission/spec_from_json/energy_interval_spec.py b/src/dodal/devices/beamlines/i19/transmission/spec_from_json/energy_interval_spec.py new file mode 100644 index 00000000000..644ce26c9d6 --- /dev/null +++ b/src/dodal/devices/beamlines/i19/transmission/spec_from_json/energy_interval_spec.py @@ -0,0 +1,33 @@ + +from pydantic import BaseModel, ConfigDict, Field, StrictFloat, model_validator + + +class EnergyIntervalSpec(BaseModel): + """JSON built sub-dict of one energy range, used in specifying material absorption spectra. + + Note: + The range of energies is a closed interval, so inclusive of end values. + + Attributes: + units: At present only 'keV' (or 'kiloelectronvolts') are supported. + lower: The lower end of the energy interval (range) for specific absorption fit curve. + upper: The upper end of the energy interval (range) for specific absorption fit curve. + """ + + units: str = Field(..., pattern=r"^(keV|kiloelectronvolts)$") + lower: StrictFloat + upper: StrictFloat + + # Base Model internal setting to make this class immutable + model_config = ConfigDict(frozen=True) + + @model_validator(mode="after") + def validate_attributes(self) -> "EnergyIntervalSpec": + if 0.0 < self.lower < self.upper: + return self + _msg = ( + f"Energy interval lower {self.lower} and upper {self.upper} bounds are in wrong order." + if self.lower > 0.0 + else f"Energy bound use {self.lower} to define lower end of x-ray energy interval, must be greater than zero." + ) + raise ValueError(_msg) diff --git a/src/dodal/devices/beamlines/i19/transmission/spec_from_json/fitted_absorption_curve_spec.py b/src/dodal/devices/beamlines/i19/transmission/spec_from_json/fitted_absorption_curve_spec.py new file mode 100644 index 00000000000..215b9d81fb0 --- /dev/null +++ b/src/dodal/devices/beamlines/i19/transmission/spec_from_json/fitted_absorption_curve_spec.py @@ -0,0 +1,45 @@ +from typing import Final + +from pydantic import BaseModel, ConfigDict, Field, StrictFloat, model_validator + +from dodal.common.general_maths.interval import ClosedInterval, FloatInterval + +PHYSICAL_ROLL_OFF: Final[FloatInterval] = ClosedInterval(lower=-4.0,upper=-2.0) + +class FittedAbsorptionCurveSpec(BaseModel): + """JSON built sub-dict of one absorption curve, used in specifying material absorption spectra. + + Note: + One spectrum absorption curve covers a specific energy range where it is valid. + One or more such curves make up one absorption spectrum. + + Attributes: + photon_absorption: + Material characteristic scaling constant which defines absorption curve. + roll_off: + Power-law exponent for variation with energy, usually something like -2.75 ∓ 0.24. + residuals_polynomial_coeffs: + Optional residuals correction, zeroth order parameter first. + """ + + photon_absorption: StrictFloat + roll_off: StrictFloat + residuals_polynomial_coeffs: list[StrictFloat] = Field(default_factory=list) + + # Base Model internal setting to make this class immutable + model_config = ConfigDict(frozen=True) + + @model_validator(mode="after") + def validate_attributes(self) -> "FittedAbsorptionCurveSpec": + if self.photon_absorption > 0.0 and self.roll_off in PHYSICAL_ROLL_OFF: + return self + if not self.photon_absorption > 0: + raise ValueError( + f"Photon absorption constant {self.photon_absorption} is invalid." + ) + _msg = ( + f"Absorption roll off {self.roll_off} does not seem likely on physics grounds." + if self.roll_off < 0 + else "Absorption roll off is a negative exponent typically close to -3.0 ∓ 1" + ) + raise ValueError(_msg) diff --git a/src/dodal/devices/beamlines/i19/transmission/spec_from_json/material_absorption_spectrum_spec.py b/src/dodal/devices/beamlines/i19/transmission/spec_from_json/material_absorption_spectrum_spec.py new file mode 100644 index 00000000000..bd05452de8f --- /dev/null +++ b/src/dodal/devices/beamlines/i19/transmission/spec_from_json/material_absorption_spectrum_spec.py @@ -0,0 +1,57 @@ +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field, RootModel + +from dodal.devices.beamlines.i19.transmission.spec_from_json.energy_interval_spec import ( + EnergyIntervalSpec, +) +from dodal.devices.beamlines.i19.transmission.spec_from_json.fitted_absorption_curve_spec import ( + FittedAbsorptionCurveSpec, +) + + +class AbsorptionVsEnergyRelation(BaseModel): + """JSON built dict for one element of an absorption spectrum. + + Pairs off the parameters of a fitted absorption curve, + against the energy interval over which the curve is valid. + + Attributes: + valid_energies: EnergyIntervalSpec + fit_parameters: FittedAbsorptionCurveSpec + """ + valid_energies: EnergyIntervalSpec + fit_parameters: FittedAbsorptionCurveSpec + + # Base Model internal setting to make this class immutable + model_config = ConfigDict(frozen=True) + + +class MaterialAbsorptionSpectrumSpec(BaseModel): + """JSON built dict for the absorption spectrum of a specific material. + + Note: + One spectrum absorption curve covers a specific energy range where it is valid. + One or more such curves make up one absorption spectrum. + + Attributes: + absorption_curves: List of absorption curves + """ + + absorption_curves: list[AbsorptionVsEnergyRelation] = Field(..., min_length=1) + + # Base Model internal setting to make this class immutable + model_config = ConfigDict(frozen=True) + + +class MaterialAbsorptionSpectrumJson(RootModel[dict[str, list[AbsorptionVsEnergyRelation]]]): + """Represents a JSON file where the top-level key is the material name. + + Maps material name to its absorption spectrum specification. + """ + + @classmethod + def extract_spectrum(cls, json_blob: dict[str, Any]) -> tuple[str, MaterialAbsorptionSpectrumSpec]: + parsed_blob = cls.model_validate(json_blob) + material, fitted_curves = next( iter(parsed_blob.root.items() )) + return material, MaterialAbsorptionSpectrumSpec(absorption_curves=fitted_curves) diff --git a/tests/devices/beamlines/i19/transmission/__init__.py b/tests/devices/beamlines/i19/transmission/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/tests/devices/beamlines/i19/transmission/spec_from_json/__init__.py b/tests/devices/beamlines/i19/transmission/spec_from_json/__init__.py new file mode 100644 index 00000000000..2d50597759e --- /dev/null +++ b/tests/devices/beamlines/i19/transmission/spec_from_json/__init__.py @@ -0,0 +1,8 @@ +"""Tests of utility library that captures system specification state from JSON. + +Beamline scientists have the freedom to edit the JSON files. + +The latest state of JSON at system start up should convert to instances of the Spec(ification) classes, +( which are pydantic (v2) BaseModel classes ). +For these validated BaseModels the absorber instrumentation classes are built, that the transmission system logic will govern. +""" diff --git a/tests/devices/beamlines/i19/transmission/spec_from_json/test_energy_interval_spec.py b/tests/devices/beamlines/i19/transmission/spec_from_json/test_energy_interval_spec.py new file mode 100644 index 00000000000..c76bf4426fa --- /dev/null +++ b/tests/devices/beamlines/i19/transmission/spec_from_json/test_energy_interval_spec.py @@ -0,0 +1,255 @@ +import math +from typing import Any + +import pytest +from pydantic import ValidationError + +from dodal.devices.beamlines.i19.transmission.spec_from_json.energy_interval_spec import ( + EnergyIntervalSpec, +) + +from .utility_constants import ( + INVALID_ENERGY_UNITS, + NON_NUMERICALS, + UNSUPPORTED_ENERGY_UNITS, +) + +# Happy path tests + +@pytest.mark.parametrize( + "upper_bound", + ( + 2.2, + 3, + 17.8, + 24.001, + 25.7, + 27, + 28.9, + ), +) +@pytest.mark.parametrize( + "lower_bound", + ( + 0.2, + 0.5, + 1, + 1.6, + 2, + ), +) +@pytest.mark.parametrize( + "energy_unit", + ( + "keV", + "kiloelectronvolts", + ), +) +def test_that_energy_interval_spec_can_be_built_from_dict_with_valid_energy_units( + energy_unit: str, lower_bound: float, upper_bound: float +) -> None: + d: dict[str, Any] = { + "units": energy_unit, + "lower": lower_bound, + "upper": upper_bound, + } + assert EnergyIntervalSpec(**d) is not None + + +# Inauspicious path tests + + +@pytest.mark.parametrize( + "upper_bound", + ( + 2.2, + 27, + 28.9, + ), +) +@pytest.mark.parametrize( + "lower_bound", + ( + 0.2, + 0.5, + 1, + ), +) +@pytest.mark.parametrize( + "key", + ( + "unitz", + "Units", + "Unit", + "unit", + "u", + "lower", + "upper", + ), +) +def test_that_energy_interval_json_validation_fails_with_typo_in_units_bound_key( + key: str, lower_bound: float, upper_bound: float +) -> None: + d: dict[str, Any] = {key: "keV", "lower": lower_bound, "upper": upper_bound} + with pytest.raises(ValidationError): + EnergyIntervalSpec(**d) + + +@pytest.mark.parametrize( + "upper_bound", + ( + 2.2, + 27, + 28.9, + ), +) +@pytest.mark.parametrize( + "lower_bound", + ( + 0.2, + 0.5, + 1, + ), +) +@pytest.mark.parametrize("key", ("loewr", "Lower", "LR", "l0wer", "upper", "units")) +def test_that_energy_interval_json_validation_fails_with_typo_in_lower_bound_key( + key: str, lower_bound: float, upper_bound: float +) -> None: + d: dict[str, Any] = {"units": "keV", key: lower_bound, "upper": upper_bound} + with pytest.raises(ValidationError): + EnergyIntervalSpec(**d) + + +@pytest.mark.parametrize( + "upper_bound", + ( + 2.2, + 27, + 28.9, + ), +) +@pytest.mark.parametrize( + "lower_bound", + ( + 0.2, + 0.5, + 1, + ), +) +@pytest.mark.parametrize( + "key", ("uppre", "Upper", "UR", "uPper", "UPPER", "lower", "units") +) +def test_that_energy_interval_json_validation_fails_with_typo_in_upper_bound_key( + key: str, lower_bound: float, upper_bound: float +) -> None: + d: dict[str, Any] = {"units": "keV", "lower": lower_bound, key: upper_bound} + with pytest.raises(ValidationError): + EnergyIntervalSpec(**d) + + +@pytest.mark.parametrize( + "unsupported_units", + UNSUPPORTED_ENERGY_UNITS +) +def test_that_energy_interval_json_validation_rejects_unsupported_energy_units( + unsupported_units, +) -> None: + d: dict[str, Any] = {"units": unsupported_units, "lower": 2.1, "upper": 11.5} + with pytest.raises(ValidationError): + EnergyIntervalSpec(**d) + + +@pytest.mark.parametrize( + "invalid_energy_units", + INVALID_ENERGY_UNITS, +) +def test_that_energy_interval_json_validation_rejects_invalid_energy_units( + invalid_energy_units, +) -> None: + d: dict[str, Any] = {"units": invalid_energy_units, "lower": 2.1, "upper": 11.5} + with pytest.raises(ValidationError): + EnergyIntervalSpec(**d) + + +@pytest.mark.parametrize( + "upper_bound", + ( + 3.9, + 3.75, + 3.4, + 3, + 2.9, + 2.405, + math.e, + 2.17, + 1.7, + 0.89, + ), +) +@pytest.mark.parametrize( + "high_lower_bound", + ( + 4, + 5.0, + 12, + 6.7, + 44.1, + 8009.23, + 23, + ), +) +def test_that_energy_interval_json_validation_rejects_bounds_when_lower_bound_above_upper_bound( + high_lower_bound: float, upper_bound: float +) -> None: + d: dict[str, Any] = { + "units": "keV", + "lower": high_lower_bound, + "upper": upper_bound, + } + with pytest.raises(ValidationError): + EnergyIntervalSpec(**d) + + +@pytest.mark.parametrize( + "invalid_lower_bound", + ( + -math.pi, + -12, + -100.89, + -0.03, + -1.4, + -2, + ), +) +def test_that_energy_interval_json_validation_rejects_negative_lower_bound( + invalid_lower_bound, +) -> None: + d: dict[str, Any] = {"units": "keV", "lower": invalid_lower_bound, "upper": 25} + with pytest.raises(ValidationError): + EnergyIntervalSpec(**d) + + +@pytest.mark.parametrize("non_numerical_lower_bound", NON_NUMERICALS) +def test_that_energy_interval_json_validation_rejects_non_numerical_lower_bound( + non_numerical_lower_bound, +) -> None: + d: dict[str, Any] = { + "units": "keV", + "lower": non_numerical_lower_bound, + "upper": 28.54, + } + with pytest.raises(ValidationError): + EnergyIntervalSpec(**d) + + +@pytest.mark.parametrize("non_numerical_upper_bound", NON_NUMERICALS) +def test_that_energy_interval_json_validation_rejects_non_numerical_upper_bound( + non_numerical_upper_bound, +) -> None: + d: dict[str, Any] = { + "units": "keV", + "lower": 15.25, + "upper": non_numerical_upper_bound, + } + with pytest.raises(ValidationError): + EnergyIntervalSpec(**d) diff --git a/tests/devices/beamlines/i19/transmission/spec_from_json/test_fitted_absorption_curve_spec.py b/tests/devices/beamlines/i19/transmission/spec_from_json/test_fitted_absorption_curve_spec.py new file mode 100644 index 00000000000..75c5f2cccf5 --- /dev/null +++ b/tests/devices/beamlines/i19/transmission/spec_from_json/test_fitted_absorption_curve_spec.py @@ -0,0 +1,182 @@ + +from math import pi +from typing import Any, Final + +import pytest +from pydantic import ValidationError + +from dodal.devices.beamlines.i19.transmission.spec_from_json.fitted_absorption_curve_spec import ( + FittedAbsorptionCurveSpec, +) + +from .utility_constants import NON_NUMERICALS + +# Happy path tests + +WITH_OR_WITHOUT_RESIDUALS: Final[tuple[list[float], ...]] = ([], [40.9271, -21.6924, 7.5198, -12.51, 8.87e-2, -5.39e-4, 4.796e-5]) + +@pytest.mark.parametrize( + "residuals", + WITH_OR_WITHOUT_RESIDUALS +) +@pytest.mark.parametrize("roll_off", (-2.983, -2.83, -2.72,)) +@pytest.mark.parametrize( + "scaling_constant", ( + 3123.4, + 7812.9, + 1.305e6, + ), +) +def test_that_fitted_absorption_curve_spec_builds_from_valid_values( + scaling_constant: float, roll_off: float, residuals: list[float] +) -> None: + d: dict[str, Any] = { + "photon_absorption": scaling_constant, + "roll_off": roll_off, + "residuals_polynomial_coeffs": residuals, + } + assert FittedAbsorptionCurveSpec(**d) is not None + + + +@pytest.mark.parametrize("roll_off", (-2.983, -2.83, -2.72,)) +@pytest.mark.parametrize( + "scaling_constant", ( + 3123.4, + 7812.9, + 1.305e6, + ), +) +def test_that_fitted_absorption_curve_spec_builds_when_residuals_are_in_tuple( + scaling_constant: float, roll_off: float +) -> None: + _residuals_tuple: tuple[float,...] = ( -1.5, 0.03, 16.8, -272.345) + d: dict[str, Any] = { + "photon_absorption": scaling_constant, + "roll_off": roll_off, + "residuals_polynomial_coeffs": _residuals_tuple, + } + assert FittedAbsorptionCurveSpec(**d) is not None + + +# Inauspicious path tests + +@pytest.mark.parametrize( + "residuals", + WITH_OR_WITHOUT_RESIDUALS +) +@pytest.mark.parametrize( + "non_positive_scaling_constant", + ( 0, -2.8, -2903.4) +) +def test_that_non_positive_scaling_constants_rejected_by_fitted_absorption_curve_spec( + non_positive_scaling_constant: float, + residuals: list[float] +) -> None: + d: dict[str, Any] = { + "photon_absorption": non_positive_scaling_constant, + "roll_off": -2.78, + "residuals_polynomial_coeffs": residuals, + } + with pytest.raises(ValidationError): + FittedAbsorptionCurveSpec(**d) + + +@pytest.mark.parametrize( + "residuals", + WITH_OR_WITHOUT_RESIDUALS, +) +@pytest.mark.parametrize( + "non_numerical_scaling_constant", + NON_NUMERICALS, +) +def test_that_non_numerical_scaling_constants_rejected_by_fitted_absorption_curve_spec( + non_numerical_scaling_constant: Any, + residuals: list[float] +) -> None: + d: dict[str, Any] = { + "photon_absorption": non_numerical_scaling_constant, + "roll_off": -2.78, + "residuals_polynomial_coeffs": residuals, + } + with pytest.raises(ValidationError): + FittedAbsorptionCurveSpec(**d) + +UNPHYSICAL_ROLL_OFF_EXPONENTS: Final[list[float]] = [ + 1.2, 2.96, -1.5, -4.5, -296, -1, pi, 1, 4, 44.8, 200004, -1289, +] + +@pytest.mark.parametrize( + "residuals", + WITH_OR_WITHOUT_RESIDUALS, +) +@pytest.mark.parametrize( + "unphysical_exponent", + UNPHYSICAL_ROLL_OFF_EXPONENTS, +) +def test_that_unphysical_roll_off_exponent_rejected_by_fitted_absorption_curve_spec( + unphysical_exponent: float, + residuals: list[float] +) -> None: + d: dict[str, Any] = { + "photon_absorption": 53408.9, + "roll_off": unphysical_exponent, + "residuals_polynomial_coeffs": residuals, + } + with pytest.raises(ValidationError): + FittedAbsorptionCurveSpec(**d) + + +@pytest.mark.parametrize( + "residuals", + WITH_OR_WITHOUT_RESIDUALS, +) +@pytest.mark.parametrize( + "non_numerical_exponent", + NON_NUMERICALS, +) +def test_that_non_numerical_roll_off_exponent_rejected_by_fitted_absorption_curve_spec( + non_numerical_exponent: Any, + residuals: list[float] +) -> None: + d: dict[str, Any] = { + "photon_absorption": 53408.9, + "roll_off": non_numerical_exponent, + "residuals_polynomial_coeffs": residuals, + } + with pytest.raises(ValidationError): + FittedAbsorptionCurveSpec(**d) + + +@pytest.mark.parametrize( + "non_numerical", + NON_NUMERICALS, +) +def test_that_non_numerical_residual_rejected_by_fitted_absorption_curve_spec( + non_numerical: Any, +) -> None: + _corrupted_residuals = [ 1.9, -9.1, non_numerical, -428.01312] + d: dict[str, Any] = { + "photon_absorption": 63091.2, + "roll_off": -2.78, + "residuals_polynomial_coeffs": _corrupted_residuals, + } + with pytest.raises(ValidationError): + FittedAbsorptionCurveSpec(**d) + + +@pytest.mark.parametrize( + "stringy_residuals", + "[1.9, 27.8, -55.81, -206.11]", +) +def test_that_string_residuals_rejected_by_fitted_absorption_curve_spec( + stringy_residuals: str +) -> None: + d: dict[str, Any] = { + "photon_absorption": 63091.2, + "roll_off": -2.78, + "residuals_polynomial_coeffs": stringy_residuals, + } + with pytest.raises(ValidationError): + FittedAbsorptionCurveSpec(**d) + diff --git a/tests/devices/beamlines/i19/transmission/spec_from_json/test_material_absorption_spectrum_spec.py b/tests/devices/beamlines/i19/transmission/spec_from_json/test_material_absorption_spectrum_spec.py new file mode 100644 index 00000000000..45fc5cf8215 --- /dev/null +++ b/tests/devices/beamlines/i19/transmission/spec_from_json/test_material_absorption_spectrum_spec.py @@ -0,0 +1,107 @@ + +from typing import Any, Final + +import pytest + +from dodal.devices.beamlines.i19.transmission.spec_from_json.material_absorption_spectrum_spec import ( + MaterialAbsorptionSpectrumJson, + MaterialAbsorptionSpectrumSpec, +) + +ALU_SPECTRUM: Final[dict[str, Any]] = { + "aluminium": [ + { + "valid_energies": { + "units": "keV", + "lower" : 5.0, + "upper" : 30.0 + }, + "fit_parameters": { + "photon_absorption": 64873.0, + "roll_off":-2.96, + "residuals_polynomial_coeffs": [] + } + } + ] +} + +GOLD_SPECTRUM: Final[dict[str, Any]] = { + "gold": [ + { + "valid_energies": { + "units": "keV", + "lower": 5.0, + "upper":11.5 + }, + "fit_parameters": { + "photon_absorption": 7.87191216e5, + "roll_off":-2.52962901, + "residuals_polynomial_coeffs": [] + } + }, + { + "valid_energies": { + "units": "keV", + "lower": 15.0, + "upper":30.0 + }, + "fit_parameters": { + "photon_absorption": 3.63187927e6, + "roll_off":-2.59815489, + "residuals_polynomial_coeffs": [] + } + } + ], +} + +IRON_SPECTRUM: Final[dict[str, Any]] = { + "iron": [ + { + "valid_energies": { + "units": "keV", + "lower": 8.0, + "upper":30.0 + }, + "fit_parameters": { + "photon_absorption": 1.034347e6, + "roll_off":-2.8655, + "residuals_polynomial_coeffs": [] + } + } + ] +} + +RESIN_SPECTRUM: Final[dict[str, Any]] = { + "resin": [ + { + "valid_energies": { + "units": "keV", + "lower": 5.0, + "upper":22.0 + }, + "fit_parameters": { + "photon_absorption": 3814.1, + "roll_off":-2.81706, + "residuals_polynomial_coeffs": [ + 50.9211, + -23.6148, + 4.2138, + -0.3814, + 1.867e-2, + -4.709e-4, + 4.796e-6 + ] + } + } + ] +} + +@pytest.mark.parametrize( + "material_spectrum_blob", + [ALU_SPECTRUM, GOLD_SPECTRUM, IRON_SPECTRUM, RESIN_SPECTRUM] +) +def test_that_material_absorption_spectrum_can_be_built_from_dict_with_valid_content(material_spectrum_blob: dict[str, Any]): + material_name, spectrum = MaterialAbsorptionSpectrumJson.extract_spectrum(material_spectrum_blob) + + assert material_name in ["aluminium", "gold", "iron", "resin"] + assert isinstance(spectrum, MaterialAbsorptionSpectrumSpec) diff --git a/tests/devices/beamlines/i19/transmission/spec_from_json/utility_constants.py b/tests/devices/beamlines/i19/transmission/spec_from_json/utility_constants.py new file mode 100644 index 00000000000..c101063a078 --- /dev/null +++ b/tests/devices/beamlines/i19/transmission/spec_from_json/utility_constants.py @@ -0,0 +1,54 @@ + +"""Handy test constants, that this module will use for parametrized testing.""" + +from math import ( + cos as cosine_function, +) +from math import ( + pi, +) +from math import ( + sin as sine_function, +) +from typing import Any, Final + +INVALID_ENERGY_UNITS: Final[tuple[Any, ...]] = ( + "", + "*^&", + 16, + cosine_function, + KeyError(), + object(), + "-9", + -9.0, + pi, + True, + False, +) + +NON_NUMERICALS: Final[tuple[Any, ...]] = ( + sine_function, + object(), + ValueError(), + "312", + "*$", + "", + " ", + "Hello World", + "7.6", + True, + False, +) + +UNSUPPORTED_ENERGY_UNITS: Final[tuple[str, ...]] = ( + "meV", + "KILOelectronvolts", + "kev", + "k_eV", + "MeV", + "PeV", + "J", + "Joules", + "s", + "GHz", + ) From 9da488de8ff3852fd98d0a6f069d151932f5c5e4 Mon Sep 17 00:00:00 2001 From: Paul Coe Date: Thu, 13 Aug 2026 15:54:51 +0100 Subject: [PATCH 2/2] Add transmission system configuration validating reader classes * Use pydantic BaseModel classes to capture blobs from Transmission System specification config JSON files for I19 beamlines EH-1 / EH-2 --- .../__init__.py | 0 .../energy_interval_spec.py | 3 +- .../fitted_absorption_curve_spec.py | 7 +- .../spec_from_config/foil_spec.py | 48 ++++ .../spec_from_config/lateral_motor_spec.py | 92 +++++++ .../material_absorption_spectrum_spec.py | 127 +++++++++ .../spec_from_config/name_validation.py | 109 ++++++++ .../system_aspect_base_parser.py | 85 ++++++ .../spec_from_config/system_configuration.py | 34 +++ .../transmission_system_spec.py | 40 +++ .../spec_from_config/wedges_spec.py | 116 +++++++++ .../spec_from_config/wheels_spec.py | 71 +++++ .../material_absorption_spectrum_spec.py | 57 ---- .../__init__.py | 0 .../spec_from_config/fake_json.py | 244 ++++++++++++++++++ .../test_energy_interval_spec.py | 8 +- .../test_fitted_absorption_curve_spec.py | 80 +++--- .../spec_from_config/test_foil_spec.py | 54 ++++ .../test_lateral_motors_spec.py | 204 +++++++++++++++ .../test_material_absorption_spectrum_spec.py | 228 ++++++++++++++++ .../test_system_configuration.py | 101 ++++++++ .../spec_from_config/test_wedges_spec.py | 234 +++++++++++++++++ .../spec_from_config/test_wheels_spec.py | 169 ++++++++++++ .../utility_constants.py | 23 +- .../test_material_absorption_spectrum_spec.py | 107 -------- 25 files changed, 2024 insertions(+), 217 deletions(-) rename src/dodal/devices/beamlines/i19/transmission/{spec_from_json => spec_from_config}/__init__.py (100%) rename src/dodal/devices/beamlines/i19/transmission/{spec_from_json => spec_from_config}/energy_interval_spec.py (95%) rename src/dodal/devices/beamlines/i19/transmission/{spec_from_json => spec_from_config}/fitted_absorption_curve_spec.py (94%) create mode 100644 src/dodal/devices/beamlines/i19/transmission/spec_from_config/foil_spec.py create mode 100644 src/dodal/devices/beamlines/i19/transmission/spec_from_config/lateral_motor_spec.py create mode 100644 src/dodal/devices/beamlines/i19/transmission/spec_from_config/material_absorption_spectrum_spec.py create mode 100644 src/dodal/devices/beamlines/i19/transmission/spec_from_config/name_validation.py create mode 100644 src/dodal/devices/beamlines/i19/transmission/spec_from_config/system_aspect_base_parser.py create mode 100644 src/dodal/devices/beamlines/i19/transmission/spec_from_config/system_configuration.py create mode 100644 src/dodal/devices/beamlines/i19/transmission/spec_from_config/transmission_system_spec.py create mode 100644 src/dodal/devices/beamlines/i19/transmission/spec_from_config/wedges_spec.py create mode 100644 src/dodal/devices/beamlines/i19/transmission/spec_from_config/wheels_spec.py delete mode 100644 src/dodal/devices/beamlines/i19/transmission/spec_from_json/material_absorption_spectrum_spec.py rename tests/devices/beamlines/i19/transmission/{spec_from_json => spec_from_config}/__init__.py (100%) create mode 100644 tests/devices/beamlines/i19/transmission/spec_from_config/fake_json.py rename tests/devices/beamlines/i19/transmission/{spec_from_json => spec_from_config}/test_energy_interval_spec.py (96%) rename tests/devices/beamlines/i19/transmission/{spec_from_json => spec_from_config}/test_fitted_absorption_curve_spec.py (77%) create mode 100644 tests/devices/beamlines/i19/transmission/spec_from_config/test_foil_spec.py create mode 100644 tests/devices/beamlines/i19/transmission/spec_from_config/test_lateral_motors_spec.py create mode 100644 tests/devices/beamlines/i19/transmission/spec_from_config/test_material_absorption_spectrum_spec.py create mode 100644 tests/devices/beamlines/i19/transmission/spec_from_config/test_system_configuration.py create mode 100644 tests/devices/beamlines/i19/transmission/spec_from_config/test_wedges_spec.py create mode 100644 tests/devices/beamlines/i19/transmission/spec_from_config/test_wheels_spec.py rename tests/devices/beamlines/i19/transmission/{spec_from_json => spec_from_config}/utility_constants.py (79%) delete mode 100644 tests/devices/beamlines/i19/transmission/spec_from_json/test_material_absorption_spectrum_spec.py diff --git a/src/dodal/devices/beamlines/i19/transmission/spec_from_json/__init__.py b/src/dodal/devices/beamlines/i19/transmission/spec_from_config/__init__.py similarity index 100% rename from src/dodal/devices/beamlines/i19/transmission/spec_from_json/__init__.py rename to src/dodal/devices/beamlines/i19/transmission/spec_from_config/__init__.py diff --git a/src/dodal/devices/beamlines/i19/transmission/spec_from_json/energy_interval_spec.py b/src/dodal/devices/beamlines/i19/transmission/spec_from_config/energy_interval_spec.py similarity index 95% rename from src/dodal/devices/beamlines/i19/transmission/spec_from_json/energy_interval_spec.py rename to src/dodal/devices/beamlines/i19/transmission/spec_from_config/energy_interval_spec.py index 644ce26c9d6..32fe94fb514 100644 --- a/src/dodal/devices/beamlines/i19/transmission/spec_from_json/energy_interval_spec.py +++ b/src/dodal/devices/beamlines/i19/transmission/spec_from_config/energy_interval_spec.py @@ -1,4 +1,3 @@ - from pydantic import BaseModel, ConfigDict, Field, StrictFloat, model_validator @@ -19,7 +18,7 @@ class EnergyIntervalSpec(BaseModel): upper: StrictFloat # Base Model internal setting to make this class immutable - model_config = ConfigDict(frozen=True) + model_config = ConfigDict(frozen=True, extra="forbid") @model_validator(mode="after") def validate_attributes(self) -> "EnergyIntervalSpec": diff --git a/src/dodal/devices/beamlines/i19/transmission/spec_from_json/fitted_absorption_curve_spec.py b/src/dodal/devices/beamlines/i19/transmission/spec_from_config/fitted_absorption_curve_spec.py similarity index 94% rename from src/dodal/devices/beamlines/i19/transmission/spec_from_json/fitted_absorption_curve_spec.py rename to src/dodal/devices/beamlines/i19/transmission/spec_from_config/fitted_absorption_curve_spec.py index 215b9d81fb0..1b2a8c830ae 100644 --- a/src/dodal/devices/beamlines/i19/transmission/spec_from_json/fitted_absorption_curve_spec.py +++ b/src/dodal/devices/beamlines/i19/transmission/spec_from_config/fitted_absorption_curve_spec.py @@ -4,7 +4,8 @@ from dodal.common.general_maths.interval import ClosedInterval, FloatInterval -PHYSICAL_ROLL_OFF: Final[FloatInterval] = ClosedInterval(lower=-4.0,upper=-2.0) +PHYSICAL_ROLL_OFF: Final[FloatInterval] = ClosedInterval(lower=-4.0, upper=-2.0) + class FittedAbsorptionCurveSpec(BaseModel): """JSON built sub-dict of one absorption curve, used in specifying material absorption spectra. @@ -27,7 +28,7 @@ class FittedAbsorptionCurveSpec(BaseModel): residuals_polynomial_coeffs: list[StrictFloat] = Field(default_factory=list) # Base Model internal setting to make this class immutable - model_config = ConfigDict(frozen=True) + model_config = ConfigDict(frozen=True, extra="forbid") @model_validator(mode="after") def validate_attributes(self) -> "FittedAbsorptionCurveSpec": @@ -38,7 +39,7 @@ def validate_attributes(self) -> "FittedAbsorptionCurveSpec": f"Photon absorption constant {self.photon_absorption} is invalid." ) _msg = ( - f"Absorption roll off {self.roll_off} does not seem likely on physics grounds." + f"Absorption roll off {self.roll_off} does not seem likely on physics grounds." if self.roll_off < 0 else "Absorption roll off is a negative exponent typically close to -3.0 ∓ 1" ) diff --git a/src/dodal/devices/beamlines/i19/transmission/spec_from_config/foil_spec.py b/src/dodal/devices/beamlines/i19/transmission/spec_from_config/foil_spec.py new file mode 100644 index 00000000000..71f96eb2932 --- /dev/null +++ b/src/dodal/devices/beamlines/i19/transmission/spec_from_config/foil_spec.py @@ -0,0 +1,48 @@ +from pydantic import ( + BaseModel, + ConfigDict, + StrictFloat, +) + + +class FoilThicknessSpec(BaseModel): + """The specification for a flat foil thickness (as specified by configuration, typically **JSON**). + + Attributes: + units: The length unit used to specify the foil thickness. + value: The foil thickness in the specified unit. + """ + + units: str + value: StrictFloat + + # Base Model internal setting to make this class immutable + model_config = ConfigDict(frozen=True, extra="forbid") + + +class AbsorberSpec(BaseModel): + """The specification for a FixedDepth absorber (as specified by configuration, typically **JSON**). + + Attributes: + material: The name of the absorber material, as present the in the materials section of the configuration. + thickness: ThicknessProvider geometry of the flat foil absorber. + """ + + material: str + thickness: FoilThicknessSpec + + # Base Model internal setting to make this class immutable + model_config = ConfigDict(frozen=True, extra="forbid") + + +class FoilSpec(BaseModel): + """The specification for a filter slot in a wheel. + + Attributes: + absorber: Absorbing filter configuration for a given slot. + """ + + absorber: AbsorberSpec + + # Base Model internal setting to make this class immutable + model_config = ConfigDict(frozen=True, extra="forbid") diff --git a/src/dodal/devices/beamlines/i19/transmission/spec_from_config/lateral_motor_spec.py b/src/dodal/devices/beamlines/i19/transmission/spec_from_config/lateral_motor_spec.py new file mode 100644 index 00000000000..1bfea4eaa57 --- /dev/null +++ b/src/dodal/devices/beamlines/i19/transmission/spec_from_config/lateral_motor_spec.py @@ -0,0 +1,92 @@ +from pydantic import ( + BaseModel, + ConfigDict, + Field, + StrictFloat, + model_validator, +) + +from dodal.common.general_maths.interval import ClosedInterval +from dodal.devices.beamlines.i19.transmission.spec_from_config.name_validation import ( + AxisNameValidation, +) +from dodal.devices.beamlines.i19.transmission.spec_from_config.system_aspect_base_parser import ( + SystemAspectBaseParser, +) +from dodal.devices.beamlines.i19.transmission.spec_from_config.system_configuration import ( + SystemConfiguration, +) + + +class LateralMotorSpec(BaseModel): + """The positions scale for an axial wedge motor (as specified by configuration, typically **JSON**). + + Notes: + - The (potentially counterintuitive) class' attribute names match beamline scientists' "domain jargon". + + - Each lateral motor drives one absorber wedge sideways across the x-ray beam: + - expected sideways motions are pure horizontal or pure vertical + - but that's a detail, azimuthal orientation around the beam should not matter. + - The position of the scale zero relative to the wedge location is hereby nailed down. + - The wedge motor scale has + - an **out** position + - and then a range of "active" positions, ( where the mathematical assumption of a linear taper is reasonable ). + + - Although the units here will default to **mm** when unspecified (i.e. when omitted), + the units setting is still likely to appear in the JSON, as a handy reminder to human readers of the configuration file. + + Attributes: + units: At present only "mm" are supported. + out: Motor position consistent with having fully retracted the wedge from the x-ray beam. + threshold: Motor position for minimal absorption in the permitted active absorbing position range. + max: Motor position for minimal absorption in the permitted active range. + tolerance: Motor position accepted margin for readout error. + """ + + units: str = Field(default="mm") + out: StrictFloat + threshold: StrictFloat + max: StrictFloat + tolerance: StrictFloat = Field(default=5.0e-3) + + # Base Model internal setting to make this class immutable + model_config = ConfigDict(frozen=True, extra="forbid") + + @model_validator(mode="after") + def validate_attributes(self) -> "LateralMotorSpec": + # counter-intuitive naming, but depending on + # motor scale orientation relative to wedge orientation + # max (thickness) position could greater or smaller than threshold + _lower = min(self.out, self.max) + _upper = max(self.out, self.max) + _interval = ClosedInterval(lower=_lower, upper=_upper) + if self.threshold in _interval: + return self + msg: str = r"Inconsistent wedge geometry: Threshold not between max and out." + raise ValueError(msg) + + +class LateralMotorsConfig(SystemAspectBaseParser[LateralMotorSpec]): + """Maps from highest level configuration (JSON) dict to extract lateral motor specifications. + + Uses base class method **extract_specifications** and some pythonic type handling magic. + + See Also: + Base of this parser class, namely **SystemAspectBaseParser** + """ + + def validate_key_name(self, *, key_name: str) -> None: + AxisNameValidation.validate_axis_name(axis_name=key_name) + + @classmethod + def extract_motors_specifications( + cls, + *, + system_configuration: SystemConfiguration, + motor_identifier: str, + ) -> LateralMotorSpec: + """Extracts wheel specification from configuration of the transmission system.""" + _motors: dict[str, LateralMotorSpec] = cls.get_aspect_specifications( + system_configuration=system_configuration + ) + return _motors[motor_identifier] diff --git a/src/dodal/devices/beamlines/i19/transmission/spec_from_config/material_absorption_spectrum_spec.py b/src/dodal/devices/beamlines/i19/transmission/spec_from_config/material_absorption_spectrum_spec.py new file mode 100644 index 00000000000..525cd7e9486 --- /dev/null +++ b/src/dodal/devices/beamlines/i19/transmission/spec_from_config/material_absorption_spectrum_spec.py @@ -0,0 +1,127 @@ +from typing import Any + +from pydantic import ( + BaseModel, + ConfigDict, + Field, + field_validator, + model_validator, +) + +from dodal.devices.beamlines.i19.transmission.spec_from_config.energy_interval_spec import ( + EnergyIntervalSpec, +) +from dodal.devices.beamlines.i19.transmission.spec_from_config.fitted_absorption_curve_spec import ( + FittedAbsorptionCurveSpec, +) +from dodal.devices.beamlines.i19.transmission.spec_from_config.name_validation import ( + MaterialNameValidation, +) +from dodal.devices.beamlines.i19.transmission.spec_from_config.system_aspect_base_parser import ( + SystemAspectBaseParser, +) +from dodal.devices.beamlines.i19.transmission.spec_from_config.system_configuration import ( + SystemConfiguration, +) + + +class AbsorptionVsEnergyRelation(BaseModel): + """Configuration dict for one piece of an absorption spectrum, one energy range specific fit curve. + + Note: + This relation class is a pairing. The internal pair consists of : + - The x-ray energy interval over which this piece of the spectrum is valid (a.k.a. domain). + - The parameters of a fitted absorption curve. + + Attributes: + valid_energies: Specifies x-ray energy *domain* valid for the fitted curve. + fit_parameters: Specifies parameters needed to calculate absorption values on the fitted curve. + + *See also, these closely related classes:* + **FittedAbsorptionCurveSpec**: Fitted curve specification + **EnergyIntervalSpec**: Energy range (interval) specification + **ClosedInterval**: General maths class underpinning interval definition + """ + + valid_energies: EnergyIntervalSpec + fit_parameters: FittedAbsorptionCurveSpec + + # Base Model internal setting to make this class immutable + model_config = ConfigDict(frozen=True, extra="forbid") + + +class MaterialAbsorptionSpectrumSpec(BaseModel): + """Configuration dict for the absorption spectrum of a specific material. + + Note: + One spectrum absorption curve covers a specific energy range where it is valid. + One or more such curves make up one absorption spectrum. + + Attributes: + absorption_curves: List of absorption curves + """ + + absorption_curves: list[AbsorptionVsEnergyRelation] = Field(..., min_length=1) + + # Base Model internal setting to make this class immutable and valid + model_config = ConfigDict(frozen=True, extra="forbid") + + @model_validator(mode="before") + @classmethod + def _coerce_raw_list_to_absorption_curves_dict(cls, data: Any) -> Any: + """Wrap spectrum pieces listing internally as a dict.""" + if isinstance(data, list): + return {"absorption_curves": data} + if isinstance(data, dict): + return data + raise ValueError( + "Absorption spectrum data should be a list or dict of fitted curves." + ) + + +class MaterialAbsorptionSpectralConfig( + SystemAspectBaseParser[MaterialAbsorptionSpectrumSpec] +): + """Configuration dict for the absorption spectra of all specified absorber materials. + + Maps each material name to its absorption spectrum specification, as extracted from configuration (JSON). + + Note: + Base class does most of the work - except for material name validation. + """ + + def validate_key_name(self, *, key_name: str) -> None: + MaterialNameValidation.validate_material_name(material_name=key_name) + + @field_validator("root") + @classmethod + def _ensure_at_least_one_absorber_material_has_been_specified( + cls, + all_absorber_materials_specifications: dict[ + str, MaterialAbsorptionSpectrumSpec + ], + ) -> dict[str, MaterialAbsorptionSpectrumSpec]: + """Invalidates configuration if that features zero absorber materials. + + Raises: + ValueError - if no absorbers are present. + """ + _specified_absorber_materials = all_absorber_materials_specifications.keys() + if len(_specified_absorber_materials) < 1: + raise ValueError( + "Empty absorber materials configuration! This is not valid input." + ) + return all_absorber_materials_specifications + + @classmethod + def extract_absorber_material_specifications( + cls, + *, + system_configuration: SystemConfiguration, + material_name: str, + ) -> MaterialAbsorptionSpectrumSpec: + """Extracts the absorption spectrum specification for a specific material.""" + _materials_spectra: dict[str, MaterialAbsorptionSpectrumSpec] = ( + cls.get_aspect_specifications(system_configuration=system_configuration) + ) + return _materials_spectra[material_name] diff --git a/src/dodal/devices/beamlines/i19/transmission/spec_from_config/name_validation.py b/src/dodal/devices/beamlines/i19/transmission/spec_from_config/name_validation.py new file mode 100644 index 00000000000..e99d555029e --- /dev/null +++ b/src/dodal/devices/beamlines/i19/transmission/spec_from_config/name_validation.py @@ -0,0 +1,109 @@ +import re +from functools import cached_property +from re import Pattern +from typing import Final + + +class NameValidator: + def __init__( + self, *, regexp_pattern: str, thing_being_named: str, error_message: str + ) -> None: + self.regexp_pattern = regexp_pattern + self.thing_being_named = thing_being_named + self.error_message = error_message + + @cached_property + def lazy_matcher(self) -> Pattern[str]: + return re.compile(self.regexp_pattern) + + def validate_name(self, *, name_to_check: str) -> None: + if self.lazy_matcher.match(name_to_check) is None: + formatted_message = self.error_message.format( + **{self.thing_being_named: name_to_check} + ) + raise ValueError(formatted_message) + + +class AxisNameValidation: + """Lateral wedge motor axis naming validation regular expression. + + Either permits (at start of name): + - leading underscore followed immediately by a letter, + or, + - leading letter, + then up to eight more optional alphanumeric characters. + """ + + _VALIDATOR: Final = NameValidator( + regexp_pattern=r"^(_[A-Za-z][A-Za-z0-9]{0,7}|[A-Za-z][A-Za-z0-9]{0,8})$", + thing_being_named="axis", + error_message="Axis name '{axis}' must start with a letter, or an underscore followed by a letter, later characters must alphanumeric: Max. length 9 characters.", + ) + + @classmethod + def validate_axis_name(cls, *, axis_name: str) -> None: + """Ensures motor axis names follow axis naming rules. + + Note: + The axis name must start either with a letter, or an underscore followed by a letter, + and subsequently optional characters can only be numbers or letters: Total length up to nine characters. + """ + cls._VALIDATOR.validate_name(name_to_check=axis_name) + + +class MaterialNameValidation: + """Absorber materials naming validation regular expression. + + Either permits (at start of name): + - leading underscore followed immediately by a letter, + or, + - leading letter, + then any number of alphanumerics or underscores or hyphens. + """ + + _VALIDATOR: Final = NameValidator( + regexp_pattern=r"^([A-Za-z]|_[A-Za-z])[A-Za-z0-9_-]*$", + thing_being_named="material", + error_message=( + "Material name '{material}' must start with a letter, or an underscore followed by any number of alphanumerics or underscores or hyphens." + ), + ) + + @classmethod + def validate_material_name(cls, *, material_name: str) -> None: + """Ensures material names follow the rules. + + Note: + The material name must start either with a letter, or an underscore followed by a letter, + and subsequent characters can only be alphanumerics, underscores or hyphens. + """ + cls._VALIDATOR.validate_name(name_to_check=material_name) + + +class WheelNameValidation: + """Wheel naming validation regular expression. + + Either permits (at start of name): + - leading underscore followed immediately by a letter, + or, + - leading letter, + then up to eleven more optional alphanumeric characters. + """ + + _VALIDATOR: Final = NameValidator( + regexp_pattern=r"^(_[A-Za-z][A-Za-z0-9]{0,10}|[A-Za-z][A-Za-z0-9]{0,11})$", + thing_being_named="wheel", + error_message=( + "Wheel name '{wheel}' must start with a letter, or an underscore followed by a letter, later optional characters must alphanumeric: Max. length 12 characters." + ), + ) + + @classmethod + def validate_wheel_name(cls, *, wheel_name: str) -> None: + """Ensures wheel names follow naming rules. + + Note: + The wheel name must start either with a letter, or an underscore followed by a letter, + and subsequently characters can only be numbers or letters: Total length up to 12 characters. + """ + cls._VALIDATOR.validate_name(name_to_check=wheel_name) diff --git a/src/dodal/devices/beamlines/i19/transmission/spec_from_config/system_aspect_base_parser.py b/src/dodal/devices/beamlines/i19/transmission/spec_from_config/system_aspect_base_parser.py new file mode 100644 index 00000000000..4427686fd77 --- /dev/null +++ b/src/dodal/devices/beamlines/i19/transmission/spec_from_config/system_aspect_base_parser.py @@ -0,0 +1,85 @@ +from abc import abstractmethod +from typing import Any, Generic, Self, TypeVar, cast + +from pydantic import ( + RootModel, + model_validator, +) + +from dodal.devices.beamlines.i19.transmission.spec_from_config.system_configuration import ( + SystemConfiguration, +) + +# 1. Define a local TypeVar for generic aspect specs +TxSubSpecT = TypeVar("TxSubSpecT") + + +class SystemAspectBaseParser(RootModel[dict[str, TxSubSpecT]], Generic[TxSubSpecT]): + """Generic Base Model handling dict validation and extraction from the (JSON) configuration highest level dict.""" + + @abstractmethod + def validate_key_name(self, *, key_name: str) -> None: ... + + @model_validator(mode="after") + def _ensure_all_keys_have_been_validly_named(self) -> Self: + """Validates all names for keys_present. + + Raises: + ValueError - if any aspect's sub-dict key names are not compliant. + """ + for key in self.keys(): + self.validate_key_name(key_name=key) + return self + + def keys(self): + """Returns the configuration sub-dict internal keys.""" + return self.root.keys() + + @classmethod + def name_all_specified( + cls, *, system_configuration: SystemConfiguration + ) -> list[str]: + """Names all the aspect instances found in the validated config, in a neat list of names.""" + _aspect_instance = cls._extract_system_aspect( + system_configuration=system_configuration + ) + _all_specified_elements = _aspect_instance.keys() + return list(_all_specified_elements) + + @classmethod + def get_aspect_specifications( + cls, *, system_configuration: SystemConfiguration + ) -> dict[str, TxSubSpecT]: + """Wraps all validated aspect instances, from config, in a dict against each instance's name. + + Args: + system_configuration: + The configuration JSON packaged with its structural template. + + Returns: + A subclass-dependent dict[ str, < aspect > ] of - for example - all: + - wedge driving motor specs against the name tag of each motor x,y etc. + - absorber material absorption spectrum specifications against the name of each material. + """ + _aspect_instance = cls._extract_system_aspect( + system_configuration=system_configuration + ) + return _aspect_instance.root + + @classmethod + def _extract_raw_sub_dict( + cls, *, system_configuration: SystemConfiguration + ) -> dict[str, Any]: + _aspect_key: str = system_configuration.get_sub_dict_key(target_type=cls) + return system_configuration.hardware_parameters[_aspect_key] + + @classmethod + def _extract_system_aspect( + cls, + *, + system_configuration: SystemConfiguration, + ) -> Self: + """Extracts specification and validates configuration for one aspect of the transmission system.""" + _sub_dict = cls._extract_raw_sub_dict(system_configuration=system_configuration) + _raw = cls.model_validate(_sub_dict) + return cast(Self, _raw) diff --git a/src/dodal/devices/beamlines/i19/transmission/spec_from_config/system_configuration.py b/src/dodal/devices/beamlines/i19/transmission/spec_from_config/system_configuration.py new file mode 100644 index 00000000000..8787bd7aedd --- /dev/null +++ b/src/dodal/devices/beamlines/i19/transmission/spec_from_config/system_configuration.py @@ -0,0 +1,34 @@ +from functools import cached_property +from typing import Any + +from pydantic import BaseModel, ConfigDict + + +class SystemConfiguration(BaseModel): + """Combination of configuration blob (typically JSON dict) and templating dict model for anticipated system config structure. + + Attributes: + structural_template: A specification class extending pydantic BaseModel (or RootModel) which has fields that map to dict elements. + hardware_parameters: A top level dict, typically extracted from JSON, which specifies the system hardware set up. + """ + + structural_template: type[BaseModel] + hardware_parameters: dict[str, dict[str, Any]] + + @cached_property + def field_mapping(self) -> dict[Any, str]: + """Maps transmission system highest level configuration annotations (types) to their corresponding names.""" + return { + info.annotation: field_name + for field_name, info in self.structural_template.model_fields.items() + } + + # Base Model internal setting to make this class immutable + model_config = ConfigDict( + arbitrary_types_allowed=True, + extra="forbid", + frozen=True, + ) + + def get_sub_dict_key(self, *, target_type: type) -> str: + return self.field_mapping[target_type] diff --git a/src/dodal/devices/beamlines/i19/transmission/spec_from_config/transmission_system_spec.py b/src/dodal/devices/beamlines/i19/transmission/spec_from_config/transmission_system_spec.py new file mode 100644 index 00000000000..e45d2c41ebb --- /dev/null +++ b/src/dodal/devices/beamlines/i19/transmission/spec_from_config/transmission_system_spec.py @@ -0,0 +1,40 @@ +from pydantic import ( + BaseModel, + ConfigDict, +) + +from dodal.devices.beamlines.i19.transmission.spec_from_config.lateral_motor_spec import ( + LateralMotorsConfig, +) +from dodal.devices.beamlines.i19.transmission.spec_from_config.material_absorption_spectrum_spec import ( + MaterialAbsorptionSpectralConfig, +) +from dodal.devices.beamlines.i19.transmission.spec_from_config.wedges_spec import ( + WedgesConfig, +) +from dodal.devices.beamlines.i19.transmission.spec_from_config.wheels_spec import ( + WheelsConfig, +) + + +class TransmissionSystemSpec(BaseModel): + """Maps input configuration (i.e. JSON ) to highest level dict for I19 transmission system. + + Note: + Sub-dictionaries carry the details of each aspect of the system. + + Attributes: + lateral_motors: Position (scale) for wedge motors. + materials: Absorption spectra for all absorber filter materials in wedges or foils. + wedges: The shape, motor and material details for all system variable depth wedge absorbers. + wheels: The slot occupancy, foil materials, motor name for all system filter wheels. + usage_priority: Beamline scientists policy on which absorbers to prefer using first. + """ + + lateral_motors: LateralMotorsConfig + materials: MaterialAbsorptionSpectralConfig + wedges: WedgesConfig + wheels: WheelsConfig + + # Base Model internal setting to make this class immutable and valid + model_config = ConfigDict(frozen=True, extra="forbid") diff --git a/src/dodal/devices/beamlines/i19/transmission/spec_from_config/wedges_spec.py b/src/dodal/devices/beamlines/i19/transmission/spec_from_config/wedges_spec.py new file mode 100644 index 00000000000..1c7b38bbe37 --- /dev/null +++ b/src/dodal/devices/beamlines/i19/transmission/spec_from_config/wedges_spec.py @@ -0,0 +1,116 @@ +from typing import Any + +from pydantic import ( + BaseModel, + ConfigDict, + Field, + RootModel, + StrictFloat, + model_validator, +) + +from dodal.common.general_maths.interval import OpenInterval +from dodal.devices.beamlines.i19.transmission.spec_from_config.name_validation import ( + AxisNameValidation, +) +from dodal.devices.beamlines.i19.transmission.spec_from_config.system_aspect_base_parser import ( + SystemAspectBaseParser, +) +from dodal.devices.beamlines.i19.transmission.spec_from_config.system_configuration import ( + SystemConfiguration, +) + + +class BubbleVoidsListParser(RootModel[list[dict[str, Any]]]): + """Parses and normalizes a list of bubble void sub-dictionaries before validation.""" + + @model_validator(mode="before") + @classmethod + def convert_bubble_voids(cls, data: Any) -> Any: + if isinstance(data, list): + # Convert scientist friendly {"from": ..., "to": ...} dicts to interval friendly {"lower": ..., "upper": ...} + return [ + {"lower": item["from"], "upper": item["to"]} + if isinstance(item, dict) and "from" in item and "to" in item + else item + for item in data + ] + _msg = f"Expecting wedge bubble voids to be specified in a list format, rather than {type(data)}." + raise ValueError(_msg) + + +class WedgeGeometrySpec(BaseModel): + """The specification for a wedge shape (as specified by configuration, typically **JSON**). + + Attributes: + taper_cotangent: slope of the wedge taper angle as defined by the cotangent of the small angle. + tip: Virtual position which ties the mathematically modelled wedge's location in space to the motor axis scale. + voids: optional listing of intervals where bubbles are to be avoided in the active range of wedge motor positions. + + See Also: + LateralMotorSpec: for the motor axis positions specification details. + """ + + taper_cotangent: StrictFloat = Field(gt=5.0) + tip: StrictFloat + voids: list[OpenInterval] = Field(default_factory=list) + + # Base Model internal setting to make this class immutable + model_config = ConfigDict(frozen=True, extra="forbid") + + @model_validator(mode="before") + @classmethod + def parse_voids(cls, data: Any) -> Any: + if isinstance(data, dict) and "voids" in data: + data["voids"] = BubbleVoidsListParser.model_validate(data["voids"]).root + return data + + +class WedgeSpec(BaseModel): + """The specification for a wedge motor (as specified by configuration, typically **JSON**). + + Notes: + - Each absorber wedge has a motor to drive it sideways across the x-ray beam: + - expected sideways motions are pure horizontal or pure vertical + - but that's a detail, azimuthal orientation around the beam should not matter. + - The wedge motor scale has + - an **out** position + - and then a range of "active" positions, ( where the mathematical assumption of a linear taper is reasonable ). + + Attributes: + material: Name of the absorber material + - as specified in the materials section of the transmission system configuration + + """ + + material: str + geometry: WedgeGeometrySpec + + # Base Model internal setting to make this class immutable + model_config = ConfigDict(frozen=True, extra="forbid") + + +class WedgesConfig(SystemAspectBaseParser[WedgeSpec]): + """Maps from highest level configuration (JSON) dict to extract wedge specifications. + + Uses base class method **extract_specifications** and some pythonic type handling magic. + + See Also: + Base of this parser class, namely **SystemAspectBaseParser** + """ + + def validate_key_name(self, *, key_name: str) -> None: + AxisNameValidation.validate_axis_name(axis_name=key_name) + + @classmethod + def extract_wedge_specifications( + cls, + *, + system_configuration: SystemConfiguration, + wedge_identifier: str, + ) -> WedgeSpec: + """Extracts wedge specification from configuration of the transmission system.""" + _wedges: dict[str, WedgeSpec] = cls.get_aspect_specifications( + system_configuration=system_configuration + ) + return _wedges[wedge_identifier] diff --git a/src/dodal/devices/beamlines/i19/transmission/spec_from_config/wheels_spec.py b/src/dodal/devices/beamlines/i19/transmission/spec_from_config/wheels_spec.py new file mode 100644 index 00000000000..38521ce44b9 --- /dev/null +++ b/src/dodal/devices/beamlines/i19/transmission/spec_from_config/wheels_spec.py @@ -0,0 +1,71 @@ +from pydantic import ( + BaseModel, + ConfigDict, + StrictInt, + model_validator, +) + +from dodal.devices.beamlines.i19.transmission.spec_from_config.foil_spec import FoilSpec +from dodal.devices.beamlines.i19.transmission.spec_from_config.name_validation import ( + WheelNameValidation, +) +from dodal.devices.beamlines.i19.transmission.spec_from_config.system_aspect_base_parser import ( + SystemAspectBaseParser, +) +from dodal.devices.beamlines.i19.transmission.spec_from_config.system_configuration import ( + SystemConfiguration, +) + + +class WheelSpec(BaseModel): + """The specification for a wheel (as specified by configuration, typically **JSON**). + + Notes: + - Each filter wheel has a motor to rotate the wheel into and out of the x-ray beam: + + Attributes: + foils: Occupancy dict mapping slot numbers to foils. + out: The slot number to use for removing all absorbers from the x-ray beam. + permissions: Those slot numbers which are permitted to be in use. + """ + + foils: dict[str, FoilSpec] + out: StrictInt + permissions: list[StrictInt] + + # Base Model internal setting to make this class immutable + model_config = ConfigDict(frozen=True, extra="forbid") + + @model_validator(mode="after") + def validate_out_in_permissions(self) -> "WheelSpec": + """Ensures the 'out' slot position is included in the 'permissions' list.""" + if self.out in self.permissions: + return self + _msg: str = f"The permitted wheel slot indices {self.permissions} must include the 'out' slot position ({self.out} - but do not!" + raise ValueError(_msg) + + +class WheelsConfig(SystemAspectBaseParser[WheelSpec]): + """Maps from highest level configuration (JSON) dict to extract wheel specifications. + + Uses base class method **extract_specifications** and some pythonic type handling magic. + + See Also: + Base of this parser class, namely **SystemAspectBaseParser** + """ + + def validate_key_name(self, *, key_name: str) -> None: + WheelNameValidation.validate_wheel_name(wheel_name=key_name) + + @classmethod + def extract_wheel_specifications( + cls, + *, + system_configuration: SystemConfiguration, + wheel_identifier: str, + ) -> WheelSpec: + """Extracts a specific wheels specification from configuration of the transmission system.""" + _wheels: dict[str, WheelSpec] = cls.get_aspect_specifications( + system_configuration=system_configuration + ) + return _wheels[wheel_identifier] diff --git a/src/dodal/devices/beamlines/i19/transmission/spec_from_json/material_absorption_spectrum_spec.py b/src/dodal/devices/beamlines/i19/transmission/spec_from_json/material_absorption_spectrum_spec.py deleted file mode 100644 index bd05452de8f..00000000000 --- a/src/dodal/devices/beamlines/i19/transmission/spec_from_json/material_absorption_spectrum_spec.py +++ /dev/null @@ -1,57 +0,0 @@ -from typing import Any - -from pydantic import BaseModel, ConfigDict, Field, RootModel - -from dodal.devices.beamlines.i19.transmission.spec_from_json.energy_interval_spec import ( - EnergyIntervalSpec, -) -from dodal.devices.beamlines.i19.transmission.spec_from_json.fitted_absorption_curve_spec import ( - FittedAbsorptionCurveSpec, -) - - -class AbsorptionVsEnergyRelation(BaseModel): - """JSON built dict for one element of an absorption spectrum. - - Pairs off the parameters of a fitted absorption curve, - against the energy interval over which the curve is valid. - - Attributes: - valid_energies: EnergyIntervalSpec - fit_parameters: FittedAbsorptionCurveSpec - """ - valid_energies: EnergyIntervalSpec - fit_parameters: FittedAbsorptionCurveSpec - - # Base Model internal setting to make this class immutable - model_config = ConfigDict(frozen=True) - - -class MaterialAbsorptionSpectrumSpec(BaseModel): - """JSON built dict for the absorption spectrum of a specific material. - - Note: - One spectrum absorption curve covers a specific energy range where it is valid. - One or more such curves make up one absorption spectrum. - - Attributes: - absorption_curves: List of absorption curves - """ - - absorption_curves: list[AbsorptionVsEnergyRelation] = Field(..., min_length=1) - - # Base Model internal setting to make this class immutable - model_config = ConfigDict(frozen=True) - - -class MaterialAbsorptionSpectrumJson(RootModel[dict[str, list[AbsorptionVsEnergyRelation]]]): - """Represents a JSON file where the top-level key is the material name. - - Maps material name to its absorption spectrum specification. - """ - - @classmethod - def extract_spectrum(cls, json_blob: dict[str, Any]) -> tuple[str, MaterialAbsorptionSpectrumSpec]: - parsed_blob = cls.model_validate(json_blob) - material, fitted_curves = next( iter(parsed_blob.root.items() )) - return material, MaterialAbsorptionSpectrumSpec(absorption_curves=fitted_curves) diff --git a/tests/devices/beamlines/i19/transmission/spec_from_json/__init__.py b/tests/devices/beamlines/i19/transmission/spec_from_config/__init__.py similarity index 100% rename from tests/devices/beamlines/i19/transmission/spec_from_json/__init__.py rename to tests/devices/beamlines/i19/transmission/spec_from_config/__init__.py diff --git a/tests/devices/beamlines/i19/transmission/spec_from_config/fake_json.py b/tests/devices/beamlines/i19/transmission/spec_from_config/fake_json.py new file mode 100644 index 00000000000..605ca063da6 --- /dev/null +++ b/tests/devices/beamlines/i19/transmission/spec_from_config/fake_json.py @@ -0,0 +1,244 @@ +from typing import Any, Final + +# By-pass the heavy weight use of config server for simple tests + +REALISTIC_SYSTEM_SPECIFICATION: Final[dict[str, Any]] = { + "lateral_motors": { + "x": { + "units": "mm", + "out": 0.1, + "threshold": 1.5, + "max": 75.0, + "tolerance": 5.0e-3, + }, + "y": { + "units": "mm", + "out": 5.0, + "threshold": 8.9, + "max": 98.0, + "tolerance": 5.0e-3, + }, + }, + "materials": { + "aluminium": [ + { + "valid_energies": {"units": "keV", "lower": 5.0, "upper": 30.0}, + "fit_parameters": { + "photon_absorption": 64873.0, + "roll_off": -2.96, + "residuals_polynomial_coeffs": [], + }, + } + ], + "gold": [ + { + "valid_energies": {"units": "keV", "lower": 5.0, "upper": 11.5}, + "fit_parameters": { + "photon_absorption": 7.87191216e5, + "roll_off": -2.52962901, + "residuals_polynomial_coeffs": [], + }, + }, + { + "valid_energies": {"units": "keV", "lower": 15.0, "upper": 30.0}, + "fit_parameters": { + "photon_absorption": 3.63187927e6, + "roll_off": -2.59815489, + "residuals_polynomial_coeffs": [], + }, + }, + ], + "iron": [ + { + "valid_energies": {"units": "keV", "lower": 8.0, "upper": 30.0}, + "fit_parameters": { + "photon_absorption": 1.034347e6, + "roll_off": -2.8655, + "residuals_polynomial_coeffs": [], + }, + } + ], + "resin1": [ + { + "valid_energies": {"units": "keV", "lower": 5.0, "upper": 22.0}, + "fit_parameters": { + "photon_absorption": 3814.1, + "roll_off": -2.81706, + "residuals_polynomial_coeffs": [ + 50.9211, + -23.6148, + 4.2138, + -0.3814, + 1.867e-2, + -4.709e-4, + 4.796e-6, + ], + }, + } + ], + }, + "wedges": { + "y": { + "material": "aluminium", + "geometry": {"taper_cotangent": 9.3985, "tip": 5.06, "voids": []}, + }, + "x": { + "material": "resin1", + "geometry": { + "taper_cotangent": 7.63359, + "tip": -3.959, + "voids": [{"from": 17.66, "to": 18.82}], + }, + }, + }, + "wheels": { + "w": { + "foils": { + "4": { + "absorber": { + "material": "iron", + "thickness": {"units": "um", "value": 516.07}, + } + }, + "2": { + "absorber": { + "material": "aluminium", + "thickness": {"units": "mm", "value": 6.285}, + } + }, + "6": { + "absorber": { + "material": "gold", + "thickness": {"units": "micron", "value": 25}, + } + }, + }, + "out": 1, + "permissions": [1], + } + }, +} + + +FAKE_SYSTEM_SPECIFICATION_1_JSON: Final[dict[str, Any]] = { + "lateral_motors": { + "a": { + "units": "mm", + "out": 0.7, + "threshold": 11.5, + "max": 61.0, + "tolerance": 5.0e-3, + }, + "b": { + "units": "mm", + "out": 3.0, + "threshold": 6.25, + "max": 85.2, + "tolerance": 5.0e-3, + }, + }, + "materials": { + "argon": [ + { + "valid_energies": {"units": "keV", "lower": 4.0, "upper": 23.0}, + "fit_parameters": { + "photon_absorption": 873.0, + "roll_off": -2.6, + "residuals_polynomial_coeffs": [], + }, + } + ], + "neon": [ + { + "valid_energies": {"units": "keV", "lower": 5.0, "upper": 7.5}, + "fit_parameters": { + "photon_absorption": 7.87191216e2, + "roll_off": -2.51, + "residuals_polynomial_coeffs": [], + }, + }, + { + "valid_energies": {"units": "keV", "lower": 12.0, "upper": 20.0}, + "fit_parameters": { + "photon_absorption": 3.63e2, + "roll_off": -2.81, + "residuals_polynomial_coeffs": [], + }, + }, + ], + "xenon": [ + { + "valid_energies": {"units": "keV", "lower": 8.0, "upper": 22.4}, + "fit_parameters": { + "photon_absorption": 1.43e2, + "roll_off": -2.79, + "residuals_polynomial_coeffs": [], + }, + } + ], + "krypton": [ + { + "valid_energies": {"units": "keV", "lower": 5.0, "upper": 12.0}, + "fit_parameters": { + "photon_absorption": 384.1, + "roll_off": -2.73, + "residuals_polynomial_coeffs": [ + 20.901, + -13.18, + 8.38, + -5.14, + 2.7e-2, + -4.691e-3, + 7.746e-5, + ], + }, + } + ], + }, + "wedges": { + "b": { + "material": "krypton", + "geometry": {"taper_cotangent": 9.2, "tip": 5.06, "voids": []}, + }, + "a": { + "material": "argon", + "geometry": { + "taper_cotangent": 7.63359, + "tip": -5.8, + "voids": [{"from": 15.66, "to": 21.82}], + }, + }, + }, + "wheels": { + "w": { + "foils": { + "4": { + "absorber": { + "material": "neon", + "thickness": {"units": "um", "value": 516.07}, + } + }, + "1": { + "absorber": { + "material": "neon", + "thickness": {"units": "um", "value": 1516.07}, + } + }, + "3": { + "absorber": { + "material": "argon", + "thickness": {"units": "mm", "value": 6.285}, + } + }, + "6": { + "absorber": { + "material": "xenon", + "thickness": {"units": "micron", "value": 65}, + } + }, + }, + "out": 5, + "permissions": [3, 4, 5], + } + }, +} diff --git a/tests/devices/beamlines/i19/transmission/spec_from_json/test_energy_interval_spec.py b/tests/devices/beamlines/i19/transmission/spec_from_config/test_energy_interval_spec.py similarity index 96% rename from tests/devices/beamlines/i19/transmission/spec_from_json/test_energy_interval_spec.py rename to tests/devices/beamlines/i19/transmission/spec_from_config/test_energy_interval_spec.py index c76bf4426fa..26e5fb7d386 100644 --- a/tests/devices/beamlines/i19/transmission/spec_from_json/test_energy_interval_spec.py +++ b/tests/devices/beamlines/i19/transmission/spec_from_config/test_energy_interval_spec.py @@ -4,7 +4,7 @@ import pytest from pydantic import ValidationError -from dodal.devices.beamlines.i19.transmission.spec_from_json.energy_interval_spec import ( +from dodal.devices.beamlines.i19.transmission.spec_from_config.energy_interval_spec import ( EnergyIntervalSpec, ) @@ -16,6 +16,7 @@ # Happy path tests + @pytest.mark.parametrize( "upper_bound", ( @@ -147,10 +148,7 @@ def test_that_energy_interval_json_validation_fails_with_typo_in_upper_bound_key EnergyIntervalSpec(**d) -@pytest.mark.parametrize( - "unsupported_units", - UNSUPPORTED_ENERGY_UNITS -) +@pytest.mark.parametrize("unsupported_units", UNSUPPORTED_ENERGY_UNITS) def test_that_energy_interval_json_validation_rejects_unsupported_energy_units( unsupported_units, ) -> None: diff --git a/tests/devices/beamlines/i19/transmission/spec_from_json/test_fitted_absorption_curve_spec.py b/tests/devices/beamlines/i19/transmission/spec_from_config/test_fitted_absorption_curve_spec.py similarity index 77% rename from tests/devices/beamlines/i19/transmission/spec_from_json/test_fitted_absorption_curve_spec.py rename to tests/devices/beamlines/i19/transmission/spec_from_config/test_fitted_absorption_curve_spec.py index 75c5f2cccf5..277ea58bcc7 100644 --- a/tests/devices/beamlines/i19/transmission/spec_from_json/test_fitted_absorption_curve_spec.py +++ b/tests/devices/beamlines/i19/transmission/spec_from_config/test_fitted_absorption_curve_spec.py @@ -1,11 +1,10 @@ - from math import pi from typing import Any, Final import pytest from pydantic import ValidationError -from dodal.devices.beamlines.i19.transmission.spec_from_json.fitted_absorption_curve_spec import ( +from dodal.devices.beamlines.i19.transmission.spec_from_config.fitted_absorption_curve_spec import ( FittedAbsorptionCurveSpec, ) @@ -13,15 +12,24 @@ # Happy path tests -WITH_OR_WITHOUT_RESIDUALS: Final[tuple[list[float], ...]] = ([], [40.9271, -21.6924, 7.5198, -12.51, 8.87e-2, -5.39e-4, 4.796e-5]) +WITH_OR_WITHOUT_RESIDUALS: Final[tuple[list[float], ...]] = ( + [], + [40.9271, -21.6924, 7.5198, -12.51, 8.87e-2, -5.39e-4, 4.796e-5], +) + +@pytest.mark.parametrize("residuals", WITH_OR_WITHOUT_RESIDUALS) @pytest.mark.parametrize( - "residuals", - WITH_OR_WITHOUT_RESIDUALS + "roll_off", + ( + -2.983, + -2.83, + -2.72, + ), ) -@pytest.mark.parametrize("roll_off", (-2.983, -2.83, -2.72,)) @pytest.mark.parametrize( - "scaling_constant", ( + "scaling_constant", + ( 3123.4, 7812.9, 1.305e6, @@ -38,10 +46,17 @@ def test_that_fitted_absorption_curve_spec_builds_from_valid_values( assert FittedAbsorptionCurveSpec(**d) is not None - -@pytest.mark.parametrize("roll_off", (-2.983, -2.83, -2.72,)) @pytest.mark.parametrize( - "scaling_constant", ( + "roll_off", + ( + -2.983, + -2.83, + -2.72, + ), +) +@pytest.mark.parametrize( + "scaling_constant", + ( 3123.4, 7812.9, 1.305e6, @@ -50,7 +65,7 @@ def test_that_fitted_absorption_curve_spec_builds_from_valid_values( def test_that_fitted_absorption_curve_spec_builds_when_residuals_are_in_tuple( scaling_constant: float, roll_off: float ) -> None: - _residuals_tuple: tuple[float,...] = ( -1.5, 0.03, 16.8, -272.345) + _residuals_tuple: tuple[float, ...] = (-1.5, 0.03, 16.8, -272.345) d: dict[str, Any] = { "photon_absorption": scaling_constant, "roll_off": roll_off, @@ -61,17 +76,11 @@ def test_that_fitted_absorption_curve_spec_builds_when_residuals_are_in_tuple( # Inauspicious path tests -@pytest.mark.parametrize( - "residuals", - WITH_OR_WITHOUT_RESIDUALS -) -@pytest.mark.parametrize( - "non_positive_scaling_constant", - ( 0, -2.8, -2903.4) -) + +@pytest.mark.parametrize("residuals", WITH_OR_WITHOUT_RESIDUALS) +@pytest.mark.parametrize("non_positive_scaling_constant", (0, -2.8, -2903.4)) def test_that_non_positive_scaling_constants_rejected_by_fitted_absorption_curve_spec( - non_positive_scaling_constant: float, - residuals: list[float] + non_positive_scaling_constant: float, residuals: list[float] ) -> None: d: dict[str, Any] = { "photon_absorption": non_positive_scaling_constant, @@ -91,8 +100,7 @@ def test_that_non_positive_scaling_constants_rejected_by_fitted_absorption_curve NON_NUMERICALS, ) def test_that_non_numerical_scaling_constants_rejected_by_fitted_absorption_curve_spec( - non_numerical_scaling_constant: Any, - residuals: list[float] + non_numerical_scaling_constant: Any, residuals: list[float] ) -> None: d: dict[str, Any] = { "photon_absorption": non_numerical_scaling_constant, @@ -102,10 +110,23 @@ def test_that_non_numerical_scaling_constants_rejected_by_fitted_absorption_curv with pytest.raises(ValidationError): FittedAbsorptionCurveSpec(**d) + UNPHYSICAL_ROLL_OFF_EXPONENTS: Final[list[float]] = [ - 1.2, 2.96, -1.5, -4.5, -296, -1, pi, 1, 4, 44.8, 200004, -1289, + 1.2, + 2.96, + -1.5, + -4.5, + -296, + -1, + pi, + 1, + 4, + 44.8, + 200004, + -1289, ] + @pytest.mark.parametrize( "residuals", WITH_OR_WITHOUT_RESIDUALS, @@ -115,8 +136,7 @@ def test_that_non_numerical_scaling_constants_rejected_by_fitted_absorption_curv UNPHYSICAL_ROLL_OFF_EXPONENTS, ) def test_that_unphysical_roll_off_exponent_rejected_by_fitted_absorption_curve_spec( - unphysical_exponent: float, - residuals: list[float] + unphysical_exponent: float, residuals: list[float] ) -> None: d: dict[str, Any] = { "photon_absorption": 53408.9, @@ -136,8 +156,7 @@ def test_that_unphysical_roll_off_exponent_rejected_by_fitted_absorption_curve_s NON_NUMERICALS, ) def test_that_non_numerical_roll_off_exponent_rejected_by_fitted_absorption_curve_spec( - non_numerical_exponent: Any, - residuals: list[float] + non_numerical_exponent: Any, residuals: list[float] ) -> None: d: dict[str, Any] = { "photon_absorption": 53408.9, @@ -155,7 +174,7 @@ def test_that_non_numerical_roll_off_exponent_rejected_by_fitted_absorption_curv def test_that_non_numerical_residual_rejected_by_fitted_absorption_curve_spec( non_numerical: Any, ) -> None: - _corrupted_residuals = [ 1.9, -9.1, non_numerical, -428.01312] + _corrupted_residuals = [1.9, -9.1, non_numerical, -428.01312] d: dict[str, Any] = { "photon_absorption": 63091.2, "roll_off": -2.78, @@ -170,7 +189,7 @@ def test_that_non_numerical_residual_rejected_by_fitted_absorption_curve_spec( "[1.9, 27.8, -55.81, -206.11]", ) def test_that_string_residuals_rejected_by_fitted_absorption_curve_spec( - stringy_residuals: str + stringy_residuals: str, ) -> None: d: dict[str, Any] = { "photon_absorption": 63091.2, @@ -179,4 +198,3 @@ def test_that_string_residuals_rejected_by_fitted_absorption_curve_spec( } with pytest.raises(ValidationError): FittedAbsorptionCurveSpec(**d) - diff --git a/tests/devices/beamlines/i19/transmission/spec_from_config/test_foil_spec.py b/tests/devices/beamlines/i19/transmission/spec_from_config/test_foil_spec.py new file mode 100644 index 00000000000..d6447d992de --- /dev/null +++ b/tests/devices/beamlines/i19/transmission/spec_from_config/test_foil_spec.py @@ -0,0 +1,54 @@ +import math +from typing import Any + +import pytest +from pydantic import ValidationError + +from dodal.devices.beamlines.i19.transmission.spec_from_config.foil_spec import FoilSpec + +# Happy path tests + + +@pytest.mark.parametrize( + "valid_foil_parameters", + [ + {"absorber": {"material": "iron", "thickness": {"units": "um", "value": 98.6}}}, + { + "absorber": { + "material": "gold", + "thickness": {"units": "um", "value": 14.08}, + } + }, + ], +) +def test_that_foil_spec_can_be_instantiated_from_valid_dict( + valid_foil_parameters: dict[str, Any], +) -> None: + _foil = FoilSpec.model_validate(valid_foil_parameters) + assert _foil is not None + + +# Happy path tests above + + +# Inauspicious path tests below + + +@pytest.mark.parametrize( + "invalid_dict", + [ + True, + False, + 0, + 8.4, + -13.2 - 9, + {"a", "nope"}, + math.log1p, + object(), + None, + KeyError(), + ], +) +def test_that_foil_spec_raises_error_when_dict_is_invalid(invalid_dict) -> None: + with pytest.raises(ValidationError): + FoilSpec.model_validate(invalid_dict) diff --git a/tests/devices/beamlines/i19/transmission/spec_from_config/test_lateral_motors_spec.py b/tests/devices/beamlines/i19/transmission/spec_from_config/test_lateral_motors_spec.py new file mode 100644 index 00000000000..6601725c741 --- /dev/null +++ b/tests/devices/beamlines/i19/transmission/spec_from_config/test_lateral_motors_spec.py @@ -0,0 +1,204 @@ +import copy +from typing import Any, Final + +import pytest +from pydantic import ValidationError + +from dodal.devices.beamlines.i19.transmission.spec_from_config.lateral_motor_spec import ( + LateralMotorsConfig, + LateralMotorSpec, +) +from dodal.devices.beamlines.i19.transmission.spec_from_config.system_configuration import ( + SystemConfiguration, +) +from dodal.devices.beamlines.i19.transmission.spec_from_config.transmission_system_spec import ( + TransmissionSystemSpec, +) +from tests.devices.beamlines.i19.transmission.spec_from_config.fake_json import ( + FAKE_SYSTEM_SPECIFICATION_1_JSON, + REALISTIC_SYSTEM_SPECIFICATION, +) + +JSON1: Final[dict[str, dict[str, Any]]] = REALISTIC_SYSTEM_SPECIFICATION +JSON2: Final[dict[str, dict[str, Any]]] = FAKE_SYSTEM_SPECIFICATION_1_JSON + +# happy path tests below + + +@pytest.mark.parametrize("hardware_parameters", [JSON1, JSON2]) +def test_that_lateral_motors_can_be_extracted_from_configuration_blob( + hardware_parameters: dict[str, dict[str, Any]], +) -> None: + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=hardware_parameters, + ) + all_motors: dict[str, LateralMotorSpec] = ( + LateralMotorsConfig.get_aspect_specifications( + system_configuration=_system_config + ) + ) + for axis, motor_spec in all_motors.items(): + assert motor_spec is not None, f"Lateral motor parameters for {axis} not found." + + +@pytest.mark.parametrize( + "hardware_parameters, expected_motor_axes", + [(JSON1, ["x", "y"]), (JSON2, ["a", "b"])], +) +def test_that_all_lateral_motor_axis_names_can_be_read( + hardware_parameters: dict[str, dict[str, Any]], + expected_motor_axes: list[str], +) -> None: + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=hardware_parameters, + ) + _axes = LateralMotorsConfig.name_all_specified(system_configuration=_system_config) + assert sorted(_axes) == sorted(expected_motor_axes) + + +@pytest.mark.parametrize( + "motor_axis, expected_maximum_position", [("b", 85.2), ("a", 61.0)] +) +def test_that_lateral_motor_specs_have_interrogatable_form( + motor_axis: str, expected_maximum_position: float +) -> None: + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=JSON2, + ) + _extracted_motor = LateralMotorsConfig.extract_motors_specifications( + system_configuration=_system_config, motor_identifier=motor_axis + ) + _max: float = _extracted_motor.max + assert _max == pytest.approx(expected=expected_maximum_position) + + +# Happy path above + +# Inauspicious path below + + +def test_that_lateral_motor_rejects_empty_json_blob() -> None: + _copied_json = copy.deepcopy(JSON1) + _null_motor = {"z": {}} + _copied_json["lateral_motors"] |= _null_motor # merge in the extra "motor" + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=_copied_json, + ) + with pytest.raises(ValidationError): + LateralMotorsConfig.get_aspect_specifications( + system_configuration=_system_config + ) + + +# typo here max <- Max +TYPO_MOTOR: Final[dict[str, Any]] = { + "t": { + "units": "mm", + "out": 0.19, + "threshold": 13.5, + "Max": 81.0, + "tolerance": 5.0e-3, + }, +} + + +def test_that_motor_json_blob_is_rejected_with_typo() -> None: + _copied_json = copy.deepcopy(JSON1) + _copied_json["lateral_motors"] |= TYPO_MOTOR # merge in the new motor entry + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=_copied_json, + ) + with pytest.raises(ValidationError): + LateralMotorsConfig.get_aspect_specifications( + system_configuration=_system_config + ) + + +# out position is mid wedge - so invalid +INVALID_MOTOR_1: Final[dict[str, Any]] = { + "t": { + "units": "mm", + "out": 49.065, + "threshold": 13.5, + "max": 81.0, + "tolerance": 5.0e-3, + }, +} + +# out position is mid wedge - so invalid +INVALID_MOTOR_2: Final[dict[str, Any]] = { + "t": { + "units": "mm", + "out": -29.134, + "threshold": -19.5, + "max": -58.4, + "tolerance": 5.0e-3, + }, +} + + +@pytest.mark.parametrize("invalid_motor", [INVALID_MOTOR_1, INVALID_MOTOR_2]) +def test_that_invalid_motor_is_rejected( + invalid_motor: dict[str, dict[str, Any]], +) -> None: + _copied_json = copy.deepcopy(JSON1) + _copied_json["lateral_motors"] |= INVALID_MOTOR_1 # merge in the new motor entry + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=_copied_json, + ) + with pytest.raises(ValidationError): + LateralMotorsConfig.get_aspect_specifications( + system_configuration=_system_config + ) + + +@pytest.mark.parametrize( + "invalid_axis_name", + ( + "", + "!", + "-4.2", + "5.8", + "7", + "+1", + " ", + "+x", + "_3", + "-62", + "Hg/Pb", + "£3.75", + "Aluminium.alloy", + "Zr€", + "99flake", + " - ", + "per: Capita", + ), +) +def test_that_motor_json_blob_is_rejected_without_valid_axis_name( + invalid_axis_name: str, +) -> None: + _misnamed_motor: Final[dict[str, Any]] = { + invalid_axis_name: { + "units": "mm", + "out": 5.87, + "threshold": 13.5, + "max": 60.78, + "tolerance": 5.0e-3, + }, + } + _copied_json = copy.deepcopy(JSON1) + _copied_json["lateral_motors"] |= _misnamed_motor + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=_copied_json, + ) + with pytest.raises(ValidationError): + LateralMotorsConfig.get_aspect_specifications( + system_configuration=_system_config + ) diff --git a/tests/devices/beamlines/i19/transmission/spec_from_config/test_material_absorption_spectrum_spec.py b/tests/devices/beamlines/i19/transmission/spec_from_config/test_material_absorption_spectrum_spec.py new file mode 100644 index 00000000000..bd32d3c9359 --- /dev/null +++ b/tests/devices/beamlines/i19/transmission/spec_from_config/test_material_absorption_spectrum_spec.py @@ -0,0 +1,228 @@ +import copy +import math +from typing import Any, Final + +import pytest +from pydantic import ValidationError + +from dodal.devices.beamlines.i19.transmission.spec_from_config.material_absorption_spectrum_spec import ( + MaterialAbsorptionSpectralConfig, + MaterialAbsorptionSpectrumSpec, +) +from dodal.devices.beamlines.i19.transmission.spec_from_config.system_configuration import ( + SystemConfiguration, +) +from dodal.devices.beamlines.i19.transmission.spec_from_config.transmission_system_spec import ( + TransmissionSystemSpec, +) +from tests.devices.beamlines.i19.transmission.spec_from_config.fake_json import ( + FAKE_SYSTEM_SPECIFICATION_1_JSON, + REALISTIC_SYSTEM_SPECIFICATION, +) + +JSON1: Final[dict[str, dict[str, Any]]] = REALISTIC_SYSTEM_SPECIFICATION +JSON2: Final[dict[str, dict[str, Any]]] = FAKE_SYSTEM_SPECIFICATION_1_JSON + +# happy path tests below + + +@pytest.mark.parametrize("hardware_parameters", [JSON1, JSON2]) +def test_that_material_absorption_spectra_can_be_extracted_from_configuration_blob( + hardware_parameters: dict[str, dict[str, Any]], +) -> None: + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=hardware_parameters, + ) + all_materials: dict[str, MaterialAbsorptionSpectrumSpec] = ( + MaterialAbsorptionSpectralConfig.get_aspect_specifications( + system_configuration=_system_config + ) + ) + for material, spectrum in all_materials.items(): + assert spectrum is not None, f"Spectrum for {material} not found." + + +@pytest.mark.parametrize( + "material_of_interest, expected_roll_off_exponents", + [ + ( + "krypton", + [-2.73], + ), + ( + "xenon", + [-2.79], + ), + ( + "argon", + [-2.6], + ), + ( + "neon", + [ + -2.51, + -2.81, + ], + ), + ], +) +def test_that_material_absorption_spectra_be_collated_in_interrogatable_form( + material_of_interest: str, expected_roll_off_exponents: list[float] +) -> None: + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=JSON2, + ) + _extracted_spectrum = ( + MaterialAbsorptionSpectralConfig.extract_absorber_material_specifications( + system_configuration=_system_config, material_name=material_of_interest + ) + ) + _roll_offs = [ + curve.fit_parameters.roll_off for curve in _extracted_spectrum.absorption_curves + ] + assert _roll_offs == pytest.approx(expected=expected_roll_off_exponents) + + +# Happy path tests above + +# Inauspicious path tests below + + +def test_that_material_absorption_spectra_rejected_if_no_materials_found_within() -> ( + None +): + _copied_json = copy.deepcopy(JSON1) + _copied_json["materials"] = {} # remove all materials + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=_copied_json, + ) + with pytest.raises(ValidationError): + MaterialAbsorptionSpectralConfig.get_aspect_specifications( + system_configuration=_system_config + ) + + +def test_that_material_absorption_spectra_rejects_empty_json_blob_for_named_material() -> ( + None +): + _copied_json = copy.deepcopy(JSON1) + _vacuous = {"Vacuum": {}} + _copied_json["materials"] |= _vacuous # merge in the new "material" entry + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=_copied_json, + ) + with pytest.raises(ValidationError): + MaterialAbsorptionSpectralConfig.get_aspect_specifications( + system_configuration=_system_config + ) + + +# typo here in residuals_polynomial_coeffs <- polNYomial +TYPO_MATERIAL: Final[dict[str, Any]] = { + "manganese": [ + { + "valid_energies": {"units": "keV", "lower": 19.0, "upper": 30.0}, + "fit_parameters": { + "photon_absorption": 2.037e5, + "roll_off": -2.85, + "residuals_polnyomial_coeffs": [], + }, + } + ] +} + + +def test_that_material_json_blob_is_rejected_with_typo() -> None: + _copied_json = copy.deepcopy(JSON1) + _copied_json["materials"] |= TYPO_MATERIAL # merge in the new "material" entry + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=_copied_json, + ) + with pytest.raises(ValidationError): + MaterialAbsorptionSpectralConfig.get_aspect_specifications( + system_configuration=_system_config + ) + + +# Noting this comment on valid names for materials in the absorption spectrum part of the JSON... +# "Material name '{m}' needed to be alphanumeric or underscores, (hyphens permitted after first character)." +@pytest.mark.parametrize( + "invalid_material_name", + ( + "", + " ", + "_3", + "-62", + "Hg/Pb", + "£3.75", + "Aluminium.alloy", + "Zr€", + "99flake", + " - ", + "per: Capita", + ), +) +def test_that_material_json_blob_is_rejected_without_valid_material_name( + invalid_material_name: str, +) -> None: + _misnamed_material: Final[dict[str, Any]] = { + invalid_material_name: [ + { + "valid_energies": {"units": "keV", "lower": 19.0, "upper": 23.0}, + "fit_parameters": { + "photon_absorption": 2.037e5, + "roll_off": -2.85, + "residuals_polynomial_coeffs": [], + }, + } + ] + } + _copied_json = copy.deepcopy(JSON1) + _copied_json["materials"] |= _misnamed_material + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=_copied_json, + ) + with pytest.raises(ValidationError): + MaterialAbsorptionSpectralConfig.get_aspect_specifications( + system_configuration=_system_config + ) + + +@pytest.mark.parametrize( + "invalid_spectrum", + [ + "neither_a_list_nor_a_dict", + True, + False, + -9.9, + 16, + 28.01, + math.sin, + math.pi, + (), + object(), + KeyError(), + ], +) +@pytest.mark.parametrize( + "material_of_interest", ["aluminium", "iron", "resin1", "gold"] +) +def test_that_material_absorption_spectra_rejects_invalidly_formatted_spectrum( + material_of_interest: str, invalid_spectrum: Any +) -> None: + _copied_json = copy.deepcopy(JSON1) + _copied_json["materials"][material_of_interest] = invalid_spectrum + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=_copied_json, + ) + with pytest.raises(ValidationError): + MaterialAbsorptionSpectralConfig.get_aspect_specifications( + system_configuration=_system_config + ) diff --git a/tests/devices/beamlines/i19/transmission/spec_from_config/test_system_configuration.py b/tests/devices/beamlines/i19/transmission/spec_from_config/test_system_configuration.py new file mode 100644 index 00000000000..07ac835446d --- /dev/null +++ b/tests/devices/beamlines/i19/transmission/spec_from_config/test_system_configuration.py @@ -0,0 +1,101 @@ +import copy +import math +from typing import Any + +import pytest +from pydantic import ValidationError + +from dodal.devices.beamlines.i19.transmission.spec_from_config.system_configuration import ( + SystemConfiguration, +) +from dodal.devices.beamlines.i19.transmission.spec_from_config.transmission_system_spec import ( + TransmissionSystemSpec, +) +from tests.devices.beamlines.i19.transmission.spec_from_config.fake_json import ( + FAKE_SYSTEM_SPECIFICATION_1_JSON, + REALISTIC_SYSTEM_SPECIFICATION, +) + +JSON1: dict[str, dict[str, Any]] = REALISTIC_SYSTEM_SPECIFICATION +JSON2: dict[str, dict[str, Any]] = FAKE_SYSTEM_SPECIFICATION_1_JSON + +# happy path test + + +@pytest.mark.parametrize( + "hardware_parameters", + [JSON1, JSON2], +) +def test_that_system_config_can_wrap_valid_configuration_dict( + *, + hardware_parameters: dict[str, Any], +) -> None: + _wrapped_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=hardware_parameters, + ) + assert _wrapped_config is not None + + +# happy path test above + +# inauspicious path tests below + + +@pytest.mark.parametrize( + "invalid_dict", + [ + True, + False, + 0, + 8.4, + -13.2 - 9, + {"a", "nope"}, + math.exp, + object(), + None, + KeyError(), + ], +) +def test_that_invalid_hardware_parameters_raise_error_when_wrapped( + *, + invalid_dict: Any, +) -> None: + with pytest.raises(ValidationError): + _wrapped_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=invalid_dict, + ) + + +@pytest.mark.parametrize( + "sub_dict_name", ["materials", "lateral_motors", "wedges", "wheels"] +) +@pytest.mark.parametrize( + "invalid_dict", + [ + {"a", None}, + True, + False, + 0, + 8.4, + -13.2, + -9, + {"a", "nope"}, + math.atan2, + object(), + None, + KeyError(), + ], +) +def test_that_test_that_config_validation_raises_error_when_subdict_is_invalid( + invalid_dict, + sub_dict_name, +) -> None: + _copied_json = copy.deepcopy(JSON1) + _copied_json[sub_dict_name] = invalid_dict + with pytest.raises(ValidationError): + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=_copied_json, + ) diff --git a/tests/devices/beamlines/i19/transmission/spec_from_config/test_wedges_spec.py b/tests/devices/beamlines/i19/transmission/spec_from_config/test_wedges_spec.py new file mode 100644 index 00000000000..cc9e500daf9 --- /dev/null +++ b/tests/devices/beamlines/i19/transmission/spec_from_config/test_wedges_spec.py @@ -0,0 +1,234 @@ +import copy +import math +from typing import Any, Final + +import pytest +from pydantic import ValidationError + +from dodal.devices.beamlines.i19.transmission.spec_from_config.system_configuration import ( + SystemConfiguration, +) +from dodal.devices.beamlines.i19.transmission.spec_from_config.transmission_system_spec import ( + TransmissionSystemSpec, +) +from dodal.devices.beamlines.i19.transmission.spec_from_config.wedges_spec import ( + WedgesConfig, +) +from tests.devices.beamlines.i19.transmission.spec_from_config.fake_json import ( + FAKE_SYSTEM_SPECIFICATION_1_JSON, + REALISTIC_SYSTEM_SPECIFICATION, +) +from tests.devices.beamlines.i19.transmission.spec_from_config.utility_constants import ( + NON_NUMERICALS, +) + +JSON1: Final[dict[str, dict[str, Any]]] = REALISTIC_SYSTEM_SPECIFICATION +JSON2: Final[dict[str, dict[str, Any]]] = FAKE_SYSTEM_SPECIFICATION_1_JSON + + +@pytest.mark.parametrize( + "hardware_parameters, wedge_identifier", + [ + ( + JSON1, + "y", + ), + ( + JSON2, + "a", + ), + ], +) +def test_that_wedges_config_can_be_constructed_from_valid_populated_dict( + *, hardware_parameters: dict[str, dict[str, Any]], wedge_identifier: str +) -> None: + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=hardware_parameters, + ) + _wedge_spec = WedgesConfig.get_aspect_specifications( + system_configuration=_system_config + ) + for wedge in _wedge_spec.keys(): + assert _wedge_spec[wedge] is not None + + +# Note an absence of wedges should not invalidate the configuration +# - what if the motors are being repaired and we just want to use filter wheels for now +# That scenario should be supported! + + +@pytest.mark.parametrize( + "hardware_parameters", + [JSON1, JSON2], +) +def test_that_wedges_config_can_validly_represent_absence( + hardware_parameters: dict[str, dict[str, Any]], +) -> None: + _copied_json = copy.deepcopy(hardware_parameters) + _copied_json["wedges"] = {} + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=_copied_json, + ) + _wedges_config = WedgesConfig._extract_system_aspect( + system_configuration=_system_config + ) + assert _wedges_config is not None + + +@pytest.mark.parametrize( + "hardware_parameters, wedge_with_voids", + [ + ( + JSON1, + "x", + ), + ( + JSON2, + "a", + ), + ], +) +def test_that_wedges_config_parses_correct_interval_for_voids( + hardware_parameters: dict[str, dict[str, Any]], + wedge_with_voids: str, +) -> None: + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=hardware_parameters, + ) + _wedge_spec = WedgesConfig.extract_wedge_specifications( + system_configuration=_system_config, wedge_identifier=wedge_with_voids + ) + # check void straddles a known bubbly position + _known_bubble_point = 17.89 + _wedge_geometry = _wedge_spec.geometry + _first_bubble_void = _wedge_geometry.voids[0] + assert _known_bubble_point in _first_bubble_void + + +# happy path tests above + +# inauspicious path tests + + +@pytest.mark.parametrize( + "invalid_cotangent", + NON_NUMERICALS, +) +@pytest.mark.parametrize("wedge", ["x", "y"]) +def test_that_wedge_spec_raises_error_when_cotangent_is_invalid( + wedge: str, + invalid_cotangent: Any, +) -> None: + _fake_json = copy.deepcopy(JSON1) + _fake_json["wedges"][wedge]["geometry"]["cotangent"] = invalid_cotangent + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=_fake_json, + ) + with pytest.raises(ValidationError): + _wedge_config = WedgesConfig.extract_wedge_specifications( + system_configuration=_system_config, + wedge_identifier=wedge, + ) + + +@pytest.mark.parametrize( + "invalid_tip", + NON_NUMERICALS, +) +@pytest.mark.parametrize("wedge", ["x", "y"]) +def test_that_wedge_spec_raises_error_when_tip_is_invalid( + wedge: str, + invalid_tip: Any, +) -> None: + _fake_json = copy.deepcopy(JSON1) + _fake_json["wedges"][wedge]["geometry"]["tip"] = invalid_tip + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=_fake_json, + ) + with pytest.raises(ValidationError): + _wedge_config = WedgesConfig.extract_wedge_specifications( + system_configuration=_system_config, + wedge_identifier=wedge, + ) + + +# deliberate typos +# first entry for material -> Naterial +# second entry for geometry -> gOEmetry + + +@pytest.mark.parametrize( + "hardware_parameters, wedge, good_key, typo_key", + [ + ( + JSON1, + "y", + "material", + "naterial", + ), + ( + JSON2, + "b", + "geometry", + "goemetry", + ), + ], +) +def test_that_system_config_with_typo_in_key_raises_error( + hardware_parameters: dict[str, dict[str, Any]], + wedge: str, + good_key: str, + typo_key: str, +) -> None: + _copied_json = copy.deepcopy(hardware_parameters) + _corruptable_wedge_sub_dict = _copied_json["wedges"][wedge] + _corruptable_wedge_sub_dict[typo_key] = _corruptable_wedge_sub_dict.pop(good_key) + + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=_copied_json, + ) + with pytest.raises(ValidationError): + _wedge_config = WedgesConfig.extract_wedge_specifications( + system_configuration=_system_config, + wedge_identifier=wedge, + ) + + +@pytest.mark.parametrize( + "invalid_voids", + [ + "Neither a list nor a dict", + "[16.2,25.3]", + -4.2, + 88, + math.pi, + -12, + True, + False, + ValueError(), + object(), + math.log, + ], +) +def test_that_wedge_specifying_voids_using_incorrect_type_is_rejected( + invalid_voids: Any, +) -> None: + _copied_json = copy.deepcopy(JSON1) + _corruptable_geometry = _copied_json["wedges"]["x"]["geometry"] + _corruptable_geometry["voids"] = ( + invalid_voids # string is not allowed to specify voids + ) + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=_copied_json, + ) + with pytest.raises(ValidationError): + _wedge_config = WedgesConfig.get_aspect_specifications( + system_configuration=_system_config, + ) diff --git a/tests/devices/beamlines/i19/transmission/spec_from_config/test_wheels_spec.py b/tests/devices/beamlines/i19/transmission/spec_from_config/test_wheels_spec.py new file mode 100644 index 00000000000..61df47d2616 --- /dev/null +++ b/tests/devices/beamlines/i19/transmission/spec_from_config/test_wheels_spec.py @@ -0,0 +1,169 @@ +import copy +from collections.abc import Iterable +from typing import Any, Final + +import pytest +from pydantic import ValidationError + +from dodal.devices.beamlines.i19.transmission.spec_from_config.system_configuration import ( + SystemConfiguration, +) +from dodal.devices.beamlines.i19.transmission.spec_from_config.transmission_system_spec import ( + TransmissionSystemSpec, +) +from dodal.devices.beamlines.i19.transmission.spec_from_config.wheels_spec import ( + WheelsConfig, +) +from tests.devices.beamlines.i19.transmission.spec_from_config.fake_json import ( + FAKE_SYSTEM_SPECIFICATION_1_JSON, + REALISTIC_SYSTEM_SPECIFICATION, +) +from tests.devices.beamlines.i19.transmission.spec_from_config.utility_constants import ( + NON_NUMERICALS, +) + +JSON1: Final[dict[str, dict[str, Any]]] = REALISTIC_SYSTEM_SPECIFICATION +JSON2: Final[dict[str, dict[str, Any]]] = FAKE_SYSTEM_SPECIFICATION_1_JSON + + +# Utility method tp prepare extracted lists etc ready for comparison with expectations. +def _convert_and_sort(original: Iterable[Any]) -> list[int]: + _integerised = [int(s) for s in original] + return sorted(_integerised) + + +def both_list_identical_integers(a: Iterable[Any], b: Iterable[Any]) -> bool: + return _convert_and_sort(a) == _convert_and_sort(b) + + +@pytest.mark.parametrize( + "hardware_parameters, expected_foils", + [ + ( + JSON1, + ["2", "4", "6"], + ), + ( + JSON2, + ["1", "3", "4", "6"], + ), + ], +) +def test_that_wheel_spec_correctly_captures_represents_slot_occupancy( + hardware_parameters: dict[str, dict[str, Any]], expected_foils: list[str] +) -> None: + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=hardware_parameters, + ) + _wheel_config = WheelsConfig.extract_wheel_specifications( + system_configuration=_system_config, wheel_identifier="w" + ) + _foils_present = _wheel_config.foils.keys() + assert both_list_identical_integers(_foils_present, expected_foils) + + +@pytest.mark.parametrize( + "hardware_parameters, expected_out_slot", + [ + (JSON1, 1), + (JSON2, 5), + ], +) +def test_that_wheel_spec_correctly_captures_slot_used_for_out_position( + hardware_parameters: dict[str, dict[str, Any]], expected_out_slot: int +) -> None: + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=hardware_parameters, + ) + _wheel_config = WheelsConfig.extract_wheel_specifications( + system_configuration=_system_config, wheel_identifier="w" + ) + _out_position = _wheel_config.out + assert _out_position == expected_out_slot + + +@pytest.mark.parametrize( + "hardware_parameters, expected_permitted_slots", + [ + (JSON1, [1]), + (JSON2, [5, 3, 4]), + ], +) +def test_that_wheel_spec_correctly_captures_slot_permissions( + hardware_parameters: dict[str, dict[str, Any]], expected_permitted_slots: list[int] +) -> None: + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=hardware_parameters, + ) + _wheel_config = WheelsConfig.extract_wheel_specifications( + system_configuration=_system_config, wheel_identifier="w" + ) + + assert both_list_identical_integers( + _wheel_config.permissions, expected_permitted_slots + ) + + +# Happy tests above + +# Inauspicious tests below + + +@pytest.mark.parametrize( + "invalid_out_slot", + NON_NUMERICALS, +) +def test_that_wheel_spec_raises_error_if_out_slot_is_invalid( + invalid_out_slot: Any, +) -> None: + _fake_json = copy.deepcopy(JSON1) + _fake_json["wheels"]["w"]["out"] = invalid_out_slot + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=_fake_json, + ) + with pytest.raises(ValidationError): + _wheel_config = WheelsConfig.extract_wheel_specifications( + system_configuration=_system_config, wheel_identifier="w" + ) + + +@pytest.mark.parametrize( + "json", + [JSON1, JSON2], +) +def test_that_wheel_spec_correctly_raises_error_if_out_slot_is_absent( + json: dict[str, dict[str, Any]], +) -> None: + _fake_json = copy.deepcopy(json) + _fake_json["wheels"]["w"].pop("out", None) + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=_fake_json, + ) + with pytest.raises(ValidationError): + _wheel_config = WheelsConfig.extract_wheel_specifications( + system_configuration=_system_config, wheel_identifier="w" + ) + + +@pytest.mark.parametrize( + "json", + [JSON1, JSON2], +) +def test_that_wheel_spec_correctly_raises_error_if_slot_permissions_empty( + json: dict[str, dict[str, Any]], +) -> None: + _fake_json = copy.deepcopy(json) + _fake_json["wheels"]["w"]["permissions"] = [] + _system_config: SystemConfiguration = SystemConfiguration( + structural_template=TransmissionSystemSpec, + hardware_parameters=_fake_json, + ) + with pytest.raises(ValidationError): + _wheel_config = WheelsConfig.extract_wheel_specifications( + system_configuration=_system_config, wheel_identifier="w" + ) diff --git a/tests/devices/beamlines/i19/transmission/spec_from_json/utility_constants.py b/tests/devices/beamlines/i19/transmission/spec_from_config/utility_constants.py similarity index 79% rename from tests/devices/beamlines/i19/transmission/spec_from_json/utility_constants.py rename to tests/devices/beamlines/i19/transmission/spec_from_config/utility_constants.py index c101063a078..12863a86506 100644 --- a/tests/devices/beamlines/i19/transmission/spec_from_json/utility_constants.py +++ b/tests/devices/beamlines/i19/transmission/spec_from_config/utility_constants.py @@ -1,4 +1,3 @@ - """Handy test constants, that this module will use for parametrized testing.""" from math import ( @@ -41,14 +40,14 @@ ) UNSUPPORTED_ENERGY_UNITS: Final[tuple[str, ...]] = ( - "meV", - "KILOelectronvolts", - "kev", - "k_eV", - "MeV", - "PeV", - "J", - "Joules", - "s", - "GHz", - ) + "meV", + "KILOelectronvolts", + "kev", + "k_eV", + "MeV", + "PeV", + "J", + "Joules", + "s", + "GHz", +) diff --git a/tests/devices/beamlines/i19/transmission/spec_from_json/test_material_absorption_spectrum_spec.py b/tests/devices/beamlines/i19/transmission/spec_from_json/test_material_absorption_spectrum_spec.py deleted file mode 100644 index 45fc5cf8215..00000000000 --- a/tests/devices/beamlines/i19/transmission/spec_from_json/test_material_absorption_spectrum_spec.py +++ /dev/null @@ -1,107 +0,0 @@ - -from typing import Any, Final - -import pytest - -from dodal.devices.beamlines.i19.transmission.spec_from_json.material_absorption_spectrum_spec import ( - MaterialAbsorptionSpectrumJson, - MaterialAbsorptionSpectrumSpec, -) - -ALU_SPECTRUM: Final[dict[str, Any]] = { - "aluminium": [ - { - "valid_energies": { - "units": "keV", - "lower" : 5.0, - "upper" : 30.0 - }, - "fit_parameters": { - "photon_absorption": 64873.0, - "roll_off":-2.96, - "residuals_polynomial_coeffs": [] - } - } - ] -} - -GOLD_SPECTRUM: Final[dict[str, Any]] = { - "gold": [ - { - "valid_energies": { - "units": "keV", - "lower": 5.0, - "upper":11.5 - }, - "fit_parameters": { - "photon_absorption": 7.87191216e5, - "roll_off":-2.52962901, - "residuals_polynomial_coeffs": [] - } - }, - { - "valid_energies": { - "units": "keV", - "lower": 15.0, - "upper":30.0 - }, - "fit_parameters": { - "photon_absorption": 3.63187927e6, - "roll_off":-2.59815489, - "residuals_polynomial_coeffs": [] - } - } - ], -} - -IRON_SPECTRUM: Final[dict[str, Any]] = { - "iron": [ - { - "valid_energies": { - "units": "keV", - "lower": 8.0, - "upper":30.0 - }, - "fit_parameters": { - "photon_absorption": 1.034347e6, - "roll_off":-2.8655, - "residuals_polynomial_coeffs": [] - } - } - ] -} - -RESIN_SPECTRUM: Final[dict[str, Any]] = { - "resin": [ - { - "valid_energies": { - "units": "keV", - "lower": 5.0, - "upper":22.0 - }, - "fit_parameters": { - "photon_absorption": 3814.1, - "roll_off":-2.81706, - "residuals_polynomial_coeffs": [ - 50.9211, - -23.6148, - 4.2138, - -0.3814, - 1.867e-2, - -4.709e-4, - 4.796e-6 - ] - } - } - ] -} - -@pytest.mark.parametrize( - "material_spectrum_blob", - [ALU_SPECTRUM, GOLD_SPECTRUM, IRON_SPECTRUM, RESIN_SPECTRUM] -) -def test_that_material_absorption_spectrum_can_be_built_from_dict_with_valid_content(material_spectrum_blob: dict[str, Any]): - material_name, spectrum = MaterialAbsorptionSpectrumJson.extract_spectrum(material_spectrum_blob) - - assert material_name in ["aluminium", "gold", "iron", "resin"] - assert isinstance(spectrum, MaterialAbsorptionSpectrumSpec)