Add i06-1 magnets

src/dodal/beamlines/i06_1.py

@@ -2,6 +2,10 @@
 from dodal.common.beamlines.beamline_utils import set_beamline as set_utils_beamline
 from dodal.device_manager import DeviceManager
 from dodal.devices.beamlines.i06_1 import DiffractionDichroism
+from dodal.devices.beamlines.i06_1.magnets import (
+    MagnetThreeAxesRampRateController,
+    SuperConductingMagnetController,
+)
 from dodal.devices.motors import XYThetaStage
 from dodal.devices.temperture_controller import Lakeshore336
 from dodal.log import set_beamline as set_log_beamline
@@ -17,12 +21,12 @@
 
 
 @devices.factory()
-def diff_cooling_temperature_controller() -> Lakeshore336:
+def ls336_cooling() -> Lakeshore336:
     return Lakeshore336(prefix=f"{PREFIX.beamline_prefix}-EA-TCTRL-02:")
 
 
 @devices.factory()
-def diff_heating_temperature_controller() -> Lakeshore336:
+def ls336_heating() -> Lakeshore336:
     return Lakeshore336(prefix=f"{PREFIX.beamline_prefix}-EA-TCTRL-03:")
 
 
@@ -34,3 +38,17 @@ def xabs() -> XYThetaStage:
 @devices.factory()
 def dd() -> DiffractionDichroism:
     return DiffractionDichroism(f"{PREFIX.beamline_prefix}-EA-DDIFF-01:")
+
+
+@devices.factory()
+def mag_ramp_rate() -> MagnetThreeAxesRampRateController:
+    return MagnetThreeAxesRampRateController(f"{PREFIX.beamline_prefix}-EA-SMC")
+
+
+@devices.factory()
+def scmc(
+    mag_ramp_rate: MagnetThreeAxesRampRateController,
+) -> SuperConductingMagnetController:
+    return SuperConductingMagnetController(
+        f"{PREFIX.beamline_prefix}-EA-MAG-01:", mag_ramp_rate
+    )

src/dodal/devices/beamlines/i06_1/magnets/__init__.py

@@ -0,0 +1,27 @@
+from .movement import MagnetPosition, MagnetPositionError, MagnetSphericalPosition
+from .ramp_controller import (
+    MagnetAxisRampRateController,
+    MagnetThreeAxesRampRateController,
+)
+from .superconducting_magnet import (
+    FlyMagnetInfo,
+    MagnetAxis,
+    MagnetLimitStatus,
+    MagnetModes,
+    MagnetRampStatus,
+    SuperConductingMagnetController,
+)
+
+__all__ = [
+    "MagnetPosition",
+    "MagnetPositionError",
+    "MagnetSphericalPosition",
+    "MagnetAxisRampRateController",
+    "MagnetThreeAxesRampRateController",
+    "FlyMagnetInfo",
+    "MagnetAxis",
+    "MagnetLimitStatus",
+    "MagnetModes",
+    "MagnetRampStatus",
+    "SuperConductingMagnetController",
+]

src/dodal/devices/beamlines/i06_1/magnets/enums.py

@@ -0,0 +1,22 @@
+from ophyd_async.core import StrictEnum
+
+
+class MagnetModes(StrictEnum):
+    UNIAXIAL_X = "UNIAXIAL_X"
+    UNIAXIAL_Y = "UNIAXIAL_Y"
+    UNIAXIAL_Z = "UNIAXIAL_Z"
+    SPHERICAL = "SPHERICAL"
+    PLANAR_XZ = "PLANAR_XZ"
+    QUADRANT_XY = "QUADRANT_XY"
+    CUBIC = "CUBIC"
+    UNDEFINED = "UNDEFINED"
+
+
+class MagnetRampStatus(StrictEnum):
+    RAMP_MADE = "RAMP MADE"
+    RAMPING = "RAMPING"
+
+
+class MagnetLimitStatus(StrictEnum):
+    OK = "OK"
+    VIOLTATION = "VIOLATION"

src/dodal/devices/beamlines/i06_1/magnets/movement.py

@@ -0,0 +1,187 @@
+import math
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+
+
+def read_theta(x: float, z: float) -> float:
+    theta = math.degrees(math.atan2(-x, z))
+    return theta % 360
+
+
+def read_rho(x: float, y: float, z: float) -> float:
+    return math.hypot(x, y, z)
+
+
+def read_phi(x: float, y: float, z: float) -> float:
+    return math.degrees(math.atan2(math.hypot(x, z), y))
+
+
+@dataclass(kw_only=True)
+class MagnetPosition:
+    x: float
+    y: float
+    z: float
+
+    def to_spherical(self) -> "MagnetSphericalPosition":
+        return MagnetSphericalPosition(
+            rho=read_rho(self.x, self.y, self.z),
+            theta=read_theta(self.x, self.z),
+            phi=read_phi(self.x, self.y, self.z),
+        )
+
+
+@dataclass(kw_only=True)
+class MagnetSphericalPosition:
+    rho: float
+    theta: float  # degrees
+    phi: float  # degrees
+
+    def to_cartesian(self) -> MagnetPosition:
+        theta = math.radians(self.theta)
+        phi = math.radians(self.phi)
+
+        return MagnetPosition(
+            x=-self.rho * math.sin(phi) * math.sin(theta),
+            y=self.rho * math.cos(phi),
+            z=self.rho * math.sin(phi) * math.cos(theta),
+        )
+
+
+@dataclass(frozen=True, kw_only=True)
+class MagnetStep:
+    x: float | None = None
+    y: float | None = None
+    z: float | None = None
+
+
+class MagnetPositionError(Exception):
+    pass
+
+
+class MovementStrategy(ABC):
+    @abstractmethod
+    def moves(
+        self, current: MagnetPosition, target: MagnetPosition
+    ) -> list[MagnetStep]:
+        """Return a sequence of safe moves."""
+
+
+class SphericalMovement(MovementStrategy):
+    def moves(
+        self, current: MagnetPosition, target: MagnetPosition
+    ) -> list[MagnetStep]:
+        steps: list[MagnetStep] = []
+
+        decreases = [
+            ("z", current.z, target.z),
+            ("x", current.x, target.x),
+            ("y", current.y, target.y),
+        ]
+
+        for axis, old, new in decreases:
+            if abs(new) < abs(old):
+                kwargs = {axis: new}
+                steps.append(MagnetStep(**kwargs))
+
+        if any(abs(new) > abs(old) for _, old, new in decreases):
+            steps.append(MagnetStep(x=target.x, y=target.y, z=target.z))
+
+        return steps
+
+
+class CubicMovement(MovementStrategy):
+    LIMIT = 1.5
+
+    def moves(
+        self, current: MagnetPosition, target: MagnetPosition
+    ) -> list[MagnetStep]:
+        for axis in (target.x, target.y, target.z):
+            if abs(axis) > self.LIMIT:
+                raise MagnetPositionError(
+                    f"Target {target} outside cubic operating region limit of {self.LIMIT}"
+                )
+
+        return [MagnetStep(x=target.x, y=target.y, z=target.z)]
+
+
+class PlanarXZMovement(MovementStrategy):
+    LIMIT = 2.0
+
+    def moves(
+        self, current: MagnetPosition, target: MagnetPosition
+    ) -> list[MagnetStep]:
+        if target.y != 0:
+            raise MagnetPositionError(
+                f"Y must remain zero in XZ mode. Requested value was {target.y}"
+            )
+
+        hypot = math.hypot(target.x, target.z)
+        if hypot > self.LIMIT:
+            raise MagnetPositionError(
+                f"Outside XZ operating region. math.hypot(x={target.x}, y={target.y}) = {hypot}. Limit is {self.LIMIT}"
+            )
+
+        return [MagnetStep(x=target.x, z=target.z)]
+
+
+@dataclass
+class UniaxialMovement(MovementStrategy):
+    axis: str
+    limit: float
+
+    def moves(
+        self, current: MagnetPosition, target: MagnetPosition
+    ) -> list[MagnetStep]:
+        coords = {"x": target.x, "y": target.y, "z": target.z}
+        for axis, value in coords.items():
+            if axis == self.axis:
+                abs_value = abs(value)
+                if abs_value > self.limit:
+                    raise MagnetPositionError(
+                        f"{axis} value of {abs_value} exceeds limit of {self.limit}"
+                    )
+            elif value != 0:
+                raise MagnetPositionError(
+                    f"{axis} must remain zero in {self.axis.upper()} mode. Requested value was {value} "
+                )
+
+        return [MagnetStep(**{self.axis: coords[self.axis]})]
+
+
+class QuadrantXYMovement(MovementStrategy):
+    """Start from zero field in all axes. Appy 2.o T in + y direction. Move field
+    direction through positive quadrant to 2.0 T in the +x direction. At all times
+    | B | <= 2.0 T. Ramp field back to zero in x direction.
+    """
+
+    LIMIT = 2.0
+
+    def moves(
+        self, current: MagnetPosition, target: MagnetPosition
+    ) -> list[MagnetStep]:
+        if target.z != 0:
+            raise MagnetPositionError(
+                f"Z must remain zero in QUADRANT_XY mode. You provided {target.z}"
+            )
+
+        hypot = math.hypot(target.x, target.y)
+        if hypot > self.LIMIT:
+            raise MagnetPositionError(
+                f"Target math.hypot(x={target.x}, y={target.y}) {hypot} is outside the XY operating region limit of {self.LIMIT}."
+            )
+
+        steps: list[MagnetStep] = []
+
+        # Step 1: remove X component first
+        if current.x != 0:
+            steps.append(MagnetStep(x=0))
+
+        # Step 2: move Y
+        if current.y != target.y:
+            steps.append(MagnetStep(y=target.y))
+
+        # Step 3: move X to its final value
+        if target.x != 0:
+            steps.append(MagnetStep(x=target.x))
+
+        return steps

src/dodal/devices/beamlines/i06_1/magnets/ramp_controller.py

@@ -0,0 +1,76 @@
+from dataclasses import dataclass
+from functools import cached_property
+
+from ophyd_async.core import (
+    InstantMovableMock,
+    MovableLogic,
+    SignalR,
+    StandardMovable,
+    StandardReadable,
+    StandardReadableFormat,
+    TimeoutCalculator,
+    default_mock_class,
+    set_mock_value,
+)
+from ophyd_async.epics.core import epics_signal_rw
+
+
+@dataclass
+class RampRateMovableLogic(MovableLogic[float]):
+    limit: SignalR[float]
+
+    async def check_move(self, new_position: float) -> None:
+        limit = await self.limit.get_value()
+        if new_position > limit:
+            raise ValueError(
+                f"Requested ramp rate {new_position} exceeds the maximum limit of {limit} for device {self.readback.name}."
+            )
+
+    async def move(self, new_position: float, timeout: TimeoutCalculator):
+        await self.setpoint.set(new_position, timeout=timeout())
+
+
+class MockMagnetAxisRampRateController(InstantMovableMock):
+    async def connect(self, device: StandardMovable):
+        await super().connect(device)
+        # Extend to set a sensible default value for the limit.
+        set_mock_value(device.limit, 2)  # type: ignore
+
+
+@default_mock_class(MockMagnetAxisRampRateController)
+class MagnetAxisRampRateController(StandardMovable[float], StandardReadable):
+    """Controls the ramp rate of a single superconducting magnet axis.
+
+    Exposes the readback ramp rate, demand ramp rate and the maximum
+    permitted ramp rate for one magnet axis.
+    """
+
+    def __init__(self, prefix: str, name: str = ""):
+        with self.add_children_as_readables(StandardReadableFormat.HINTED_SIGNAL):
+            self.readback = epics_signal_rw(float, prefix + "STS:RAMPRATE:TPM")
+        self.demand = epics_signal_rw(float, prefix + "SET:DMD:RAMPRATE:TPM")
+        self.limit = epics_signal_rw(float, prefix + "LIM:RAMPRATE:TPM")
+        super().__init__(name)
+
+    @cached_property
+    def movable_logic(self) -> RampRateMovableLogic:
+        return RampRateMovableLogic(
+            readback=self.readback, setpoint=self.demand, limit=self.limit
+        )
+
+
+class MagnetThreeAxesRampRateController(StandardReadable):
+    """Groups the ramp rate controllers for the x, y and z magnet axes.
+
+    This device is passed to :class:`SuperConductingMagnetController` so that each
+    :class:`MagnetAxis` can configure its own ramp rate during preparation for
+    fly scans.
+    """
+
+    def __init__(self, prefix: str, name: str = ""):
+        with self.add_children_as_readables():
+            self.x = MagnetAxisRampRateController(prefix + "-01:")
+            self.y = MagnetAxisRampRateController(prefix + "-02:")
+            self.z = MagnetAxisRampRateController(prefix + "-03:")
+
+        super().__init__(name)