Add the Eurotherm (eurotherm), copied from nslsii#264

src/pdftools/eurotherm3k.py

@@ -0,0 +1,148 @@
+# ruff: noqa: D101  # copied as is; nslsii does not require class docstrings
+"""Eurotherm 3000-series temperature controller (the PDF ``eurotherm3504``).
+
+Copied from nslsii#264 (26efeef), reformatted to pdftools' line length only.
+When that PR merges, import it from nslsii.ophyd_async.devices and delete this file.
+"""
+
+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")]

src/pdftools/profiles/collection.py

@@ -22,6 +22,7 @@
 
 import pdftools
 from pdftools.detectors import XSPDetector
+from pdftools.eurotherm3k import Eurotherm3k
 from pdftools.motors import Mirror, OCMTable, OpticsTableADC, SideBounceMono, Slits
 from pdftools.shutter import FastShutter, FilterBank
 from pdftools.utils import (
@@ -161,6 +162,10 @@
     # Filter bank: the old profile's fb_two_button_shutters (fb2), now just fb
     fb = FilterBank("XF:28ID1B-OP{Fltr:", name="fb")
 
+    # Eurotherm 3504, as in pdf-profile-collection 16-eurotherm_HAB;
+    # the old eurotherm3504 is loop1
+    eurotherm = Eurotherm3k("XF:28ID1-ES{ET:05}", name="eurotherm")
+
     # Optics, as in pdf-profile-collection 12-optics
     OCM_slits = Slits("XF:28ID1B-OP{Slt:2-Ax:", name="OCM_slits")
     BDM_slits = Slits("XF:28ID1A-OP{Slt:1-Ax:", name="BDM_slits")

tests/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 pdftools.eurotherm3k 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}"

tests/test_eurotherm_pvs.py

@@ -0,0 +1,23 @@
+from bluesky.run_engine import RunEngine
+from ophyd_async.core import init_devices
+
+from pdftools.eurotherm3k import Eurotherm3k
+
+# Records listed in pdf-profile-collection 16-eurotherm_HAB.py for XF:28ID1-ES{ET:05}
+OLD_PROFILE_LOOP1 = {
+    "setpoint": "LOOP1:SP",
+    "readback": "LOOP1:PV:RBV",
+    "working_setpoint": "LOOP1:WSP:RBV",
+    "ramp_rate": "LOOP1:RR",
+    "output": "LOOP1:O",
+    "autotune": "LOOP1:AUTOTUNE",
+}
+
+
+def test_pdf_eurotherm_pvs_match_the_old_profile(RE: RunEngine):
+    with init_devices(mock=True):
+        eurotherm = Eurotherm3k("XF:28ID1-ES{ET:05}", name="eurotherm")
+    for attr, suffix in OLD_PROFILE_LOOP1.items():
+        source = getattr(eurotherm.loop1, attr).source
+        assert source.startswith(f"mock+ca://XF:28ID1-ES{{ET:05}}{suffix}"), source
+    assert eurotherm.disable.source == "mock+ca://XF:28ID1-ES{ET:05}DISABLE"

tests/test_utils.py

@@ -153,6 +153,7 @@ def test_profile_boots_in_mock_mode(tmp_path: Path):
         "pe2c",
         "fs",
         "fb",
+        "eurotherm",
         "OCM_slits",
         "BDM_slits",
         "WB_slits",