add temperature controller

src/dodal/beamlines/i10_1.py

@@ -8,6 +8,9 @@
     HighFieldMagnet,
     I10JScalerCard,
 )
+from dodal.devices.beamlines.i10_1.high_field_magnet.temperature_controller import (
+    HFMTemperatureController,
+)
 from dodal.devices.common_mirror import XYZPiezoCollimatingMirror
 from dodal.devices.current_amplifiers import (
     SR570,
@@ -254,3 +257,18 @@ def high_field_magnet() -> HighFieldMagnet:
     return HighFieldMagnet(
         prefix=f"{PREFIX.beamline_prefix}-EA-SMC-01:",
     )
+
+
+@devices.factory()
+def hfm_temp() -> HFMTemperatureController:
+    return HFMTemperatureController(prefix=f"{PREFIX.beamline_prefix}-EA-TCTRL-01:")
+
+
+@devices.factory()
+def hfm_vti() -> HFMTemperatureController:
+    return HFMTemperatureController(prefix=f"{PREFIX.beamline_prefix}-EA-TCTRL-02:")
+
+
+@devices.factory()
+def hfm_he3() -> HFMTemperatureController:
+    return HFMTemperatureController(prefix=f"{PREFIX.beamline_prefix}-EA-TCTRL-03:")

src/dodal/devices/beamlines/i10_1/high_field_magnet/temperature_controller.py

@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+from ophyd_async.core import (
+    StandardReadableFormat,
+    StrictEnum,
+)
+from ophyd_async.epics.core import epics_signal_r, epics_signal_rw
+
+from dodal.devices.temperture_controller import (
+    PID,
+    BaseHeater,
+    BaseTemperatureSensor,
+    TemperatureController,
+)
+
+
+class HeaterMode(StrictEnum):
+    MANUAL = "Manual"
+    AUTO = "Auto"
+
+
+class HighFieldMagnetTemperatureSensor(BaseTemperatureSensor):
+    def __init__(
+        self,
+        prefix: str,
+        suffix: str = "STEMP",
+        config_suffixes: list[str] | None = None,
+        name: str = "",
+    ):
+        config_suffixes = config_suffixes or ["2", "3"]
+
+        with self.add_children_as_readables(StandardReadableFormat.HINTED_SIGNAL):
+            self.temperature = epics_signal_r(float, prefix + suffix)
+
+        with self.add_children_as_readables(StandardReadableFormat.CONFIG_SIGNAL):
+            for sfx in config_suffixes:
+                signal_name = f"config_{sfx}"
+                if hasattr(self, signal_name):
+                    raise AttributeError(
+                        f"Cannot add configuration signal '{signal_name}': attribute already exists."
+                    )
+                signal = epics_signal_r(float, prefix + suffix + sfx)
+                setattr(self, signal_name, signal)
+        super().__init__(name=name)
+
+
+class HighFieldMagnetHeater(BaseHeater):
+    def __init__(self, prefix: str, name: str = ""):
+        with self.add_children_as_readables(StandardReadableFormat.CONFIG_SIGNAL):
+            self.mode = epics_signal_rw(
+                HeaterMode,
+                read_pv=prefix + "ACTIVITY",
+                write_pv=prefix + "ACTIVITY:SET",
+            )
+            self.setpoint = epics_signal_rw(float, prefix + "MANV:SET")
+        with self.add_children_as_readables(StandardReadableFormat.HINTED_SIGNAL):
+            self.output = epics_signal_r(float, prefix + "HEATERP")
+
+        super().__init__(name=name)
+
+
+class HFMTemperatureController(TemperatureController):
+    def __init__(
+        self,
+        prefix: str,
+        suffix: str = "TTEMP:SET",
+        config_suffixes: list[str] | None = None,
+        name: str = "",
+    ):
+        sensor = HighFieldMagnetTemperatureSensor(
+            prefix=prefix, config_suffixes=config_suffixes
+        )
+        heater = HighFieldMagnetHeater(prefix=prefix)
+        pid = PID(prefix=prefix)
+        super().__init__(
+            prefix=prefix,
+            suffix=suffix,
+            sensor=sensor,
+            heater=heater,
+            pid=pid,
+            name=name,
+        )

src/dodal/devices/temperture_controller/__init__.py

@@ -1,3 +1,17 @@
+from .base_temperature_controller import (
+    PID,
+    BaseHeater,
+    BaseTemperatureSensor,
+    TemperatureController,
+)
 from .lakeshore.lakeshore import Lakeshore, Lakeshore336, Lakeshore340
 
-__all__ = ["Lakeshore336", "Lakeshore340", "Lakeshore"]
+__all__ = [
+    "Lakeshore336",
+    "Lakeshore340",
+    "Lakeshore",
+    "BaseHeater",
+    "BaseTemperatureSensor",
+    "PID",
+    "TemperatureController",
+]

src/dodal/devices/temperture_controller/base_temperature_controller.py

@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from functools import cached_property
+from typing import Generic, TypeVar
+
+from ophyd_async.core import (
+    SignalR,
+    SignalRW,
+    StandardMovable,
+    StandardReadable,
+    StandardReadableFormat,
+    soft_signal_rw,
+)
+from ophyd_async.epics.core import epics_signal_rw
+
+from dodal.devices.movable import MovableWithToleranceLogic
+
+
+@dataclass
+class TemperatureMovableLogic(MovableWithToleranceLogic):
+    async def stop(self):
+        current_val = await self.readback.get_value()
+        await self.setpoint.set(current_val)
+
+
+class PID(StandardReadable):
+    def __init__(
+        self,
+        prefix: str,
+        suffix_p: str = "P",
+        suffix_i: str = "I",
+        suffix_d: str = "D",
+        name: str = "",
+    ):
+        with self.add_children_as_readables():
+            self.p = epics_signal_rw(float, prefix + suffix_p)
+            self.i = epics_signal_rw(float, prefix + suffix_i)
+            self.d = epics_signal_rw(float, prefix + suffix_d)
+        super().__init__(name=name)
+
+
+class BaseHeater(StandardReadable):
+    setpoint: SignalRW[float]
+    output: SignalR[float]
+
+
+class BaseTemperatureSensor(StandardReadable):
+    temperature: SignalR[float]
+
+
+SensorT = TypeVar("SensorT", bound=BaseTemperatureSensor)
+HeaterT = TypeVar("HeaterT", bound=BaseHeater)
+
+
+class TemperatureController(
+    StandardReadable, StandardMovable, Generic[SensorT, HeaterT]
+):
+    def __init__(
+        self,
+        prefix: str,
+        suffix: str,
+        sensor: SensorT,
+        heater: HeaterT,
+        pid: PID,
+        name: str = "",
+    ):
+
+        self.pid = pid
+        self.heater = heater
+        with self.add_children_as_readables():
+            self.sensor = sensor
+
+        with self.add_children_as_readables(StandardReadableFormat.CONFIG_SIGNAL):
+            self.tolerance = soft_signal_rw(float, initial_value=0.1)
+            self.user_setpoint = epics_signal_rw(float, prefix + suffix)
+
+        super().__init__(name=name)
+
+    @cached_property
+    def movable_logic(self) -> TemperatureMovableLogic:
+        return TemperatureMovableLogic(
+            setpoint=self.user_setpoint,
+            readback=self.sensor.temperature,
+            tolerance=self.tolerance,
+        )

tests/devices/beamlines/i10_1/high_field_magnet/test_temperature_controller.py

@@ -0,0 +1,58 @@
+import asyncio
+
+import pytest
+from ophyd_async.core import init_devices
+from ophyd_async.testing import (
+    assert_configuration,
+    assert_reading,
+    partial_reading,
+)
+
+from dodal.devices.beamlines.i10_1.high_field_magnet.temperature_controller import (
+    HFMTemperatureController,
+)
+
+
+@pytest.fixture
+def hfm_temp_controller() -> HFMTemperatureController:
+    with init_devices(mock=True):
+        hfm_temp_controller = HFMTemperatureController(
+            prefix="I10J-MAGNET-01:TEMP:",
+            suffix="TTEMP:SET",
+        )
+    return hfm_temp_controller
+
+
+async def test_temperature_controller_readback(
+    hfm_temp_controller: HFMTemperatureController,
+):
+    await hfm_temp_controller.set(1.0)
+
+    await assert_reading(
+        hfm_temp_controller,
+        {
+            "hfm_temp_controller": partial_reading(1.0),
+        },
+    )
+    await assert_configuration(
+        hfm_temp_controller,
+        {
+            "hfm_temp_controller-sensor-config_2": partial_reading(0.0),
+            "hfm_temp_controller-sensor-config_3": partial_reading(0.0),
+            "hfm_temp_controller-tolerance": partial_reading(0.1),
+            "hfm_temp_controller-user_setpoint": partial_reading(1.0),
+        },
+    )
+    await asyncio.gather(
+        hfm_temp_controller.pid.p.set(1.0),
+        hfm_temp_controller.pid.i.set(2.0),
+        hfm_temp_controller.pid.d.set(3.0),
+    )
+    await assert_reading(
+        hfm_temp_controller.pid,
+        {
+            "hfm_temp_controller-pid-p": partial_reading(1.0),
+            "hfm_temp_controller-pid-i": partial_reading(2.0),
+            "hfm_temp_controller-pid-d": partial_reading(3.0),
+        },
+    )

tests/devices/temperature_controller/test_base_temperature_controller.py

@@ -0,0 +1,116 @@
+import asyncio
+
+import pytest
+from ophyd_async.core import (
+    StandardReadableFormat,
+    init_devices,
+    set_mock_value,
+)
+from ophyd_async.epics.core import epics_signal_r, epics_signal_rw
+from ophyd_async.testing import assert_configuration, assert_reading, partial_reading
+
+from dodal.devices.temperture_controller import (
+    PID,
+    BaseHeater,
+    BaseTemperatureSensor,
+    TemperatureController,
+)
+
+
+class MinimalMockSensor(BaseTemperatureSensor):
+    def __init__(self, name: str = ""):
+        with self.add_children_as_readables(StandardReadableFormat.HINTED_SIGNAL):
+            self.temperature = epics_signal_r(float, "prefix + suffix)")
+        super().__init__(name=name)
+
+
+class MockHeater(BaseHeater):
+    def __init__(self, name: str = ""):
+        self.setpoint = epics_signal_rw(float, "prefix + suffix:SET")
+        self.output = epics_signal_r(float, "prefix + suffix")
+        super().__init__(name=name)
+
+
+class MockTemperatureController(TemperatureController):
+    def __init__(self, name: str = ""):
+        sensor = MinimalMockSensor()
+        heater = MockHeater()
+        pid = PID(prefix="TEST-PID:")
+        super().__init__(
+            prefix="TEST-CTRL:",
+            suffix="SET",
+            sensor=sensor,
+            heater=heater,
+            pid=pid,
+            name=name,
+        )
+
+
+@pytest.fixture
+def mock_controller() -> MockTemperatureController:
+    with init_devices(mock=True):
+        mock_controller = MockTemperatureController()
+    return mock_controller
+
+
+async def test_temperature_movable(mock_controller: MockTemperatureController):
+
+    set_mock_value(mock_controller.tolerance, 0.5)
+    set_mock_value(mock_controller.sensor.temperature, 10.0)
+
+    status = mock_controller.set(20.0)
+    assert not status.done
+    set_mock_value(mock_controller.sensor.temperature, 19.4)
+
+    assert not status.done
+    set_mock_value(mock_controller.sensor.temperature, 19.8)
+    await status
+    assert status.done
+    assert status.success
+
+
+async def test_temperature_controller_readback(
+    mock_controller: MockTemperatureController,
+):
+    await mock_controller.set(1.0)
+    await assert_reading(
+        mock_controller,
+        {
+            "mock_controller": partial_reading(1.0),
+        },
+    )
+    await assert_configuration(
+        mock_controller,
+        {
+            "mock_controller-tolerance": partial_reading(0.1),
+            "mock_controller-user_setpoint": partial_reading(1.0),
+        },
+    )
+    await asyncio.gather(
+        mock_controller.pid.p.set(1.0),
+        mock_controller.pid.i.set(2.0),
+        mock_controller.pid.d.set(3.0),
+    )
+    await assert_reading(
+        mock_controller.pid,
+        {
+            "mock_controller-pid-p": partial_reading(1.0),
+            "mock_controller-pid-i": partial_reading(2.0),
+            "mock_controller-pid-d": partial_reading(3.0),
+        },
+    )
+
+
+async def test_temperature_controller_stop(
+    mock_controller: MockTemperatureController,
+):
+
+    set_mock_value(mock_controller.user_setpoint, 1.0)
+    set_mock_value(mock_controller.sensor.temperature, 0.5)
+    await mock_controller.stop()
+    await assert_reading(
+        mock_controller,
+        {
+            "mock_controller": partial_reading(0.5),
+        },
+    )