Add Eurotherm3k ophyd-async device (HXM-1305)

nslsii/ophyd_async/devices/__init__.py

@@ -1,11 +1,18 @@
-from rbd9103 import (
+from .rbd9103 import (
     RBD9103,
     RBD9103Range,
     RBD9103Input,
     RBD9103SamplingMode,
     RBD9103InRangeState,
     RBD9103Filter,
 )
+from .eurotherm3k import (
+    Eurotherm3k,
+    Eurotherm3kLoop,
+    Eurotherm3kControlMode,
+    Eurotherm3kOnOff,
+    Eurotherm3kRampRateUnit,
+)
 
 __all__ = [
     "RBD9103",
@@ -14,4 +21,9 @@
     "RBD9103SamplingMode",
     "RBD9103InRangeState",
     "RBD9103Filter",
-]
+    "Eurotherm3k",
+    "Eurotherm3kLoop",
+    "Eurotherm3kControlMode",
+    "Eurotherm3kOnOff",
+    "Eurotherm3kRampRateUnit",
+]

nslsii/ophyd_async/devices/eurotherm3k.py

@@ -0,0 +1,125 @@
+from dataclasses import dataclass
+from functools import cached_property
+from typing import Annotated as A
+
+from ophyd_async.core import (
+    MovableLogic,
+    SignalR,
+    SignalRW,
+    StandardMovable,
+    StandardReadable,
+    StrictEnum,
+    TimeoutCalculator,
+    set_and_wait_for_other_value,
+    wait_for_value,
+)
+from ophyd_async.core import StandardReadableFormat as Format
+from ophyd_async.epics.core import EpicsDevice, PvSuffix
+
+
+class Eurotherm3kControlMode(StrictEnum):
+    AUTOMATIC = "Automatic"
+    MANUAL = "Manual"
+
+
+class Eurotherm3kOnOff(StrictEnum):
+    OFF = "Off"
+    ON = "On"
+
+
+class Eurotherm3kRampRateUnit(StrictEnum):
+    PER_SECOND = "C/s"
+    PER_MINUTE = "C/min"
+
+
+@dataclass
+class _Eurotherm3kLoopLogic(MovableLogic[float]):
+    loop: "Eurotherm3kLoop"
+
+    async def calculate_timeout(self, old_position: float, new_position: float) -> float:
+        return self.loop.move_timeout
+
+    async def move(self, new_position: float, timeout: TimeoutCalculator) -> None:
+        def in_band(value: float) -> bool:
+            return abs(value - new_position) <= self.loop.tolerance
+
+        def out_of_band(value: float) -> bool:
+            return not in_band(value)
+
+        await set_and_wait_for_other_value(
+            self.setpoint,
+            new_position,
+            self.readback,
+            in_band,
+            timeout=timeout(),
+        )
+        if self.loop.settle_time <= 0:
+            return
+
+        while True:
+            settle_time = self.loop.settle_time
+            if settle_time <= 0:
+                return
+            remaining_timeout = timeout()
+            hold_timeout = (
+                settle_time if remaining_timeout is None or remaining_timeout >= settle_time else remaining_timeout
+            )
+            try:
+                await wait_for_value(self.readback, out_of_band, timeout=hold_timeout)
+            except TimeoutError:
+                if remaining_timeout is None or remaining_timeout >= settle_time:
+                    return
+                raise
+            await wait_for_value(self.readback, in_band, timeout=timeout())
+
+
+class Eurotherm3kLoop(EpicsDevice, StandardReadable, StandardMovable[float]):
+    """One PID control loop of a Eurotherm 3000-series controller.
+
+    Behaves as a settable temperature axis: ``set(T)`` writes the setpoint and
+    completes once the readback has settled within ``tolerance`` of ``T``.
+    """
+
+    # Settle behaviour (override per instance, e.g. ``dev.loop1.tolerance = 0.5``)
+    tolerance = 1.0  # deg: |readback - setpoint| band counted as "there"
+    settle_time = 0.0  # s: must stay in band this long (0.0 = first touch)
+    move_timeout = 3600.0  # s: overall timeout for a move
+
+    setpoint: A[SignalRW[float], PvSuffix.rbv("SP", ":RBV")]
+    readback: A[SignalR[float], PvSuffix("PV:RBV"), Format.HINTED_SIGNAL]
+    working_setpoint: A[SignalR[float], PvSuffix("WSP:RBV")]
+    ramp_rate: A[SignalRW[float], PvSuffix.rbv("RR", ":RBV"), Format.CONFIG_SIGNAL]
+    ramp_rate_unit: A[SignalR[Eurotherm3kRampRateUnit], PvSuffix("RR:UNIT"), Format.CONFIG_SIGNAL]
+    p: A[SignalRW[float], PvSuffix.rbv("P", ":RBV"), Format.CONFIG_SIGNAL]
+    i: A[SignalRW[float], PvSuffix.rbv("I", ":RBV"), Format.CONFIG_SIGNAL]
+    d: A[SignalRW[float], PvSuffix.rbv("D", ":RBV"), Format.CONFIG_SIGNAL]
+    control_mode: A[
+        SignalRW[Eurotherm3kControlMode],
+        PvSuffix.rbv("MAN", ":RBV"),
+        Format.CONFIG_SIGNAL,
+    ]
+    autotune: A[SignalRW[Eurotherm3kOnOff], PvSuffix.rbv("AUTOTUNE", ":RBV")]
+    output: A[SignalRW[float], PvSuffix.rbv("O", ":RBV")]
+    output_high: A[SignalRW[float], PvSuffix.rbv("OUTPHI", ":RBV"), Format.CONFIG_SIGNAL]
+    output_low: A[SignalRW[float], PvSuffix.rbv("OUTPLO", ":RBV"), Format.CONFIG_SIGNAL]
+    loop_break_time: A[SignalRW[float], PvSuffix.rbv("LBT", ":RBV"), Format.CONFIG_SIGNAL]
+
+    @cached_property
+    def movable_logic(self) -> MovableLogic[float]:
+        return _Eurotherm3kLoopLogic(setpoint=self.setpoint, readback=self.readback, loop=self)
+
+
+class Eurotherm3k(StandardReadable, EpicsDevice):
+    """Eurotherm 3000-series controller (nsls2.ioc_deploy eurotherm3k role)."""
+
+    loop1: A[Eurotherm3kLoop, PvSuffix("LOOP1:"), Format.CHILD]
+    loop2: A[Eurotherm3kLoop, PvSuffix("LOOP2:"), Format.CHILD]
+
+    # --- Global (controller-wide) ---
+    disable: A[SignalRW[str], PvSuffix("DISABLE"), Format.CONFIG_SIGNAL]
+    programmer_number: A[SignalRW[int], PvSuffix.rbv("PROGNUM", ":RBV"), Format.CONFIG_SIGNAL]
+    programmer_run: A[SignalRW[bool], PvSuffix.rbv("PROGRUN", ":RBV")]
+    programmer_reset: A[SignalRW[bool], PvSuffix("PROGRESET")]
+    programmer_status: A[SignalR[str], PvSuffix("PROGSTAT:RBV")]
+    recipe_select: A[SignalRW[str], PvSuffix.rbv("RECSEL", ":RBV"), Format.CONFIG_SIGNAL]
+    recipe_status: A[SignalR[str], PvSuffix("RECSTAT:RBV")]

nslsii/ophyd_async/devices/rbd9103.py

@@ -6,7 +6,7 @@
     AsyncStatus,
 )
 from ophyd_async.core import StandardReadableFormat as Format
-from ophyd_async.epics.signal import PvSuffix, EpicsDevice
+from ophyd_async.epics.core import PvSuffix, EpicsDevice
 from typing import Annotated as A
 
 

nslsii/tests/ophyd_async/test_eurotherm3k.py

@@ -0,0 +1,147 @@
+import asyncio
+import time
+
+import pytest
+from ophyd_async.core import (
+    LazyMock,
+    callback_on_mock_put,
+    get_mock_put,
+    init_devices,
+    set_mock_value,
+)
+
+from nslsii.ophyd_async.devices import (
+    Eurotherm3k,
+    Eurotherm3kControlMode,
+    Eurotherm3kLoop,
+)
+
+PREFIX = "XF:28ID1-ES{ET:05}"
+
+
+@pytest.mark.asyncio
+async def test_pvs_are_addressed():
+    async with init_devices(mock=True):
+        dev = Eurotherm3k(PREFIX, name="et")
+    # child loop signals compose under <prefix>LOOPn:
+    assert dev.loop1.setpoint.source == f"mock+ca://{PREFIX}LOOP1:SP:RBV"
+    assert dev.loop1.readback.source == f"mock+ca://{PREFIX}LOOP1:PV:RBV"
+    assert dev.loop2.setpoint.source == f"mock+ca://{PREFIX}LOOP2:SP:RBV"
+    # global signal sits directly under the prefix
+    assert dev.programmer_status.source == f"mock+ca://{PREFIX}PROGSTAT:RBV"
+
+
+@pytest.mark.asyncio
+async def test_format_tags_route_signals():
+    async with init_devices(mock=True):
+        dev = Eurotherm3k(PREFIX, name="et")
+    reading = await dev.read()
+    config = await dev.read_configuration()
+    # HINTED readbacks (merged from both loop CHILDren) appear in read()
+    assert "et-loop1" in reading
+    assert "et-loop2" in reading
+    # CONFIG signals appear in read_configuration()
+    assert "et-loop1-ramp_rate" in config
+
+
+@pytest.mark.asyncio
+async def test_control_mode_enum_roundtrip():
+    async with init_devices(mock=True):
+        dev = Eurotherm3k(PREFIX, name="et")
+    await dev.loop1.control_mode.set(Eurotherm3kControlMode.MANUAL)
+    assert await dev.loop1.control_mode.get_value() is Eurotherm3kControlMode.MANUAL
+
+
+@pytest.mark.asyncio
+async def test_set_commands_setpoint_and_settles():
+    async with init_devices(mock=True):
+        dev = Eurotherm3k(PREFIX, name="et")
+    await dev.loop1.set(300.0)
+    get_mock_put(dev.loop1.setpoint).assert_called_once_with(300.0)
+    location = await dev.loop1.locate()
+    assert location["setpoint"] == 300.0
+    assert location["readback"] == 300.0
+
+
+@pytest.mark.asyncio
+async def test_set_holds_in_band_for_settle_time():
+    """With settle_time > 0, set() waits the hold time even if the readback is
+    already in band and stops updating -- and must not block until move_timeout."""
+    loop = Eurotherm3kLoop(f"{PREFIX}LOOP1:", name="et-loop1")
+    await loop.connect(mock=LazyMock())
+    loop.tolerance = 1.0
+    loop.settle_time = 0.2
+    loop.move_timeout = 5.0
+    # already in band and stable (no further updates arrive)
+    set_mock_value(loop.readback, 300.0)
+    start = time.monotonic()
+    await loop.set(300.0)
+    elapsed = time.monotonic() - start
+    # waited ~settle_time, and crucially did NOT hang until move_timeout
+    assert 0.2 <= elapsed < 5.0
+    get_mock_put(loop.setpoint).assert_called_once_with(300.0)
+
+
+@pytest.mark.asyncio
+async def test_set_restarts_settle_time_after_out_of_band_excursion():
+    loop = Eurotherm3kLoop(f"{PREFIX}LOOP1:", name="et-loop1")
+    await loop.connect(mock=LazyMock())
+    loop.tolerance = 1.0
+    loop.settle_time = 0.2
+    loop.move_timeout = 5.0
+    set_mock_value(loop.readback, 300.0)
+
+    async def make_excursion() -> None:
+        await asyncio.sleep(loop.settle_time / 2)
+        set_mock_value(loop.readback, 0.0)
+        await asyncio.sleep(0)
+        set_mock_value(loop.readback, 300.0)
+
+    excursion_task: asyncio.Task[None] | None = None
+
+    def start_excursion(value: float) -> None:
+        nonlocal excursion_task
+        excursion_task = asyncio.create_task(make_excursion())
+
+    callback_on_mock_put(loop.setpoint, start_excursion)
+    start = time.monotonic()
+    await loop.set(300.0)
+    elapsed = time.monotonic() - start
+    assert excursion_task is not None
+    await excursion_task
+    assert 0.3 <= elapsed < 5.0
+    get_mock_put(loop.setpoint).assert_called_once_with(300.0)
+
+
+@pytest.mark.asyncio
+async def test_set_times_out_if_never_in_band():
+    loop = Eurotherm3kLoop(f"{PREFIX}LOOP1:", name="et-loop1")
+    await loop.connect(mock=LazyMock())
+    loop.tolerance = 0.5
+    loop.move_timeout = 0.3  # keep the test fast
+    set_mock_value(loop.readback, 0.0)  # stays far from 300
+    with pytest.raises(TimeoutError):
+        await loop.set(300.0)
+
+
+# Legacy PDF eurotherm3k (pdf-profile-collection/startup/16-eurotherm_HAB.py):
+# each component -> the ophyd-async signal that must address the same PV.
+_LEGACY_PV_MAP = {
+    "setpoint": ("loop1", "setpoint", "LOOP1:SP:RBV"),
+    "readback": ("loop1", "readback", "LOOP1:PV:RBV"),
+    "working": ("loop1", "working_setpoint", "LOOP1:WSP:RBV"),
+    "output": ("loop1", "output", "LOOP1:O:RBV"),
+    "ramprate": ("loop1", "ramp_rate", "LOOP1:RR:RBV"),
+    "manual_mode": ("loop1", "control_mode", "LOOP1:MAN:RBV"),
+    "autotune": ("loop1", "autotune", "LOOP1:AUTOTUNE:RBV"),
+}
+
+
+@pytest.mark.asyncio
+async def test_legacy_pv_surface_equivalence():
+    """Every PV of the legacy PDF eurotherm3k is addressed identically here."""
+    async with init_devices(mock=True):
+        dev = Eurotherm3k(PREFIX, name="et")
+    for _legacy, (loop, attr, suffix) in _LEGACY_PV_MAP.items():
+        signal = getattr(getattr(dev, loop), attr)
+        assert signal.source == f"mock+ca://{PREFIX}{suffix}"