From 672675775765ac4e72dc888ea863e07ab025a077 Mon Sep 17 00:00:00 2001 From: thomashopkins32 Date: Mon, 12 Oct 2026 16:59:09 -0400 Subject: [PATCH 1/3] Add pumps as ophyd-async devices; Add pump bluesky plans as separate module Assisted-by: oh-my-pi:gpt-6-sol --- pyproject.toml | 4 +- src/xpd_tools/pump_plans.py | 309 ++++++++++++++++++++++++++++++++++++ src/xpd_tools/pumps.py | 283 +++++++++++++++++++++++++++++++++ tests/test_pump_plans.py | 232 +++++++++++++++++++++++++++ tests/test_pumps.py | 62 ++++++++ uv.lock | 16 +- 6 files changed, 898 insertions(+), 8 deletions(-) create mode 100644 src/xpd_tools/pump_plans.py create mode 100644 src/xpd_tools/pumps.py create mode 100644 tests/test_pump_plans.py create mode 100644 tests/test_pumps.py diff --git a/pyproject.toml b/pyproject.toml index e5eec1b..4108e7e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,7 +32,7 @@ dependencies = [ "bluesky-tiled-plugins>=2.0.6", "ipython>=9.15.0", "nslsii>=0.11.9", - "ophyd-async[ca,pva] @ git+https://github.com/bluesky/ophyd-async@main", + "ophyd-async[ca,pva]==0.21.3", "redis-json-dict>=0.2.2", "rich>=15.0.0", "tiled[client]>=0.2.12", @@ -41,7 +41,7 @@ dependencies = [ [dependency-groups] dev = [ - "ophyd-async[ca, pva, sim] @ git+https://github.com/bluesky/ophyd-async@main", + "ophyd-async[ca,pva,sim]==0.21.3", "inflection", "import-linter", "numpydoc", diff --git a/src/xpd_tools/pump_plans.py b/src/xpd_tools/pump_plans.py new file mode 100644 index 0000000..ba4fc5a --- /dev/null +++ b/src/xpd_tools/pump_plans.py @@ -0,0 +1,309 @@ +"""Bluesky plans for syringe-pump configuration, operation and safe cleanup.""" + +from __future__ import annotations + +import math +import warnings +from collections.abc import Callable, Generator, Sequence + +from bluesky import plan_stubs as bps +from bluesky import preprocessors as bpp +from ophyd_async.core import wait_for_value + +from .pumps import ( + DDSSyringeAxis, + UltraSyringePump, + rate_limits, + rate_unit_converter, + validate_pump_settings, + vol_unit_converter, +) + +Pump = DDSSyringeAxis | UltraSyringePump + + +def _rate(value: float | str, unit: str) -> float: + if isinstance(value, str): + number, unit = value.split() + value = float(number) + return float(value) * rate_unit_converter(unit, "ul/min") + + +def _target(value: float | str) -> float: + if isinstance(value, str): + number, unit = value.split() + return float(number) * vol_unit_converter(unit, "ml") + return float(value) + + +def _count(pumps: Sequence[Pump], **parameters: Sequence) -> None: + for name, values in parameters.items(): + if len(values) != len(pumps): + raise ValueError(f"{name} has {len(values)} entries for {len(pumps)} pumps") + + +def _configure( + pumps: Sequence[Pump], + syringe_sizes_ml: Sequence[float], + target_volumes: Sequence[float | str], + rates: Sequence[float | str], + materials: Sequence[str], + set_targets: Sequence[bool], + direction: str, + rate_unit: str, +) -> Generator: + _count( + pumps, + syringe_sizes_ml=syringe_sizes_ml, + target_volumes=target_volumes, + rates=rates, + materials=materials, + set_targets=set_targets, + ) + normalized = [] + for pump, size, target, rate, material, set_target in zip( + pumps, + syringe_sizes_ml, + target_volumes, + rates, + materials, + set_targets, + strict=True, + ): + rate_ul_min = _rate(rate, rate_unit) + target_ml = _target(target) if set_target else None + validate_pump_settings(pump, material, size, target_ml, rate_ul_min) + normalized.append((pump, size, target_ml, rate_ul_min, material)) + + # Range refresh/read is performed for *every* requested DDS axis before any + # target or rate write; a wrong syringe on one axis cannot partly configure another. + for pump, size, _, rate, material in normalized: + if not rate: + continue + if isinstance(pump, DDSSyringeAxis): + yield from bps.trigger(pump.set_infuse_range, wait=True) + yield from bps.trigger(pump.set_withdraw_range, wait=True) + selected = yield from bps.rd(pump.read_min_infuse) + selected_unit = yield from bps.rd(pump.read_min_infuse_unit) + expected, _ = rate_limits(pump, material, size) + if not math.isclose( + selected * rate_unit_converter(selected_unit, "nl/min"), + expected, + rel_tol=1e-3, + abs_tol=0, + ): + raise ValueError( + f"{pump.name}: selected DDS syringe does not match " + f"{size} mL {material}" + ) + lower = yield from bps.rd(getattr(pump, f"read_min_{direction}")) + lower_unit = yield from bps.rd(getattr(pump, f"read_min_{direction}_unit")) + upper = yield from bps.rd(getattr(pump, f"read_max_{direction}")) + upper_unit = yield from bps.rd(getattr(pump, f"read_max_{direction}_unit")) + lower *= rate_unit_converter(lower_unit, "ul/min") + upper *= rate_unit_converter(upper_unit, "ul/min") + if not ( + math.isfinite(lower) + and math.isfinite(upper) + and 0 < lower <= rate <= upper + ): + raise ValueError(f"{pump.name}: {direction} rate outside IOC limits") + else: + warnings.warn( + f"{pump.name}: confirm {size} mL {material} syringe on Ultra panel", + UserWarning, + stacklevel=2, + ) + + for pump, _, target, rate, _ in normalized: + if not rate: + continue + if target is not None: + yield from bps.mv(pump.target_vol_unit, "ml") + yield from bps.mv(pump.target_vol, target) + yield from bps.mv(getattr(pump, f"{direction}_rate_unit"), "ul/min") + yield from bps.mv(getattr(pump, f"{direction}_rate"), rate) + + +def configure_infusion( + pumps, + syringe_sizes_ml, + target_volumes, + rates, + materials, + set_targets, + *, + rate_unit="ul/min", +): + """Validate every syringe and IOC limit before setting infusion parameters.""" + yield from _configure( + pumps, + syringe_sizes_ml, + target_volumes, + rates, + materials, + set_targets, + "infuse", + rate_unit, + ) + + +def configure_withdrawal( + pumps, + syringe_sizes_ml, + target_volumes, + rates, + materials, + set_targets, + *, + rate_unit="ul/min", +): + """Validate every syringe and IOC limit before setting withdrawal parameters.""" + yield from _configure( + pumps, + syringe_sizes_ml, + target_volumes, + rates, + materials, + set_targets, + "withdraw", + rate_unit, + ) + + +def _start(pumps, rates, direction, rate_unit): + _count(pumps, rates=rates) + normalized = [_rate(rate, rate_unit) for rate in rates] + if any(not math.isfinite(rate) or rate < 0 for rate in normalized): + raise ValueError("Start rates must be finite and nonnegative") + for pump, rate in zip(pumps, normalized, strict=True): + if rate: + yield from bps.trigger(getattr(pump, f"pump_{direction}"), wait=True) + expected = "Infusing" if direction == "infuse" else "Withdrawing" + yield from bps.wait_for( + [ + lambda pump=pump, expected=expected: wait_for_value( + pump.status, expected, timeout=10 + ) + ] + ) + + +def start_infusion(pumps, rates, *, rate_unit="ul/min"): + """Start pumps with nonzero rates, waiting for IOC status confirmation.""" + yield from _start(pumps, rates, "infuse", rate_unit) + + +def start_withdrawal(pumps, rates, *, rate_unit="ul/min"): + """Start withdrawal for pumps with nonzero rates.""" + yield from _start(pumps, rates, "withdraw", rate_unit) + + +def stop_pumps(pumps): + """Attempt STOP and status confirmation for every pump, even on failure.""" + failures = [] + for pump in pumps: + try: + yield from bps.trigger(pump.pump_stop, wait=True) + yield from bps.wait_for( + [ + lambda pump=pump: wait_for_value( + pump.status, + lambda status: status not in ("Infusing", "Withdrawing"), + timeout=10, + ) + ] + ) + except Exception as exc: + failures.append(exc) + if failures: + raise ExceptionGroup("Pump stop failures", failures) + + +def enable_pumps(pumps, *, update=".5 second", clear=False): + """Enable each DDS controller once; leave counter clearing opt-in.""" + controllers = set() + for pump in pumps: + if isinstance(pump, DDSSyringeAxis): + controller = pump.controller + if controller in controllers: + continue + controllers.add(controller) + yield from bps.mv(controller.communication, "Enabled") + else: + yield from bps.mv(pump.communication, "Enabled") + yield from bps.mv(pump.update_pump, update) + if clear: + for pump in pumps: + yield from bps.trigger(pump.clear_infused, wait=True) + yield from bps.trigger(pump.clear_withdrawn, wait=True) + + +def read_pump_metadata(pumps): + """Return live readback metadata for run-start documents.""" + result = {"pumps": [], "infuse_rate": [], "infuse_rate_unit": [], "pump_status": []} + for pump in pumps: + result["pumps"].append(pump.name) + result["infuse_rate"].append((yield from bps.rd(pump.read_infuse_rate))) + result["infuse_rate_unit"].append( + (yield from bps.rd(pump.read_infuse_rate_unit)) + ) + result["pump_status"].append((yield from bps.rd(pump.status))) + return result + + +def wait_for_equilibrium(mixer_pump_list, *, ratio=1, tubing_ID_mm=1.016): # noqa: N803 + """Wait cumulative residence times for successive tubing/flow segments.""" + if ( + not mixer_pump_list + or not math.isfinite(ratio) + or ratio < 0 + or tubing_ID_mm <= 0 + ): + raise ValueError("Nonempty mixer and nonnegative residence geometry required") + duration = 0.0 + lengths = {"mm": 1, "cm": 10, "m": 1000} + for length_spec, *pumps in mixer_pump_list: + length, unit = length_spec.split() + length_mm = float(length) * lengths[unit] + if not pumps or not math.isfinite(length_mm) or length_mm < 0: + raise ValueError("Each mixer segment requires pumps and valid length") + flow = 0.0 + for pump in pumps: + if (yield from bps.rd(pump.status)) == "Infusing": + rate = yield from bps.rd(pump.read_infuse_rate) + unit = yield from bps.rd(pump.read_infuse_rate_unit) + flow += rate * rate_unit_converter(unit, "ul/min") + if not math.isfinite(flow) or flow <= 0: + raise ValueError("Mixer segment has no positive infusing flow") + duration += 60 * math.pi * (tubing_ID_mm / 2) ** 2 * length_mm / flow + yield from bps.sleep(duration * ratio) + return duration * ratio + + +def pump_session( + body_factory: Callable[[], Generator], *, setup: Generator, pumps: list[Pump] +): + """Finalize every potentially started pump, including on setup failure.""" + failure = None + + def body(): + nonlocal failure + try: + yield from setup + return (yield from body_factory()) + except Exception as exc: + failure = exc + raise + + def cleanup(): + try: + yield from stop_pumps(pumps) + except Exception as exc: + if failure is not None: + raise ExceptionGroup( + "Pump session and cleanup failures", [failure, exc] + ) from None + raise + + return (yield from bpp.finalize_wrapper(body(), cleanup())) diff --git a/src/xpd_tools/pumps.py b/src/xpd_tools/pumps.py new file mode 100644 index 0000000..1cb14d0 --- /dev/null +++ b/src/xpd_tools/pumps.py @@ -0,0 +1,283 @@ +"""EPICS syringe-pump interfaces and syringe/rate policy.""" + +from __future__ import annotations + +import math +import weakref +from collections.abc import Mapping +from types import MappingProxyType +from typing import Annotated as A + +from ophyd_async.core import SignalR, SignalRW, StandardReadable, TriggerableCommand +from ophyd_async.core import StandardReadableFormat as Format +from ophyd_async.epics.core import EpicsDevice, PvSuffix + + +class DDSSyringeController(EpicsDevice): + """Shared DDS controller; an axis must not own or disable its sibling.""" + + communication: A[SignalRW[str], PvSuffix("}DISABLE")] + update_pump: A[SignalRW[str], PvSuffix("}UPDATE.SCAN")] + set_condition: A[SignalRW[str], PvSuffix("}CONDITION")] + show_condition: A[SignalR[str], PvSuffix("}CONDITION:RBV")] + + +class DDSSyringeAxis(StandardReadable, EpicsDevice): + """One DDS axis; controller ownership remains with the startup profile.""" + + status: A[SignalR[str], PvSuffix("STATUS"), Format.UNCACHED_SIGNAL] + pump_infuse: A[TriggerableCommand, PvSuffix("IRUN")] + pump_withdraw: A[TriggerableCommand, PvSuffix("WRUN")] + pump_stop: A[TriggerableCommand, PvSuffix("STOP")] + target_vol: A[SignalRW[float], PvSuffix("TVOLUME"), Format.CONFIG_SIGNAL] + target_vol_unit: A[SignalRW[str], PvSuffix("TVOLUME:EGU"), Format.CONFIG_SIGNAL] + read_target_vol: A[SignalR[float], PvSuffix("TVOLUME:RBV"), Format.UNCACHED_SIGNAL] + read_target_vol_unit: A[ + SignalR[str], PvSuffix("TVOLUME:RBV:EGU"), Format.UNCACHED_SIGNAL + ] + clear_infused: A[TriggerableCommand, PvSuffix("CIVOLUME")] + clear_withdrawn: A[TriggerableCommand, PvSuffix("CWVOLUME")] + read_infused: A[SignalR[float], PvSuffix("IVOLUME:RBV"), Format.UNCACHED_SIGNAL] + read_withdrawn: A[SignalR[float], PvSuffix("WVOLUME:RBV"), Format.UNCACHED_SIGNAL] + read_infused_unit: A[ + SignalR[str], PvSuffix("IVOLUME:RBV:EGU"), Format.UNCACHED_SIGNAL + ] + read_withdrawn_unit: A[ + SignalR[str], PvSuffix("WVOLUME:RBV:EGU"), Format.UNCACHED_SIGNAL + ] + infuse_rate: A[SignalRW[float], PvSuffix("IRATE"), Format.CONFIG_SIGNAL] + infuse_rate_unit: A[SignalRW[str], PvSuffix("IRATE:EGU"), Format.CONFIG_SIGNAL] + withdraw_rate: A[SignalRW[float], PvSuffix("WRATE"), Format.CONFIG_SIGNAL] + withdraw_rate_unit: A[SignalRW[str], PvSuffix("WRATE:EGU"), Format.CONFIG_SIGNAL] + read_infuse_rate: A[ + SignalR[float], PvSuffix("IRATE:RBV"), Format.HINTED_UNCACHED_SIGNAL + ] + read_infuse_rate_unit: A[ + SignalR[str], PvSuffix("IRATE:RBV:EGU"), Format.UNCACHED_SIGNAL + ] + read_withdraw_rate: A[SignalR[float], PvSuffix("WRATE:RBV"), Format.UNCACHED_SIGNAL] + read_withdraw_rate_unit: A[ + SignalR[str], PvSuffix("WRATE:RBV:EGU"), Format.UNCACHED_SIGNAL + ] + max_infuse: A[SignalRW[float], PvSuffix("IRATE:MAX"), Format.CONFIG_SIGNAL] + max_withdraw: A[SignalRW[float], PvSuffix("WRATE:MAX"), Format.CONFIG_SIGNAL] + min_infuse: A[SignalRW[float], PvSuffix("IRATE:MIN"), Format.CONFIG_SIGNAL] + min_withdraw: A[SignalRW[float], PvSuffix("WRATE:MIN"), Format.CONFIG_SIGNAL] + set_infuse_range: A[TriggerableCommand, PvSuffix("IRATE:LIM.PROC")] + set_withdraw_range: A[TriggerableCommand, PvSuffix("WRATE:LIM.PROC")] + read_max_infuse: A[ + SignalR[float], PvSuffix("IRATE:MAX:RBV"), Format.UNCACHED_SIGNAL + ] + read_max_infuse_unit: A[ + SignalR[str], PvSuffix("IRATE:MAX:RBV:EGU"), Format.UNCACHED_SIGNAL + ] + read_max_withdraw: A[ + SignalR[float], PvSuffix("WRATE:MAX:RBV"), Format.UNCACHED_SIGNAL + ] + read_max_withdraw_unit: A[ + SignalR[str], PvSuffix("WRATE:MAX:RBV:EGU"), Format.UNCACHED_SIGNAL + ] + read_min_infuse: A[ + SignalR[float], PvSuffix("IRATE:MIN:RBV"), Format.UNCACHED_SIGNAL + ] + read_min_infuse_unit: A[ + SignalR[str], PvSuffix("IRATE:MIN:RBV:EGU"), Format.UNCACHED_SIGNAL + ] + read_min_withdraw: A[ + SignalR[float], PvSuffix("WRATE:MIN:RBV"), Format.UNCACHED_SIGNAL + ] + read_min_withdraw_unit: A[ + SignalR[str], PvSuffix("WRATE:MIN:RBV:EGU"), Format.UNCACHED_SIGNAL + ] + LIM: A[SignalRW[str], PvSuffix("IRATE-LIM:ENABLE"), Format.CONFIG_SIGNAL] + target_time: A[SignalRW[float], PvSuffix("TTIME"), Format.CONFIG_SIGNAL] + read_target_time: A[SignalR[float], PvSuffix("TTIME:RBV"), Format.UNCACHED_SIGNAL] + force: A[SignalRW[float], PvSuffix("FORCE"), Format.CONFIG_SIGNAL] + read_force: A[SignalR[float], PvSuffix("FORCE:RBV"), Format.UNCACHED_SIGNAL] + read_force_unit: A[SignalR[str], PvSuffix("FORCE:RBV:EGU"), Format.UNCACHED_SIGNAL] + + def __init__(self, prefix: str, controller: DDSSyringeController, name: str = ""): + self._controller = weakref.ref(controller) + super().__init__(prefix, name=name) + + @property + def controller(self) -> DDSSyringeController: + """The separately connected, shared DDS controller.""" + controller = self._controller() + if controller is None: + raise RuntimeError("DDS controller has been released") + return controller + + +class UltraSyringePump(StandardReadable, EpicsDevice): + """Ultra syringe pump; no IOC syringe-size feedback is available.""" + + status: A[SignalR[str], PvSuffix("STATUS"), Format.UNCACHED_SIGNAL] + communication: A[SignalRW[str], PvSuffix("DISABLE"), Format.CONFIG_SIGNAL] + update_pump: A[SignalRW[str], PvSuffix("UPDATE.SCAN"), Format.CONFIG_SIGNAL] + pump_infuse: A[TriggerableCommand, PvSuffix("IRUN")] + pump_withdraw: A[TriggerableCommand, PvSuffix("WRUN")] + pump_stop: A[TriggerableCommand, PvSuffix("STOP")] + target_vol: A[SignalRW[float], PvSuffix("TVOLUME"), Format.CONFIG_SIGNAL] + target_vol_unit: A[SignalRW[str], PvSuffix("TVOLUMEUNITS"), Format.CONFIG_SIGNAL] + read_target_vol: A[SignalR[float], PvSuffix("TVOLUME:RBV"), Format.UNCACHED_SIGNAL] + read_target_vol_unit: A[ + SignalR[str], PvSuffix("TVOLUMEUNITS:RBV"), Format.UNCACHED_SIGNAL + ] + clear_infused: A[TriggerableCommand, PvSuffix("CLEARINFUSED")] + clear_withdrawn: A[TriggerableCommand, PvSuffix("CLEARWITHDRAWN")] + read_infused: A[SignalR[float], PvSuffix("IVOLUME:RBV"), Format.UNCACHED_SIGNAL] + read_withdrawn: A[SignalR[float], PvSuffix("WVOLUME:RBV"), Format.UNCACHED_SIGNAL] + read_infused_unit: A[ + SignalR[str], PvSuffix("IVOLUMEUNITS:RBV"), Format.UNCACHED_SIGNAL + ] + read_withdrawn_unit: A[ + SignalR[str], PvSuffix("WVOLUMEUNITS:RBV"), Format.UNCACHED_SIGNAL + ] + infuse_rate: A[SignalRW[float], PvSuffix("IRATE"), Format.CONFIG_SIGNAL] + infuse_rate_unit: A[SignalRW[str], PvSuffix("IRATEUNITS"), Format.CONFIG_SIGNAL] + withdraw_rate: A[SignalRW[float], PvSuffix("WRATE"), Format.CONFIG_SIGNAL] + withdraw_rate_unit: A[SignalRW[str], PvSuffix("WRATEUNITS"), Format.CONFIG_SIGNAL] + read_infuse_rate: A[ + SignalR[float], PvSuffix("IRATE:RBV"), Format.HINTED_UNCACHED_SIGNAL + ] + read_infuse_rate_unit: A[ + SignalR[str], PvSuffix("IRATEUNITS:RBV"), Format.UNCACHED_SIGNAL + ] + read_withdraw_rate: A[SignalR[float], PvSuffix("WRATE:RBV"), Format.UNCACHED_SIGNAL] + read_withdraw_rate_unit: A[ + SignalR[str], PvSuffix("WRATEUNITS:RBV"), Format.UNCACHED_SIGNAL + ] + target_time: A[SignalRW[float], PvSuffix("TTIME"), Format.CONFIG_SIGNAL] + read_target_time: A[SignalR[float], PvSuffix("TTIME:RBV"), Format.UNCACHED_SIGNAL] + force: A[SignalRW[float], PvSuffix("FORCE"), Format.CONFIG_SIGNAL] + read_force: A[SignalR[float], PvSuffix("FORCE:RBV"), Format.UNCACHED_SIGNAL] + diameter: A[SignalR[float], PvSuffix("DIAMETER"), Format.CONFIG_SIGNAL] + + +_DDS_RANGES: Mapping[str, Mapping[float, tuple[float, float]]] = MappingProxyType( + { + "steel": MappingProxyType( + { + 2.5: (2.264, 2.351), + 8: (8.728, 9.064), + 20: (35.21, 36.56), + 50: (78.69, 81.72), + } + ), + "glass_H1000": MappingProxyType( + { + 1: (2.043, 2.121), + 1.25: (2.553, 2.651), + 2.5: (5.106, 5.302), + 5: (10.21, 10.6), + 10: (20.41, 21.2), + 25: (51.04, 53), + 50: (102.1, 106), + } + ), + "plastic_BD": MappingProxyType( + { + 1: (2.124, 2.206), + 3: (7.09, 7.363), + 5: (13.83, 14.36), + 10: (20.02, 20.79), + 20: (34.91, 36.26), + 30: (44.84, 46.57), + 50: (68.04, 70.66), + } + ), + } +) +_ULTRA_RANGES: Mapping[str, Mapping[float, tuple[float, float]]] = MappingProxyType( + { + "steel": MappingProxyType( + { + 2.5: (3.39588, 3.5265), + 8: (13.0924, 13.596), + 20: (52.8105, 54.8417), + 50: (118.038, 122.578), + 100: (175.769, 182.529), + } + ), + "glass_H1000": MappingProxyType( + { + 1: (3.06414, 3.18204), + 1.25: (3.82884, 3.97616), + 2.5: (7.65858, 7.95317), + 5: (15.3096, 15.8985), + 10: (30.6218, 31.7996), + 25: (76.5582, 79.5028), + 50: (153.111, 159.0), + 100: (153.111, 159.0), + } + ), + "plastic_BD": MappingProxyType( + { + 1: (3.18636, 3.30896), + 3: (10.6358, 11.0449), + 5: (20.7422, 21.5401), + 10: (30.036, 31.1913), + 20: (52.3697, 54.384), + 30: (67.2661, 69.8532), + 50: (102.061, 105.986), + 60: (102.061, 105.986), + } + ), + } +) + +_VOLUME_ML = MappingProxyType({"pl": 1e-9, "nl": 1e-6, "ul": 1e-3, "ml": 1.0}) +_TIME_MIN = MappingProxyType({"sec": 1 / 60, "min": 1.0, "hr": 60.0}) + + +def vol_unit_converter(v0: str, v1: str) -> float: + """Return the factor converting a volume from v0 to v1.""" + return _VOLUME_ML[v0] / _VOLUME_ML[v1] + + +def rate_unit_converter(r0: str, r1: str) -> float: + """Return the factor converting a volume/time rate from r0 to r1.""" + v0, t0 = r0.split("/") + v1, t1 = r1.split("/") + return vol_unit_converter(v0, v1) * _TIME_MIN[t1] / _TIME_MIN[t0] + + +def rate_limits( + pump: DDSSyringeAxis | UltraSyringePump, + syringe_material: str, + syringe_size_ml: float, +) -> tuple[float, float]: + """Return tabulated (minimum nl/min, maximum ml/min) for a syringe.""" + table = _DDS_RANGES if isinstance(pump, DDSSyringeAxis) else _ULTRA_RANGES + try: + return table[syringe_material][syringe_size_ml] + except KeyError as exc: + raise ValueError( + f"Unsupported {syringe_material} syringe size {syringe_size_ml} mL" + ) from exc + + +def validate_pump_settings( + pump: DDSSyringeAxis | UltraSyringePump, + syringe_material: str, + syringe_size_ml: float, + target_ml: float | None, + rate_ul_min: float, +) -> None: + """Reject settings beyond the syringe's documented theoretical range.""" + minimum, maximum = rate_limits(pump, syringe_material, syringe_size_ml) + if target_ml is not None and ( + not math.isfinite(target_ml) or target_ml < 0 or target_ml > syringe_size_ml + ): + raise ValueError( + f"Target {target_ml} mL exceeds syringe capacity {syringe_size_ml} mL" + ) + if not math.isfinite(rate_ul_min) or rate_ul_min < 0: + raise ValueError(f"Invalid pump rate {rate_ul_min} ul/min") + if rate_ul_min and not ( + minimum * rate_unit_converter("nl/min", "ul/min") + <= rate_ul_min + <= maximum * rate_unit_converter("ml/min", "ul/min") + ): + raise ValueError(f"Rate {rate_ul_min} ul/min outside syringe range") diff --git a/tests/test_pump_plans.py b/tests/test_pump_plans.py new file mode 100644 index 0000000..0a08c1e --- /dev/null +++ b/tests/test_pump_plans.py @@ -0,0 +1,232 @@ +"""RunEngine pump behavior without contacting EPICS hardware.""" + +import pytest +from bluesky import plan_stubs as bps +from ophyd_async.core import ( + callback_on_mock_execute, + callback_on_mock_put, + init_devices, + set_mock_value, +) + +from xpd_tools.pump_plans import ( + configure_infusion, + configure_withdrawal, + pump_session, + read_pump_metadata, + start_infusion, + start_withdrawal, + stop_pumps, + wait_for_equilibrium, +) +from xpd_tools.pumps import DDSSyringeAxis, DDSSyringeController, UltraSyringePump + + +@pytest.fixture +async def pumps(): + async with init_devices(mock=True): + controller = DDSSyringeController("DDS:1", name="DDS1") + first = DDSSyringeAxis("DDS:1-Ax:A}", controller, name="DDS1_p1") + second = DDSSyringeAxis("DDS:1-Ax:B}", controller, name="DDS1_p2") + ultra = UltraSyringePump("Ultra:1}:", name="Pump_Ultra1") + for pump in (first, second, ultra): + set_mock_value(pump.status, "Idle") + for direction in ("infuse", "withdraw"): + callback_on_mock_put( + getattr(pump, f"{direction}_rate"), + lambda value, pump=pump, direction=direction: set_mock_value( + getattr(pump, f"read_{direction}_rate"), value + ), + ) + callback_on_mock_put( + getattr(pump, f"{direction}_rate_unit"), + lambda value, pump=pump, direction=direction: set_mock_value( + getattr(pump, f"read_{direction}_rate_unit"), value + ), + ) + callback_on_mock_execute( + pump.pump_infuse, lambda pump=pump: set_mock_value(pump.status, "Infusing") + ) + callback_on_mock_execute( + pump.pump_withdraw, + lambda pump=pump: set_mock_value(pump.status, "Withdrawing"), + ) + callback_on_mock_execute( + pump.pump_stop, lambda pump=pump: set_mock_value(pump.status, "Idle") + ) + for pump in (first, second): + for direction in ("infuse", "withdraw"): + set_mock_value(getattr(pump, f"read_min_{direction}"), 78.69) + set_mock_value(getattr(pump, f"read_min_{direction}_unit"), "nl/min") + set_mock_value(getattr(pump, f"read_max_{direction}"), 81.72) + set_mock_value(getattr(pump, f"read_max_{direction}_unit"), "ml/min") + return first, second, ultra + + +def test_configure_start_zero_and_metadata(RE, pumps): + first, second, _ = pumps + RE( + configure_infusion( + [first, second], + [50, 50], + ["30 ml", "30 ml"], + ["100 ul/min", "0 ul/min"], + ["steel", "steel"], + [True, True], + ) + ) + assert RE(bps.rd(first.infuse_rate)).plan_result == 100 + assert RE(bps.rd(first.infuse_rate_unit)).plan_result == "ul/min" + assert RE(bps.rd(first.target_vol)).plan_result == 30 + RE(start_infusion([first, second], [100, "0 ul/min"])) + assert RE(read_pump_metadata([first, second])).plan_result == { + "pumps": ["DDS1_p1", "DDS1_p2"], + "infuse_rate": [100, 0], + "infuse_rate_unit": ["ul/min", ""], + "pump_status": ["Infusing", "Idle"], + } + RE(stop_pumps([first, second])) + assert RE(bps.rd(first.status)).plan_result == "Idle" + + +def test_preflight_rejects_wrong_syringe_and_limit(RE, pumps): + first, second, _ = pumps + set_mock_value(second.read_min_infuse, 35.21) + with pytest.raises(ValueError, match="selected DDS syringe"): + RE( + configure_infusion( + [first, second], + [50, 50], + ["30 ml"] * 2, + [100, 100], + ["steel"] * 2, + [True] * 2, + ) + ) + assert RE(bps.rd(first.infuse_rate)).plan_result == 0 + with pytest.raises(ValueError, match="outside syringe range"): + RE(configure_infusion([first], [50], ["30 ml"], [90000], ["steel"], [True])) + + +def test_ultra_and_withdrawal(RE, pumps): + first, _, ultra = pumps + with pytest.warns(UserWarning, match="Ultra panel"): + RE(configure_infusion([ultra], [50], ["30 ml"], [100], ["steel"], [True])) + RE(configure_withdrawal([first], [50], ["30 ml"], [100], ["steel"], [True])) + RE(start_withdrawal([first], [100])) + assert RE(bps.rd(first.status)).plan_result == "Withdrawing" + assert RE(bps.rd(first.read_withdraw_rate)).plan_result == 100 + RE(stop_pumps([first])) + + +def test_session_stops_on_success_failure_and_cleanup_failure(RE, pumps): + first, second, _ = pumps + + def setup(): + yield from bps.null() + + def body(): + yield from start_infusion([first], [100]) + yield from bps.null() + + RE(pump_session(body, setup=setup(), pumps=[first])) + assert RE(bps.rd(first.status)).plan_result == "Idle" + + def failing(): + yield from start_infusion([first, second], [100, 100]) + raise RuntimeError("acquisition failed") + + with pytest.raises(RuntimeError, match="acquisition failed"): + RE(pump_session(failing, setup=setup(), pumps=[first, second])) + assert RE(bps.rd(second.status)).plan_result == "Idle" + tracked = [] + + def failing_setup(): + tracked.append(first) + yield from start_infusion([first], [100]) + raise RuntimeError("setup failed") + + with pytest.raises(RuntimeError, match="setup failed"): + RE(pump_session(body, setup=failing_setup(), pumps=tracked)) + assert RE(bps.rd(first.status)).plan_result == "Idle" + + def failed_stop(): + raise RuntimeError("STOP failed") + + callback_on_mock_execute(first.pump_stop, failed_stop) + with pytest.raises(ExceptionGroup, match="Pump stop failures"): + RE(stop_pumps([first, second])) + assert RE(bps.rd(second.status)).plan_result == "Idle" + with pytest.raises(ExceptionGroup, match="Pump session and cleanup failures"): + RE(pump_session(failing, setup=setup(), pumps=[first, second])) + + +def test_equilibrium_segment_flow(RE, pumps): + first, second, _ = pumps + set_mock_value(first.status, "Infusing") + set_mock_value(first.read_infuse_rate, 100) + set_mock_value(first.read_infuse_rate_unit, "ul/min") + duration = RE(wait_for_equilibrium([["0 cm", first]], ratio=1)).plan_result + assert duration == 0 + with pytest.raises(ValueError, match="no positive infusing flow"): + RE(wait_for_equilibrium([["10 cm", second]], ratio=1)) + + +def test_flow_session_reads_after_start_and_stops(RE, pumps): + from pathlib import Path + + namespace = {} + startup = ( + Path(__file__).resolve().parents[1] + / "xpd-profile-collection/startup/97-xray_uvvis.py" + ) + exec(compile(startup.read_text(), str(startup), "exec"), namespace) + ultra = pumps[-1] + observed = [] + + def acquisition(): + observed.append((yield from read_pump_metadata([ultra]))) + yield from bps.null() + + with pytest.warns(UserWarning, match="Ultra panel"): + RE( + namespace["steady_state_flow"]( + acquisition, + [ultra], + [100], + syringe_list=[50], + target_vol_list=["30 ml"], + set_target_list=[True], + syringe_mater_list=["steel"], + mixer_lengths_cm=[0], + resident_t_ratio=0, + ) + ) + assert observed[0]["pump_status"] == ["Infusing"] + assert observed[0]["infuse_rate"] == [100] + assert RE(bps.rd(ultra.status)).plan_result == "Idle" + + def fail_sleep(_): + raise RuntimeError("wash interrupted") + yield + + original_sleep = namespace["bps"].sleep + namespace["bps"].sleep = fail_sleep + try: + with pytest.warns(UserWarning, match="Ultra panel"): + with pytest.raises(RuntimeError, match="wash interrupted"): + RE( + namespace["_wash_loop"]( + [ultra], + [50], + [100], + 1, + syringe_mater_list=["steel"], + target_vol_list=["30 ml"], + set_target_list=[True], + rate_unit="ul/min", + ) + ) + finally: + namespace["bps"].sleep = original_sleep + assert RE(bps.rd(ultra.status)).plan_result == "Idle" diff --git a/tests/test_pumps.py b/tests/test_pumps.py new file mode 100644 index 0000000..dbf4528 --- /dev/null +++ b/tests/test_pumps.py @@ -0,0 +1,62 @@ +"""Offline EPICS source and pump-policy checks.""" + +import math + +import pytest +from ophyd_async.core import init_devices, set_mock_value + +from xpd_tools.pumps import ( + DDSSyringeAxis, + DDSSyringeController, + UltraSyringePump, + rate_limits, + rate_unit_converter, + validate_pump_settings, + vol_unit_converter, +) + + +@pytest.mark.asyncio +async def test_pump_sources_and_readables(): + async with init_devices(mock=True): + controller = DDSSyringeController("XF:28IDC-ES:1{Pump:Syrng-DDS:1", name="DDS1") + axis = DDSSyringeAxis( + "XF:28IDC-ES:1{Pump:Syrng-DDS:1-Ax:A}", controller, name="DDS1_p1" + ) + ultra = UltraSyringePump( + "XF:28IDC-ES:1{Pump:Syrng-Ultra:1}:", name="Pump_Ultra1" + ) + assert axis.controller is controller + for pump, suffix in ((axis, "DDS:1-Ax:A}"), (ultra, "Ultra:1}:")): + set_mock_value(pump.status, "Idle") + set_mock_value(pump.read_infuse_rate, 100.0) + set_mock_value(pump.read_infuse_rate_unit, "ul/min") + readings = await pump.read() + assert readings[pump.status.name]["value"] == "Idle" + assert readings[pump.read_infuse_rate.name]["value"] == 100.0 + source = (await pump.status.describe())[pump.status.name]["source"] + assert source.endswith(f"{suffix}STATUS") + assert pump.infuse_rate.name in await pump.read_configuration() + assert pump.pump_infuse.name not in readings + assert (await controller.communication.describe())[controller.communication.name][ + "source" + ].endswith("DDS:1}DISABLE") + assert (await axis.read_min_infuse.describe())[axis.read_min_infuse.name][ + "source" + ].endswith("DDS:1-Ax:A}IRATE:MIN:RBV") + assert (await ultra.read_infuse_rate_unit.describe())[ + ultra.read_infuse_rate_unit.name + ]["source"].endswith("Ultra:1}:IRATEUNITS:RBV") + + +def test_syringe_policy(): + pump = UltraSyringePump("mock:", name="ultra") + assert rate_limits(pump, "steel", 50) == (118.038, 122.578) + assert math.isclose(rate_unit_converter("ml/hr", "ul/min"), 1000 / 60) + assert math.isclose(vol_unit_converter("nl", "ml"), 1e-6) + validate_pump_settings(pump, "steel", 50, 30, 100) + for rate in (-1, float("nan"), 200000): + with pytest.raises(ValueError): + validate_pump_settings(pump, "steel", 50, 30, rate) + with pytest.raises(ValueError): + validate_pump_settings(pump, "steel", 50, 51, 100) diff --git a/uv.lock b/uv.lock index 67b8c13..a36141c 100644 --- a/uv.lock +++ b/uv.lock @@ -3,10 +3,10 @@ revision = 3 requires-python = ">=3.11, <3.14" resolution-markers = [ "python_full_version >= '3.12' and sys_platform == 'win32'", - "python_full_version < '3.12' and sys_platform == 'win32'", "python_full_version >= '3.12' and sys_platform == 'emscripten'", - "python_full_version < '3.12' and sys_platform == 'emscripten'", "python_full_version >= '3.12' and sys_platform != 'emscripten' and sys_platform != 'win32'", + "python_full_version < '3.12' and sys_platform == 'win32'", + "python_full_version < '3.12' and sys_platform == 'emscripten'", "python_full_version < '3.12' and sys_platform != 'emscripten' and sys_platform != 'win32'", ] @@ -2458,8 +2458,8 @@ wheels = [ [[package]] name = "ophyd-async" -version = "0.21.2.dev4+gd2383e9a3" -source = { git = "https://github.com/bluesky/ophyd-async?rev=main#d2383e9a3f103f8f1759883f120689c278fb898f" } +version = "0.21.3" +source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "bluesky" }, { name = "colorlog" }, @@ -2472,6 +2472,10 @@ dependencies = [ { name = "stamina" }, { name = "velocity-profile" }, ] +sdist = { url = "https://files.pythonhosted.org/packages/2a/f6/d1878af8283b63c7af960e160122b0c7d565c6440ec6abd900776388dfce/ophyd_async-0.21.3.tar.gz", hash = "sha256:373b60c432366cb12b8ce40f289826ab53708b2b95d0e6be0c75404ea39bf9c0", size = 679766, upload-time = "2026-09-23T08:37:03.14Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/6c/01/626bf9585228485b82f0afb4460413f34d2898c8482fab9bfc200be369e1/ophyd_async-0.21.3-py3-none-any.whl", hash = "sha256:63a868c0c64cd2b349983dcce5a46ba9dcea6c922de1f7a3ea92c717f918afd4", size = 261955, upload-time = "2026-09-23T08:37:01.818Z" }, +] [package.optional-dependencies] ca = [ @@ -4835,7 +4839,7 @@ requires-dist = [ { name = "bluesky-tiled-plugins", specifier = ">=2.0.6" }, { name = "ipython", specifier = ">=9.15.0" }, { name = "nslsii", specifier = ">=0.11.9" }, - { name = "ophyd-async", extras = ["ca", "pva"], git = "https://github.com/bluesky/ophyd-async?rev=main" }, + { name = "ophyd-async", extras = ["ca", "pva"], specifier = "==0.21.3" }, { name = "pyside6", marker = "extra == 'profiles'", specifier = ">=6.9,<6.10" }, { name = "qtpy", marker = "extra == 'profiles'", specifier = ">=2.4.3" }, { name = "redis-json-dict", specifier = ">=0.2.2" }, @@ -4852,7 +4856,7 @@ dev = [ { name = "import-linter" }, { name = "inflection" }, { name = "numpydoc" }, - { name = "ophyd-async", extras = ["ca", "pva", "sim"], git = "https://github.com/bluesky/ophyd-async?rev=main" }, + { name = "ophyd-async", extras = ["ca", "pva", "sim"], specifier = "==0.21.3" }, { name = "pickleshare" }, { name = "pipdeptree" }, { name = "pre-commit" }, From 2987eb0f565a1a5e254b5146da00d8d210db75d7 Mon Sep 17 00:00:00 2001 From: thomashopkins32 Date: Mon, 12 Oct 2026 17:18:29 -0400 Subject: [PATCH 2/3] Switch back to ophyd-async@main --- pyproject.toml | 4 ++-- uv.lock | 12 ++++-------- 2 files changed, 6 insertions(+), 10 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 4108e7e..4df9d4d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,7 +32,7 @@ dependencies = [ "bluesky-tiled-plugins>=2.0.6", "ipython>=9.15.0", "nslsii>=0.11.9", - "ophyd-async[ca,pva]==0.21.3", + "ophyd-async[ca,pva] @ git+https://github.com/bluesky/ophyd-async@main", "redis-json-dict>=0.2.2", "rich>=15.0.0", "tiled[client]>=0.2.12", @@ -41,7 +41,7 @@ dependencies = [ [dependency-groups] dev = [ - "ophyd-async[ca,pva,sim]==0.21.3", + "ophyd-async[ca,pva,sim] @ git+https://github.com/bluesky/ophyd-async@main", "inflection", "import-linter", "numpydoc", diff --git a/uv.lock b/uv.lock index a36141c..2f814b8 100644 --- a/uv.lock +++ b/uv.lock @@ -2458,8 +2458,8 @@ wheels = [ [[package]] name = "ophyd-async" -version = "0.21.3" -source = { registry = "https://pypi.org/simple" } +version = "0.21.4.dev11+g888f84c63" +source = { git = "https://github.com/bluesky/ophyd-async?rev=main#888f84c633bce8962e5f2f941da5ce507de247e9" } dependencies = [ { name = "bluesky" }, { name = "colorlog" }, @@ -2472,10 +2472,6 @@ dependencies = [ { name = "stamina" }, { name = "velocity-profile" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/2a/f6/d1878af8283b63c7af960e160122b0c7d565c6440ec6abd900776388dfce/ophyd_async-0.21.3.tar.gz", hash = "sha256:373b60c432366cb12b8ce40f289826ab53708b2b95d0e6be0c75404ea39bf9c0", size = 679766, upload-time = "2026-09-23T08:37:03.14Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/6c/01/626bf9585228485b82f0afb4460413f34d2898c8482fab9bfc200be369e1/ophyd_async-0.21.3-py3-none-any.whl", hash = "sha256:63a868c0c64cd2b349983dcce5a46ba9dcea6c922de1f7a3ea92c717f918afd4", size = 261955, upload-time = "2026-09-23T08:37:01.818Z" }, -] [package.optional-dependencies] ca = [ @@ -4839,7 +4835,7 @@ requires-dist = [ { name = "bluesky-tiled-plugins", specifier = ">=2.0.6" }, { name = "ipython", specifier = ">=9.15.0" }, { name = "nslsii", specifier = ">=0.11.9" }, - { name = "ophyd-async", extras = ["ca", "pva"], specifier = "==0.21.3" }, + { name = "ophyd-async", extras = ["ca", "pva"], git = "https://github.com/bluesky/ophyd-async?rev=main" }, { name = "pyside6", marker = "extra == 'profiles'", specifier = ">=6.9,<6.10" }, { name = "qtpy", marker = "extra == 'profiles'", specifier = ">=2.4.3" }, { name = "redis-json-dict", specifier = ">=0.2.2" }, @@ -4856,7 +4852,7 @@ dev = [ { name = "import-linter" }, { name = "inflection" }, { name = "numpydoc" }, - { name = "ophyd-async", extras = ["ca", "pva", "sim"], specifier = "==0.21.3" }, + { name = "ophyd-async", extras = ["ca", "pva", "sim"], git = "https://github.com/bluesky/ophyd-async?rev=main" }, { name = "pickleshare" }, { name = "pipdeptree" }, { name = "pre-commit" }, From a79069976309259372d8c46d69a993e67d75c67f Mon Sep 17 00:00:00 2001 From: thomashopkins32 Date: Mon, 12 Oct 2026 17:22:57 -0400 Subject: [PATCH 3/3] Remove test that depends on xpd-profile-collection --- tests/test_pump_plans.py | 60 ---------------------------------------- 1 file changed, 60 deletions(-) diff --git a/tests/test_pump_plans.py b/tests/test_pump_plans.py index 0a08c1e..2d774b2 100644 --- a/tests/test_pump_plans.py +++ b/tests/test_pump_plans.py @@ -170,63 +170,3 @@ def test_equilibrium_segment_flow(RE, pumps): assert duration == 0 with pytest.raises(ValueError, match="no positive infusing flow"): RE(wait_for_equilibrium([["10 cm", second]], ratio=1)) - - -def test_flow_session_reads_after_start_and_stops(RE, pumps): - from pathlib import Path - - namespace = {} - startup = ( - Path(__file__).resolve().parents[1] - / "xpd-profile-collection/startup/97-xray_uvvis.py" - ) - exec(compile(startup.read_text(), str(startup), "exec"), namespace) - ultra = pumps[-1] - observed = [] - - def acquisition(): - observed.append((yield from read_pump_metadata([ultra]))) - yield from bps.null() - - with pytest.warns(UserWarning, match="Ultra panel"): - RE( - namespace["steady_state_flow"]( - acquisition, - [ultra], - [100], - syringe_list=[50], - target_vol_list=["30 ml"], - set_target_list=[True], - syringe_mater_list=["steel"], - mixer_lengths_cm=[0], - resident_t_ratio=0, - ) - ) - assert observed[0]["pump_status"] == ["Infusing"] - assert observed[0]["infuse_rate"] == [100] - assert RE(bps.rd(ultra.status)).plan_result == "Idle" - - def fail_sleep(_): - raise RuntimeError("wash interrupted") - yield - - original_sleep = namespace["bps"].sleep - namespace["bps"].sleep = fail_sleep - try: - with pytest.warns(UserWarning, match="Ultra panel"): - with pytest.raises(RuntimeError, match="wash interrupted"): - RE( - namespace["_wash_loop"]( - [ultra], - [50], - [100], - 1, - syringe_mater_list=["steel"], - target_vol_list=["30 ml"], - set_target_list=[True], - rate_unit="ul/min", - ) - ) - finally: - namespace["bps"].sleep = original_sleep - assert RE(bps.rd(ultra.status)).plan_result == "Idle"