Add Mbs driver and region classes

src/dodal/beamlines/i05.py

@@ -2,8 +2,9 @@
 from dodal.common.beamlines.beamline_utils import set_beamline as set_utils_beamline
 from dodal.device_manager import DeviceManager
 from dodal.devices.beamlines.i05 import I05Goniometer
-from dodal.devices.beamlines.i05_shared import M4M5Mirror
+from dodal.devices.beamlines.i05_shared import LensMode, M4M5Mirror, PassEnergy
 from dodal.devices.common_mirror import XYZSwitchingMirror
+from dodal.devices.electron_analyser.mbs import MbsAnalyserDriverIO
 from dodal.devices.hutch_shutter import HutchShutter
 from dodal.devices.temperture_controller import Lakeshore336
 from dodal.log import set_beamline as set_log_beamline
@@ -46,3 +47,12 @@ def sa() -> I05Goniometer:
         y_infix="SAY",
         z_infix="SAZ",
     )
+
+
+@devices.factory
+def analyser_driver() -> MbsAnalyserDriverIO:
+    return MbsAnalyserDriverIO[LensMode, PassEnergy](
+        prefix=f"{PREFIX.beamline_prefix}-EA-DET-02:CAM:",
+        lens_mode_type=LensMode,
+        pass_energy_type=PassEnergy,
+    )

src/dodal/beamlines/i05_1.py

@@ -2,8 +2,9 @@
 from dodal.common.beamlines.beamline_utils import set_beamline as set_utils_beamline
 from dodal.device_manager import DeviceManager
 from dodal.devices.beamlines.i05_1 import XYZAzimuthPolarDefocusStage
-from dodal.devices.beamlines.i05_shared import Mj7j8Mirror
+from dodal.devices.beamlines.i05_shared import LensMode, Mj7j8Mirror, PassEnergy
 from dodal.devices.common_mirror import XYZPiezoSwitchingMirror
+from dodal.devices.electron_analyser.mbs import MbsAnalyserDriverIO
 from dodal.devices.hutch_shutter import HutchShutter
 from dodal.log import set_beamline as set_log_beamline
 from dodal.utils import BeamlinePrefix, get_beamline_name
@@ -35,3 +36,12 @@ def nano_shutter() -> HutchShutter:
 def sm() -> XYZAzimuthPolarDefocusStage:
     """Sample Manipulator."""
     return XYZAzimuthPolarDefocusStage(prefix=f"{PREFIX.beamline_prefix}-EA-SM-01:")
+
+
+@devices.factory
+def analyser_driver() -> MbsAnalyserDriverIO[LensMode, PassEnergy]:
+    return MbsAnalyserDriverIO[LensMode, PassEnergy](
+        prefix=f"{PREFIX.beamline_prefix}-EA-DET-04:CAM:",
+        lens_mode_type=LensMode,
+        pass_energy_type=PassEnergy,
+    )

src/dodal/common/data_util.py

@@ -1,5 +1,6 @@
+from collections.abc import Callable
 from os.path import isabs, isfile, join, split
-from typing import Protocol, Self, TypeVar
+from typing import Generic, Self, TypeVar
 
 from pydantic import BaseModel
 
@@ -36,11 +37,14 @@ def save_class_to_json_file(model: BaseModel, file: str) -> None:
         f.write(model.model_dump_json())
 
 
-class JsonModelLoader(Protocol[TBaseModel]):
-    def __call__(self, file: str | None = None) -> TBaseModel: ...
+LoadModelFromFile = Callable[[str], TBaseModel]
 
 
-class JsonLoaderConfig(BaseModel):
+def json_model_loader(model: type[TBaseModel]) -> LoadModelFromFile[TBaseModel]:
+    return lambda file: load_json_file_to_class(model, file)
+
+
+class ModelLoaderConfig(BaseModel):
     default_path: str
     default_file: str | None
 
@@ -60,15 +64,40 @@ def update_config_from_file(self, new_file: str) -> None:
         self.default_path, self.default_file = split(new_file)
 
 
-def json_model_loader(
-    model: type[TBaseModel], config: JsonLoaderConfig | None = None
-) -> JsonModelLoader[TBaseModel]:
-    """Factory to create a function that loads a json file into a configured pydantic
-    model and with optional configuration for default path and file to use.
+class ModelLoader(Generic[TBaseModel]):
+    """A generic model loader that can be configured with any kind of method to read in
+    a file and convert the data into a pydantic model. It also takes configuration
+    to handle the file paths before they are passed to the method to convert to a
+    pydantic model.
     """
 
-    def load_json(file: str | None = None) -> TBaseModel:
-        """Load a json file and return it is as the configured pydantic model.
+    def __init__(
+        self,
+        load_model_from_file: LoadModelFromFile[TBaseModel],
+        cfg: ModelLoaderConfig | None = None,
+    ):
+        self._load_model_from_file = load_model_from_file
+        self._cfg = cfg
+
+    def _handle_file_path(self, file: str | None) -> str:
+        """Handle the file path based on the configuration provided. If a default path
+        is given and a relative file path used, it will join the default path and
+        relative path together. If a default file is configured, then you don't need to
+        provide a file when using __call__.
+        """
+        if file is None:
+            if self._cfg is None or self._cfg.default_file is None:
+                raise RuntimeError(
+                    "Model loader has no default file configured and no file was provided."
+                )
+            file = self._cfg.default_file
+
+        if not isabs(file) and self._cfg is not None:
+            file = join(self._cfg.default_path, file)
+        return file
+
+    def __call__(self, file: str | None = None) -> TBaseModel:
+        """Load a file and return it is as the configured pydantic model.
 
         Args:
             file (str, optional): The file to load into a pydantic class. If None
@@ -77,16 +106,5 @@ def load_json(file: str | None = None) -> TBaseModel:
         Returns:
             An instance of the configurated pydantic base_model type.
         """
-        if file is None:
-            if config is None or config.default_file is None:
-                raise RuntimeError(
-                    f"{model.__name__} loader has no default file configured "
-                    "and no file was provided."
-                )
-            file = config.default_file
-
-        if not isabs(file) and config is not None:
-            file = join(config.default_path, file)
-        return load_json_file_to_class(model, file)
-
-    return load_json
+        file = self._handle_file_path(file)
+        return self._load_model_from_file(file)

src/dodal/devices/beamlines/i05/__init__.py

@@ -1,5 +1,3 @@
 from .i05_motors import I05Goniometer
 
-__all__ = [
-    "I05Goniometer",
-]
+__all__ = ["I05Goniometer"]

src/dodal/devices/beamlines/i05_shared/__init__.py

@@ -1,23 +1,20 @@
-from dodal.devices.beamlines.i05_shared.apple_knot_constants import (
+from .apple_knot_constants import (
     APPLE_KNOT_EXCLUSION_ZONES,
     energy_to_gap_converter,
     energy_to_phase_converter,
 )
-from dodal.devices.beamlines.i05_shared.compound_motors import PolynomCompoundMotors
-from dodal.devices.beamlines.i05_shared.enums import (
-    Grating,
-    M3MJ6Mirror,
-    M4M5Mirror,
-    Mj7j8Mirror,
-)
+from .compound_motors import PolynomCompoundMotors
+from .enums import Grating, LensMode, M3MJ6Mirror, M4M5Mirror, Mj7j8Mirror, PassEnergy
 
 __all__ = [
     "Grating",
+    "LensMode",
     "Mj7j8Mirror",
     "M3MJ6Mirror",
     "M4M5Mirror",
     "PolynomCompoundMotors",
     "energy_to_gap_converter",
     "energy_to_phase_converter",
     "APPLE_KNOT_EXCLUSION_ZONES",
+    "PassEnergy",
 ]

src/dodal/devices/beamlines/i05_shared/enums.py

@@ -27,3 +27,22 @@ class Mj7j8Mirror(StrictEnum):
     MJ8 = "MJ8"
     MJ7 = "MJ7"
     REFERENCE = "Reference"
+
+
+class LensMode(StrictEnum):
+    L4_ANG0_D8 = "L4Ang0d8"
+    L4_ANG1_D6 = "L4Ang1d6"
+    L4_ANG3_D9 = "L4Ang3d9"
+    L4M_ANG0_D7 = "L4MAng0d7"
+    L4M_SPAT_5 = "L4MSpat5"
+
+
+class PassEnergy(StrictEnum):
+    PE001 = "PE001"
+    PE002 = "PE002"
+    PE005 = "PE005"
+    PE010 = "PE010"
+    PE020 = "PE020"
+    PE050 = "PE050"
+    PE100 = "PE100"
+    PE200 = "PE200"

src/dodal/devices/electron_analyser/base/base_driver_io.py

@@ -1,5 +1,6 @@
 from abc import ABC, abstractmethod
-from typing import Generic, TypeAlias, TypeVar
+from dataclasses import dataclass
+from typing import ClassVar, Generic, TypeAlias, TypeVar
 
 import numpy as np
 from bluesky.protocols import Movable
@@ -29,10 +30,27 @@
     TPassEnergy,
 )
 
-AnyPsuMode: TypeAlias = SupersetEnum | StrictEnum
+AnyPsuMode: TypeAlias = SupersetEnum | StrictEnum | str
 TPsuMode = TypeVar("TPsuMode", bound=AnyPsuMode)
 
-_PSU = "PSU_MODE"
+
+@dataclass(frozen=True)
+class ElectronAnalyserPVConfig:
+    """Configuration for PV's. Temporary work around until PV's are standardised between
+    beamlines.
+    """
+
+    low_energy: str = "LOW_ENERGY"
+    high_energy: str = "HIGH_ENERGY"
+    centre_energy: str = "CENTRE_ENERGY"
+    slices: str = "SLICES"
+    lens_mode: str = "LENS_MODE"
+    pass_energy: str = "PASS_ENERGY"
+    energy_step: str = "STEP_SIZE"
+    iterations: str = "NumExposures"
+    acquisition_mode: str = "ACQ_MODE"
+    psu_mode: str = "PSU_MODE"
+    total_steps: str = "TOTAL_POINTS_RBV"
 
 
 class AbstractAnalyserDriverIO(
@@ -57,56 +75,56 @@ class AbstractAnalyserDriverIO(
         pass_energy_type (type[TPassEnergy]): Can be enum or float, depending on
             electron analyser model. If enum, it determines the available pass
             energies for this device.
-        psu_suffix (str, optional): The psu infix to connect to EPICS. Defaults to PSU_MODE.
         name (str, optional): Name of the device.
     """
 
+    PV_CFG: ClassVar[ElectronAnalyserPVConfig]
+
     def __init__(
         self,
         prefix: str,
         acquisition_mode_type: type[TAcquisitionMode],
         lens_mode_type: type[TLensMode],
         psu_mode_type: type[TPsuMode],
         pass_energy_type: type[TPassEnergy],
-        psu_suffix: str = _PSU,
         name: str = "",
     ) -> None:
         self.acquisition_mode_type = acquisition_mode_type
         self.lens_mode_type = lens_mode_type
         self.psu_mode_type = psu_mode_type
         self.pass_energy_type = pass_energy_type
 
-        # must call first to initiate parent variables
+        # Must call first to initiate parent variables
         super().__init__(prefix=prefix, name=name)
 
-        with self.add_children_as_readables():
-            self.image = epics_signal_r(Array1D[np.float64], prefix + "IMAGE")
-            self.spectrum = epics_signal_r(Array1D[np.float64], prefix + "INT_SPECTRUM")
-            self.total_intensity = derived_signal_r(
-                self._calculate_total_intensity, spectrum=self.spectrum
-            )
-
         with self.add_children_as_readables(StandardReadableFormat.CONFIG_SIGNAL):
             # Read once per scan after data acquired
             # Used for setting up region data acquisition
             self.region_name = soft_signal_rw(str, initial_value="null")
             self.energy_mode = soft_signal_rw(
                 EnergyMode, initial_value=EnergyMode.KINETIC
             )
-            self.low_energy = epics_signal_rw(float, prefix + "LOW_ENERGY")
-            self.centre_energy = epics_signal_rw(float, prefix + "CENTRE_ENERGY")
-            self.high_energy = epics_signal_rw(float, prefix + "HIGH_ENERGY")
-            self.slices = epics_signal_rw(int, prefix + "SLICES")
-            self.lens_mode = epics_signal_rw(lens_mode_type, prefix + "LENS_MODE")
-            self.pass_energy = epics_signal_rw(pass_energy_type, prefix + "PASS_ENERGY")
-            self.energy_step = epics_signal_rw(float, prefix + "STEP_SIZE")
-            self.iterations = epics_signal_rw(int, prefix + "NumExposures")
+            self.low_energy = epics_signal_rw(float, prefix + self.PV_CFG.low_energy)
+            self.centre_energy = epics_signal_rw(
+                float, prefix + self.PV_CFG.centre_energy
+            )
+            self.high_energy = epics_signal_rw(float, prefix + self.PV_CFG.high_energy)
+            self.slices = epics_signal_rw(int, prefix + self.PV_CFG.slices)
+            self.lens_mode = epics_signal_rw(
+                lens_mode_type, prefix + self.PV_CFG.lens_mode
+            )
+            self.pass_energy = epics_signal_rw(
+                pass_energy_type, prefix + self.PV_CFG.pass_energy
+            )
+            self.energy_step = epics_signal_rw(float, prefix + self.PV_CFG.energy_step)
+            self.iterations = epics_signal_rw(int, prefix + self.PV_CFG.iterations)
             self.acquisition_mode = epics_signal_rw(
-                acquisition_mode_type, prefix + "ACQ_MODE"
+                acquisition_mode_type, prefix + self.PV_CFG.acquisition_mode
             )
-            # This is used by each electron analyser, however it depends on the electron
-            # analyser type to know if is moved with region settings.
-            self.psu_mode = epics_signal_rw(psu_mode_type, prefix + psu_suffix)
+            # This is used by each electron analyser, however it is not writeable for
+            # all types and it depends on the electron analyser type to know if is moved
+            # with region settings.
+            self.psu_mode = epics_signal_r(psu_mode_type, prefix + self.PV_CFG.psu_mode)
 
         # This is defined in the parent class, add it as readable configuration.
         self.add_readables([self.acquire_time], StandardReadableFormat.CONFIG_SIGNAL)
@@ -115,7 +133,7 @@ def __init__(
             # NOT used for setting up region data acquisition.
             self.energy_axis = self._create_energy_axis_signal(prefix)
             self.angle_axis = self._create_angle_axis_signal(prefix)
-            self.total_steps = epics_signal_r(int, prefix + "TOTAL_POINTS_RBV")
+            self.total_steps = epics_signal_r(int, prefix + self.PV_CFG.total_steps)
             self.total_time = derived_signal_r(
                 self._calculate_total_time,
                 "s",

src/dodal/devices/electron_analyser/base/base_region.py

@@ -74,16 +74,15 @@ class BaseRegion(
 
     name: str = "New_region"
     enabled: bool = False
-    slices: int = 1
     iterations: int = 1
     excitation_energy_source: SelectedSource = SelectedSource.SOURCE1
     # These ones we need subclasses to provide sensible default values
     lens_mode: TLensMode
     pass_energy: TPassEnergy
     acquisition_mode: TAcquisitionMode
     low_energy: float
-    centre_energy: float
     high_energy: float
+    centre_energy: float
     acquire_time: float
     energy_step: float  # in eV
     energy_mode: EnergyMode = EnergyMode.KINETIC

src/dodal/devices/electron_analyser/mbs/__init__.py

@@ -0,0 +1,5 @@
+from .mbs_driver_io import MbsAnalyserDriverIO
+from .mbs_enums import AcquisitionMode
+from .mbs_region import MbsRegion, MbsSequence
+
+__all__ = ["MbsAnalyserDriverIO", "AcquisitionMode", "MbsRegion", "MbsSequence"]