diff --git a/csaxs_bec/device_configs/smargon.yaml b/csaxs_bec/device_configs/smargon.yaml new file mode 100644 index 00000000..b0d23421 --- /dev/null +++ b/csaxs_bec/device_configs/smargon.yaml @@ -0,0 +1,61 @@ +# Example BEC device-config entries for the SmarGon goniometer (via smargopolo). +# +# The device controls the virtual SCS axes; smargopolo runs the kinematics and +# drives the underlying q1..q6 stages. Referencing is a deliberate operator action: +# after loading, call `smargon.reference()` once smargopolo is up. + +# --- Real goniometer (smargopolo REST API) --------------------------------------- +smargon: + description: SmarGon goniometer (SCS axes via smargopolo REST API) + deviceClass: csaxs_bec.devices.smargon.smargon.SmarGon + deviceConfig: + host: smargopolo # smargopolo server hostname + port: 3000 # REST API port + timeout: 5.0 # per-request transport timeout [s] + has_omega: false # set true on builds with the OMEGA rotation drive + # Per-move completion tolerance: a float (all axes) or per-axis mapping. + tolerance: + SHX: 0.001 # mm + SHY: 0.001 + SHZ: 0.001 + CHI: 0.01 # deg + PHI: 0.01 + # Soft limits in SCS units [low, high]. These are typically USER-set (per + # experiment), not hardware ranges: the true per-axis travel of a SmarGon is + # coupled across all SCS axes and cannot be reliably pre-declared. smargopolo + # owns the real hardware limits; if a target violates them the move fails (the + # PUT is rejected, or -- if the fault only shows after partial motion -- the + # device aborts the move on the controller's ERROR mode). Omit an axis here to + # leave it unbounded in BEC and rely on the controller. + limits: + SHX: [-2.5, 2.5] + SHY: [-2.5, 2.5] + SHZ: [-2.5, 2.5] + poll_interval: 0.1 # readback polling cadence while moving [s] + move_timeout: 60.0 # per-move timeout [s] + deviceTags: + - goniometer + - smargon + enabled: true + onFailure: retry + readOnly: false + readoutPriority: baseline + softwareTrigger: false + +# --- Simulation / offline (no server; uses FakeTransport) ------------------------ +smargon_sim: + description: SmarGon goniometer (simulation, no smargopolo server required) + deviceClass: csaxs_bec.devices.smargon.smargon.SmarGon + deviceConfig: + simulation: true + has_omega: true # exercise OMEGA in simulation + gradual_motion: true # simulate finite-speed travel (exercise move status) + deviceTags: + - goniometer + - smargon + - simulation + enabled: true + onFailure: retry + readOnly: false + readoutPriority: baseline + softwareTrigger: false diff --git a/csaxs_bec/devices/smargon/README.md b/csaxs_bec/devices/smargon/README.md new file mode 100644 index 00000000..53e02a46 --- /dev/null +++ b/csaxs_bec/devices/smargon/README.md @@ -0,0 +1,103 @@ +# SmarGon goniometer (BEC integration via smargopolo) + +BEC ophyd device for the [SmarGon](https://www.smaract.com/en/smargon) goniometer. +It controls the **virtual SmarGon Coordinate System (SCS) axes** through the +[smargopolo](http://smargopolo:3000) RESTful API. smargopolo runs the kinematics and +drives the underlying `q1..q6` SmarAct MCS2 stages; this device never commands those +directly (that is the separate low-level `csaxs_bec.devices.smaract` device). + +## Layout + +``` +smargon/ + transport/ # ophyd-free control plane (unit-testable, no BEC needed) + base.py # SmargopoloTransport ABC + Axis/Mode enums + AXIS_INFO + rest.py # RestTransport -- real urllib client against :3000 + fake.py # FakeTransport -- in-memory sim (offline mode + tests) + errors.py # typed exceptions + positioner.py # SmargonAxis(Device, PositionerBase) -- one SCS axis + smargon.py # SmarGon(PSIDeviceBase) -- parent: transport, axes, mode +``` + +The transport mirrors the Canon CR-N300 device's structure (HTTP transport + an +in-memory fake + threaded polling positioners), since both wrap a non-EPICS, +poll-to-confirm HTTP backend. + +## Axes (v1) + +Movable positioners: `shx shy shz` (mm) and `chi phi` (deg). `omega` (deg) is +**optional hardware** — enable it per deployment with `has_omega: true`; otherwise +it is `Kind.omitted` and refuses to move. The O-vector (`OX/OY/OZ`) and beamline +nudges (`BX/BY/BZ`) are reachable through the transport but not exposed as v1 +positioners; `nudge_beamline()` provides the BCS nudge path. + +Raw `q1..q6` motor positions are read-only diagnostics via `motor_positions()`. + +## Referencing (read this before moving) + +smargopolo only follows targets in `Mode.READY`. **Referencing physically moves the +goniometer**, so it is a deliberate operator action — never automatic: + +```python +smargon.get_mode() # 'UNINITIALIZED' on a fresh server +smargon.reference() # PUT /mode?mode=1, waits until READY (moves hardware!) +smargon.is_ready() # True +smargon.shx.move(0.5) # now honoured +``` + +Any axis move while not READY raises `SmargopoloModeError` with a "reference first" +message rather than silently doing nothing. `mode` and `rosout` signals surface the +state machine and last ROS message (error detail) for diagnostics. + +## Usage + +```python +smargon.shx.move(0.5) # single-axis move (mm) +smargon.move_scs(SHX=0.1, SHZ=18) # coordinated move in one /targetSCS call +smargon.nudge_beamline(BX=0.01) # beamline-frame nudge (/nudgeBCS) +smargon.stop_all() # halt all enabled axes +smargon.transport_info() # connection / identity diagnostics +``` + +## Configuration + +See `csaxs_bec/device_configs/smargon.yaml` for real + simulation entries. +Key `deviceConfig` keys: `host`, `port`, `has_omega`, `tolerance` (float or per-axis +map), `limits` (per-axis `[low, high]` soft limits — **verify against the build**), +`poll_interval`, `move_timeout`. Simulation: `simulation: true` (+ optional +`gradual_motion: true`) uses `FakeTransport`, no server required. + +## Tests + +```sh +pytest tests/tests_devices/test_smargon_transport.py tests/tests_devices/test_smargon_device.py +``` + +All run against `FakeTransport` — no hardware or server needed. + +## Limits & coupled-axis faults + +SCS soft limits are **user-set** (per experiment), not hardware ranges: a SmarGon's +true per-axis travel is coupled across all SCS axes (like a hexapod) and can't be +reliably declared in advance. smargopolo owns the real limits. The device handles +both failure modes: + +- **Rejected up front** — if smargopolo refuses an out-of-range target, the PUT + returns an error and the move status fails immediately. +- **Faulted mid-move** — if a target is accepted (`202`) but the controller goes to + `Mode.ERROR` only after partial motion, the move waiters detect it on the next + poll and fail the status with the controller's `rosout` detail (rather than + running to the move timeout). See `SmarGon._raise_if_error`. + +Move completion is by **readback tolerance** (user-set; done on the first in-tolerance +sample, so active position-hold dithering doesn't block completion). + +## Open items / to confirm with the smargopolo maintainer (Wayne Glettig) + +- **Stop semantics**: no explicit stop endpoint; we retarget axes to their readback + ("Follow Target" halt). Confirm this is the intended way to abort a move. +- **Mid-move fault signal**: we infer faults from `Mode.ERROR`. Confirm an + out-of-range / coupled-limit violation reliably drives `/mode` to 99 (and that + `rosout` carries the reason), so the abort path is dependable. +- *(resolved)* OMEGA is drivable via `/targetSCS` wherever it physically exists — + handled by `has_omega`. diff --git a/csaxs_bec/devices/smargon/__init__.py b/csaxs_bec/devices/smargon/__init__.py new file mode 100644 index 00000000..e869ba65 --- /dev/null +++ b/csaxs_bec/devices/smargon/__init__.py @@ -0,0 +1,32 @@ +"""SmarGon goniometer integration for BEC (via the smargopolo REST API). + +Public entry point is :class:`SmarGon`. The control transport +(:mod:`.transport`) is ophyd-free and independently importable, so the REST client +and its in-memory fake can be used and tested without the BEC stack. +""" + +from __future__ import annotations + +from .positioner import SmargonAxis +from .smargon import SmarGon +from .transport import ( + Axis, + BeamlineAxis, + FakeTransport, + Mode, + RestTransport, + SmargopoloInfo, + SmargopoloTransport, +) + +__all__ = [ + "SmarGon", + "SmargonAxis", + "Axis", + "BeamlineAxis", + "Mode", + "SmargopoloTransport", + "SmargopoloInfo", + "FakeTransport", + "RestTransport", +] diff --git a/csaxs_bec/devices/smargon/positioner.py b/csaxs_bec/devices/smargon/positioner.py new file mode 100644 index 00000000..18a03279 --- /dev/null +++ b/csaxs_bec/devices/smargon/positioner.py @@ -0,0 +1,227 @@ +"""Transport-backed positioner for a single SmarGon SCS axis. + +``SmargonAxis`` is a real ophyd positioner (``Device`` + ``PositionerBase``) so BEC +treats each SCS axis (SHX/SHY/SHZ/CHI/PHI/OMEGA) as a motor: ``move()``/``set()`` +return a ``DeviceStatus`` that completes when the axis reaches target (or on +timeout/stop), plus ``stop()``. It follows the threaded-move pattern of the Canon +``CanonAxis`` but reads/writes through the parent device's injected +:class:`~csaxs_bec.devices.smargon.transport.base.SmargopoloTransport`. + +Because smargopolo runs the kinematics, each axis simply PUTs its own SCS target +and polls its own SCS readback; the coupling to the q1..q6 motors is the server's +job, not ours. The axis never opens its own connection: it borrows the parent's +transport and a shared lock so concurrent axis commands serialise onto one session. +""" + +from __future__ import annotations + +import threading +import time +from typing import Callable + +from ophyd import Component as Cpt +from ophyd import Device, DeviceStatus, Kind, PositionerBase, Signal +from ophyd.utils import LimitError + +from .transport.base import Axis, SmargopoloTransport + + +class SmargonAxis(Device, PositionerBase): + """One SCS axis of the SmarGon goniometer. + + Parameters + ---------- + axis: + Which :class:`Axis` this positioner drives. + egu: + Engineering units for the readback (``"mm"`` or ``"deg"``). + limits: + Optional soft ``(low, high)`` limits. ``None`` (default) means unbounded + -- the SCS travel ranges are build-specific and should be set in config. + tolerance: + Absolute readback tolerance for declaring a move complete. + poll_interval: + Seconds between readback polls while moving. + move_timeout: + Seconds before a move fails with ``TimeoutError``. + """ + + readback = Cpt(Signal, value=0.0, kind=Kind.hinted) + setpoint = Cpt(Signal, value=0.0, kind=Kind.normal) + motor_is_moving = Cpt(Signal, value=0, kind=Kind.normal) + + SUB_READBACK = "readback" + _default_sub = SUB_READBACK + + def __init__( + self, + prefix: str = "", + *, + name: str, + axis: Axis, + egu: str = "", + limits: tuple[float, float] | None = None, + tolerance: float = 0.001, + poll_interval: float = 0.1, + move_timeout: float = 60.0, + parent=None, + kind=None, + **kwargs, + ) -> None: + self._axis = Axis(axis) + self._egu = egu + self._tolerance = tolerance + self._poll_interval = poll_interval + self._move_timeout = move_timeout + self._limits = tuple(limits) if limits is not None else None + self._stopped = False + self._move_thread: threading.Thread | None = None + self._fallback_lock = threading.RLock() + # Optional gate set by the parent (e.g. refuse to move unless READY). + self.pre_move_hook: Callable[[float], None] | None = None + + super().__init__(prefix, name=name, parent=parent, kind=kind, **kwargs) + + # -- transport access -------------------------------------------------- + + @property + def _transport(self) -> SmargopoloTransport: + transport = getattr(self.parent, "transport", None) + if transport is None: + raise RuntimeError(f"{self.name}: parent has no transport") + return transport + + @property + def _lock(self) -> threading.RLock: + return getattr(self.parent, "_transport_lock", None) or self._fallback_lock + + def _call(self, fn): + """Run a transport call under the shared lock.""" + with self._lock: + return fn() + + # -- positioner interface ---------------------------------------------- + + @property + def limits(self) -> tuple[float, float]: + return self._limits if self._limits is not None else (0.0, 0.0) + + @property + def low_limit(self) -> float: + return self.limits[0] + + @property + def high_limit(self) -> float: + return self.limits[1] + + @property + def egu(self) -> str: + return self._egu + + @property + def position(self) -> float: + return self.readback.get() + + def check_value(self, value: float) -> None: + if self._limits is None: + return + low, high = self._limits + if low != high and not low <= value <= high: + raise LimitError(f"{self.name}: position {value} outside limits {self._limits}") + + def _update_readback(self, value: float) -> None: + old = self.readback.get() + self.readback.put(value) + self._set_position(value) + self._run_subs( + sub_type=self.SUB_READBACK, old_value=old, value=value, timestamp=time.time() + ) + + def _check_controller_error(self) -> None: + """Raise if the parent reports smargopolo went to ERROR mid-move. + + Coupled hardware limits (the underlying stages' true ranges depend on all + SCS axes and are not reliably known to BEC) mean a target accepted with a + ``202`` can still fault *after* partial motion -- like a hexapod hitting a + joint limit. The parent installs this so the move fails loudly with the + controller's ``rosout`` detail rather than silently timing out. + """ + check = getattr(self.parent, "_raise_if_error", None) + if check is not None: + self._call(check) + + def _poll_until_target(self, target: float, status: DeviceStatus) -> None: + deadline = time.monotonic() + self._move_timeout + try: + while not self._stopped: + pos = self._call(lambda: self._transport.get_position(self._axis)) + self._update_readback(pos) + if abs(pos - target) <= self._tolerance: + break + self._check_controller_error() + if time.monotonic() > deadline: + raise TimeoutError( + f"{self.name}: move to {target} timed out after {self._move_timeout}s" + ) + time.sleep(self._poll_interval) + pos = self._call(lambda: self._transport.get_position(self._axis)) + self._update_readback(pos) + except Exception as exc: # pylint: disable=broad-except + self.motor_is_moving.put(0) + if not status.done: + status.set_exception(exc) + return + self.motor_is_moving.put(0) + if self._stopped: + if not status.done: + status.set_exception(RuntimeError(f"{self.name}: move interrupted by stop()")) + return + if not status.done: + status.set_finished() + + def move(self, position: float, wait: bool = True, timeout=None, moved_cb=None) -> DeviceStatus: + """Move the axis to *position*; return a status that completes at target.""" + self.check_value(position) + if self.pre_move_hook is not None: + self.pre_move_hook(position) + self._stopped = False + self.setpoint.put(position) + + status = DeviceStatus(self, timeout=timeout or self._move_timeout) + if moved_cb is not None: + status.add_callback(moved_cb) + self.motor_is_moving.put(1) + + try: + self._call(lambda: self._transport.move_absolute({self._axis: position})) + except Exception as exc: # pylint: disable=broad-except + self.motor_is_moving.put(0) + status.set_exception(exc) + return status + + self._move_thread = threading.Thread( + target=self._poll_until_target, args=(position, status), daemon=True + ) + self._move_thread.start() + + if wait: + status.wait() + return status + + def set(self, position: float, **kwargs) -> DeviceStatus: + """Alias for :meth:`move` with ``wait=False`` (BEC positioner interface).""" + return self.move(position, wait=False, **kwargs) + + def stop(self, *, success: bool = False) -> None: + """Halt this axis immediately (best-effort: never raises during teardown).""" + self._stopped = True + try: + self._call(lambda: self._transport.stop([self._axis])) + except Exception as exc: # pylint: disable=broad-except + self.log.warning("%s: transport stop failed: %s", self.name, exc) + finally: + if self._move_thread is not None: + self._move_thread.join(timeout=2.0) + self._move_thread = None + self.motor_is_moving.put(0) + super().stop(success=success) diff --git a/csaxs_bec/devices/smargon/smargon.py b/csaxs_bec/devices/smargon/smargon.py new file mode 100644 index 00000000..b5d2ee77 --- /dev/null +++ b/csaxs_bec/devices/smargon/smargon.py @@ -0,0 +1,397 @@ +"""SmarGon goniometer -- BEC ophyd device (via the smargopolo REST API). + +This device controls the *virtual* SmarGon Coordinate System (SCS) axes. The +smargopolo server runs the kinematics and drives the underlying q1..q6 stages; we +never command those directly (that is the separate low-level ``smaract`` device). + +Architecture +------------ +* ``self.transport`` -- a :class:`SmargopoloTransport` (real :class:`RestTransport` + or, when ``simulation: true``, :class:`FakeTransport`). All control goes through + it, serialised behind ``self._transport_lock``. +* ``shx/shy/shz/chi/phi`` (+ optional ``omega``) -- :class:`SmargonAxis` + positioners sharing the transport; they look like motors to BEC. +* ``mode`` / ``rosout`` -- read-only signals proxying the state machine + last ROS + message for diagnostics. + +State machine / referencing +--------------------------- +smargopolo only follows targets in ``Mode.READY``. Referencing physically moves +the goniometer, so it is a **deliberate operator action** here -- call +:meth:`reference`. Axes refuse to move unless READY (a pre-move guard surfaces a +clear "reference first" error instead of silently doing nothing). +""" + +from __future__ import annotations + +import threading +import time +from typing import TYPE_CHECKING + +from ophyd import Component as Cpt +from ophyd import Kind, Signal +from ophyd.utils import ReadOnlyError +from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase + +from .positioner import SmargonAxis +from .transport.base import Axis, BeamlineAxis, Mode, SmargopoloTransport +from .transport.errors import SmargopoloModeError +from .transport.fake import FakeTransport +from .transport.rest import RestTransport + +if TYPE_CHECKING: # pragma: no cover + from bec_lib.devicemanager import ScanInfo + +try: # pragma: no cover + from bec_lib.logger import bec_logger + + logger = bec_logger.logger +except Exception: # pragma: no cover + import logging + + logger = logging.getLogger(__name__) + + +# Axes exposed as movable positioners in v1. OMEGA is appended at runtime when the +# deployment has the rotation drive (``has_omega: true``). +_V1_AXES: tuple[Axis, ...] = (Axis.SHX, Axis.SHY, Axis.SHZ, Axis.CHI, Axis.PHI) + + +class SmargopoloModeSignal(Signal): + """Read-only signal exposing the smargopolo state-machine mode (as an int).""" + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self._metadata.update(connected=True, write_access=False) + + # pylint: disable=arguments-differ + def get(self, **kwargs): + value = int(self.parent._get_mode()) # noqa: SLF001 + self._readback = value + return value + + def put(self, value, **kwargs): + raise ReadOnlyError(f"{self.name} is read-only; use reference() to change mode") + + @property + def timestamp(self): + return time.time() + + +class SmargopoloRosoutSignal(Signal): + """Read-only signal exposing the last ROS message (carries error detail).""" + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self._metadata.update(connected=True, write_access=False) + + # pylint: disable=arguments-differ + def get(self, **kwargs): + value = self.parent._get_info().rosout # noqa: SLF001 + self._readback = value + return value + + def put(self, value, **kwargs): + raise ReadOnlyError(f"{self.name} is read-only") + + @property + def timestamp(self): + return time.time() + + +class SmarGon(PSIDeviceBase): + """BEC ophyd device for the SmarGon goniometer driven through smargopolo. + + See the module docstring for architecture and the referencing contract. + """ + + USER_ACCESS = [ + "reference", + "is_ready", + "wait_until_ready", + "get_mode", + "stop_all", + "motor_positions", + "move_scs", + "nudge_beamline", + "transport_info", + ] + + # -- motion axes (look like motors to BEC) ----------------------------- + shx = Cpt(SmargonAxis, axis=Axis.SHX, egu="mm", kind=Kind.normal) + shy = Cpt(SmargonAxis, axis=Axis.SHY, egu="mm", kind=Kind.normal) + shz = Cpt(SmargonAxis, axis=Axis.SHZ, egu="mm", kind=Kind.normal) + chi = Cpt(SmargonAxis, axis=Axis.CHI, egu="deg", kind=Kind.normal) + phi = Cpt(SmargonAxis, axis=Axis.PHI, egu="deg", kind=Kind.normal) + # Optional rotation drive -- omitted unless ``has_omega`` is set (see __init__). + omega = Cpt(SmargonAxis, axis=Axis.OMEGA, egu="deg", kind=Kind.normal) + + # -- diagnostics ------------------------------------------------------- + mode = Cpt(SmargopoloModeSignal, value=int(Mode.UNINITIALIZED), kind=Kind.normal) + rosout = Cpt(SmargopoloRosoutSignal, value="", kind=Kind.config) + + def __init__( + self, + *, + name: str, + host: str = "smargopolo", + port: int = 3000, + simulation: bool = False, + timeout: float = 5.0, + has_omega: bool = False, + limits: dict | None = None, + tolerance: float | dict | None = None, + poll_interval: float = 0.1, + move_timeout: float = 60.0, + gradual_motion: bool = False, + scan_info: "ScanInfo | None" = None, + device_manager=None, + **kwargs, + ) -> None: + """See class/module docstring for the architecture. + + Args: + host/port: smargopolo server address (ignored when ``simulation``). + simulation: use :class:`FakeTransport` -- no server needed. + timeout: per-request transport timeout in seconds. + has_omega: enable the OMEGA rotation axis (build-dependent). + limits: optional ``{axis_name: [low, high]}`` soft limits (SCS units). + tolerance: per-move completion tolerance; a float (all axes) or a + ``{axis_name: value}`` mapping. Defaults to 1e-3. + poll_interval/move_timeout: readback polling cadence / move timeout. + gradual_motion: simulate finite-speed travel in the fake transport. + """ + self._simulation = simulation + self._has_omega = has_omega + self._transport_lock = threading.RLock() + + # Build the transport BEFORE super().__init__ so axis components can reach it. + if simulation: + self.transport: SmargopoloTransport = FakeTransport(gradual_motion=gradual_motion) + else: + self.transport = RestTransport(host=host, port=port, timeout=timeout) + + super().__init__(name=name, scan_info=scan_info, device_manager=device_manager, **kwargs) + + # Configure the enabled axes; disable OMEGA unless the build has it. + enabled = set(_V1_AXES) | ({Axis.OMEGA} if has_omega else set()) + limits = limits or {} + for ax in self._all_axes(): + if ax._axis not in enabled: # noqa: SLF001 + ax.kind = Kind.omitted + ax.pre_move_hook = self._make_disabled_hook(ax._axis) # noqa: SLF001 + continue + ax.pre_move_hook = self._ready_guard + ax._poll_interval = poll_interval # noqa: SLF001 + ax._move_timeout = move_timeout # noqa: SLF001 + lim = limits.get(ax._axis.value) # noqa: SLF001 + if lim is not None: + ax._limits = (float(lim[0]), float(lim[1])) # noqa: SLF001 + tol = self._axis_tolerance(tolerance, ax._axis) # noqa: SLF001 + if tol is not None: + ax._tolerance = float(tol) # noqa: SLF001 + + # -- axis helpers ------------------------------------------------------ + + def _all_axes(self) -> tuple[SmargonAxis, ...]: + return (self.shx, self.shy, self.shz, self.chi, self.phi, self.omega) + + def _enabled_axes(self) -> tuple[SmargonAxis, ...]: + return tuple(ax for ax in self._all_axes() if ax.kind != Kind.omitted) + + @staticmethod + def _axis_tolerance(tolerance, axis: Axis): + if tolerance is None: + return None + if isinstance(tolerance, dict): + return tolerance.get(axis.value) + return tolerance + + @staticmethod + def _make_disabled_hook(axis: Axis): + def _hook(_position: float) -> None: + raise SmargopoloModeError( + f"axis {axis.value} is not enabled on this SmarGon " + "(set has_omega: true in the device config to enable OMEGA)" + ) + + return _hook + + def _ready_guard(self, _position: float) -> None: + mode = self._get_mode() + if mode is not Mode.READY: + raise SmargopoloModeError( + f"{self.name} is in {mode.name} (mode {int(mode)}); call reference() and wait " + "for READY before moving" + ) + + # -- transport routing ------------------------------------------------- + + def _get_mode(self) -> Mode: + with self._transport_lock: + return self.transport.get_mode() + + def _get_info(self): + with self._transport_lock: + return self.transport.get_info() + + def _raise_if_error(self) -> None: + """Raise if smargopolo is in ERROR (used to abort in-flight moves). + + smargopolo accepts a target with ``202`` even when the coupled hardware + limits will be violated, then faults once motion hits the real boundary. + Move waiters poll this so a mid-move fault surfaces the controller's + ``rosout`` message instead of running until the move timeout. + """ + info = self._get_info() + if info.mode is Mode.ERROR: + raise SmargopoloModeError( + f"{self.name}: smargopolo entered ERROR during motion " + f"(likely a hardware/coupled-axis limit): {info.rosout or 'no detail'}" + ) + + # -- connection lifecycle ---------------------------------------------- + + # pylint: disable=arguments-differ + def wait_for_connection(self, all_signals: bool = False, timeout: float = 10.0) -> None: + """Open the transport session, then wait for signals.""" + with self._transport_lock: + self.transport.connect(timeout=timeout) + super().wait_for_connection(all_signals=all_signals, timeout=timeout) + + def on_connected(self) -> None: + """Log identity and seed enabled-axis readbacks.""" + try: + info = self._get_info() + logger.info( + f"{self.name} connected to smargopolo (S/N {info.serial_number!r}, " + f"mode {info.mode.name})" + ) + if info.mode is not Mode.READY: + logger.warning( + f"{self.name}: smargopolo is {info.mode.name}; call reference() before moving" + ) + except Exception as exc: # pylint: disable=broad-except + logger.warning(f"{self.name}: could not read smargopolo info: {exc}") + for ax in self._enabled_axes(): + try: + with self._transport_lock: + ax._update_readback(self.transport.get_position(ax._axis)) # noqa: SLF001 + except Exception as exc: # pylint: disable=broad-except + logger.warning(f"{self.name}: initial readback for {ax.name} failed: {exc}") + + def on_stop(self) -> None: + """Halt all enabled axes.""" + self.stop_all() + + def on_destroy(self) -> None: + """Close the transport session.""" + try: + self.transport.close() + except Exception as exc: # pylint: disable=broad-except + logger.warning(f"{self.name}: transport close failed: {exc}") + + # -- control convenience (USER_ACCESS) --------------------------------- + + def reference(self, wait: bool = True, timeout: float = 120.0) -> Mode: + """Reference the goniometer (``Mode.REFERENCING`` -> ``READY``). + + WARNING: this physically moves the goniometer to find its reference marks. + It is a deliberate operator action. Returns the resulting mode. + """ + with self._transport_lock: + self.transport.set_mode(Mode.REFERENCING) + logger.info(f"{self.name}: referencing started") + if wait: + mode = self.wait_until_ready(timeout=timeout) + logger.info(f"{self.name}: referencing finished -> {mode.name}") + return mode + return self._get_mode() + + def is_ready(self) -> bool: + """Whether smargopolo is in ``Mode.READY`` (follow-target).""" + return self._get_mode() is Mode.READY + + def wait_until_ready(self, timeout: float = 120.0, poll: float = 0.5) -> Mode: + """Block until the mode is READY (or ERROR), or *timeout* elapses.""" + deadline = time.monotonic() + timeout + while True: + mode = self._get_mode() + if mode is Mode.READY: + return mode + if mode is Mode.ERROR: + info = self._get_info() + raise SmargopoloModeError(f"{self.name}: smargopolo in ERROR: {info.rosout}") + if time.monotonic() > deadline: + raise TimeoutError(f"{self.name}: not READY after {timeout}s (mode {mode.name})") + time.sleep(poll) + + def get_mode(self) -> str: + """Return the current state-machine mode name (for the user console).""" + return self._get_mode().name + + def stop_all(self) -> None: + """Halt every enabled axis immediately.""" + for ax in self._enabled_axes(): + ax.stop() + + def move_scs(self, wait: bool = True, **targets: float) -> None: + """Move several SCS axes at once in a single ``/targetSCS`` call. + + More efficient than moving axes one-by-one for coordinated moves, e.g. + ``smargon.move_scs(SHX=0.1, SHZ=18)``. Keys are SCS axis names. + """ + self._ready_guard(0.0) + parsed = {Axis(k): float(v) for k, v in targets.items()} + for ax in parsed: + self._axis_for(ax).check_value(parsed[ax]) + with self._transport_lock: + self.transport.move_absolute(parsed) + if wait: + for ax_enum, val in parsed.items(): + self._wait_axis(self._axis_for(ax_enum), val) + + def _wait_axis(self, axis: SmargonAxis, target: float, timeout: float | None = None) -> None: + deadline = time.monotonic() + (timeout or axis._move_timeout) # noqa: SLF001 + while True: + with self._transport_lock: + pos = self.transport.get_position(axis._axis) # noqa: SLF001 + axis._update_readback(pos) # noqa: SLF001 + if abs(pos - target) <= axis._tolerance: # noqa: SLF001 + return + self._raise_if_error() + if time.monotonic() > deadline: + raise TimeoutError(f"{axis.name}: move to {target} timed out") + time.sleep(axis._poll_interval) # noqa: SLF001 + + def _axis_for(self, axis: Axis) -> SmargonAxis: + for ax in self._all_axes(): + if ax._axis is axis: # noqa: SLF001 + return ax + raise KeyError(axis) + + def nudge_beamline(self, modify_offset: bool = False, **deltas: float) -> None: + """Nudge the sample along beamline axes (``BX``/``BY``/``BZ``). + + With ``modify_offset=True`` the O-vector is nudged (``/nudgeOBCS``). + """ + self._ready_guard(0.0) + parsed = {BeamlineAxis(k): float(v) for k, v in deltas.items()} + with self._transport_lock: + self.transport.nudge_beamline(parsed, modify_offset=modify_offset) + + def motor_positions(self) -> dict: + """Return the raw ``q1..q6`` motor positions (read-only diagnostics).""" + return self._get_info().motor_positions + + def transport_info(self) -> dict: + """Return connection/identity info for diagnostics.""" + info = self._get_info() if self.transport.connected else None + return { + "connected": self.transport.connected, + "simulation": self._simulation, + "has_omega": self._has_omega, + "mode": info.mode.name if info else None, + "serial_number": getattr(info, "serial_number", None), + "seq": getattr(info, "seq", None), + } diff --git a/csaxs_bec/devices/smargon/transport/__init__.py b/csaxs_bec/devices/smargon/transport/__init__.py new file mode 100644 index 00000000..1dd3cf7e --- /dev/null +++ b/csaxs_bec/devices/smargon/transport/__init__.py @@ -0,0 +1,47 @@ +"""Control-plane transport for the SmarGon goniometer (via smargopolo REST API). + +Pure standard-library; imports neither ophyd nor BEC, so the transport is unit +testable without hardware and standalone control works with the BEC stack absent. +""" + +from __future__ import annotations + +from .base import ( + AXIS_INFO, + MOTOR_CHANNELS, + Axis, + AxisLimits, + BeamlineAxis, + Mode, + SmargopoloInfo, + SmargopoloTransport, +) +from .errors import ( + SmargopoloAxisError, + SmargopoloCommandError, + SmargopoloConnectionError, + SmargopoloModeError, + SmargopoloTimeoutError, + SmargopoloTransportError, +) +from .fake import FakeTransport +from .rest import RestTransport + +__all__ = [ + "Axis", + "AxisLimits", + "AXIS_INFO", + "BeamlineAxis", + "Mode", + "MOTOR_CHANNELS", + "SmargopoloInfo", + "SmargopoloTransport", + "FakeTransport", + "RestTransport", + "SmargopoloTransportError", + "SmargopoloConnectionError", + "SmargopoloTimeoutError", + "SmargopoloModeError", + "SmargopoloCommandError", + "SmargopoloAxisError", +] diff --git a/csaxs_bec/devices/smargon/transport/base.py b/csaxs_bec/devices/smargon/transport/base.py new file mode 100644 index 00000000..530d42a6 --- /dev/null +++ b/csaxs_bec/devices/smargon/transport/base.py @@ -0,0 +1,217 @@ +"""Transport abstraction for the SmarGon goniometer via the smargopolo server. + +This module is intentionally free of any ophyd / BEC imports so it can be unit +tested without hardware and so the ophyd device depends only on this small +interface. The device talks to the goniometer exclusively through the +:class:`SmargopoloTransport` interface; concrete implementations are: + +* :class:`~csaxs_bec.devices.smargon.transport.rest.RestTransport` -- the real + client against the smargopolo RESTful API (``http://:``), and +* :class:`~csaxs_bec.devices.smargon.transport.fake.FakeTransport` -- an in-memory + simulator used for offline mode and the test-suite. + +Coordinate frames +----------------- +smargopolo runs the kinematics; we drive the *virtual* SmarGon Coordinate System +(SCS) axes, never the underlying q1..q6 motor stages. The frames are: + +* **SCS** -- the user axes: ``SHX SHY SHZ OMEGA CHI PHI OX OY OZ`` (mm and deg). + Settable via ``/targetSCS`` (+ ``/targetSCS_rel``), read via ``/readbackSCS``. +* **MCS** -- the raw motor positions ``q1..q6``, read-only via ``/readbackMCS`` + (diagnostics only; we never command them directly). +* **BCS** -- beamline-frame nudges ``BX BY BZ`` via ``/nudgeBCS`` / ``/nudgeOBCS`` + (relative only; no absolute target or readback). + +Units: linear axes in mm, angular axes in deg. +""" + +from __future__ import annotations + +import abc +import enum +from dataclasses import dataclass, field + + +class Axis(str, enum.Enum): + """The SmarGon Coordinate System (SCS) axes. + + ``OMEGA`` is optional hardware: some SLS SmarGons have a rotation drive, some + do not. The ophyd device enables/disables it per deployment via config; the + transport supports it unconditionally. + """ + + SHX = "SHX" + SHY = "SHY" + SHZ = "SHZ" + OMEGA = "OMEGA" + CHI = "CHI" + PHI = "PHI" + OX = "OX" + OY = "OY" + OZ = "OZ" + + +class BeamlineAxis(str, enum.Enum): + """Beamline Coordinate System (BCS) axes -- nudge-only.""" + + BX = "BX" + BY = "BY" + BZ = "BZ" + + +# Raw motor channels reported by ``/readbackMCS`` (read-only diagnostics). +MOTOR_CHANNELS: tuple[str, ...] = ("q1", "q2", "q3", "q4", "q5", "q6") + + +class Mode(enum.IntEnum): + """The smargopolo overall state machine (``/mode``).""" + + UNINITIALIZED = 0 + REFERENCING = 1 + READY = 2 # "Follow Target" -- the only mode in which moves are honoured + ERROR = 99 + + +@dataclass(frozen=True) +class AxisLimits: + """Inclusive low/high travel limits for an SCS axis, plus its unit. + + Authoritative SCS travel ranges depend on the specific SmarGon build and are + not published in the REST API docs. The defaults below are deliberately + ``None`` (unbounded): per-deployment soft limits should be supplied via the + device config and are enforced in the ophyd layer. ``egu`` is informational. + """ + + egu: str + low: float | None = None + high: float | None = None + + +# Engineering units per SCS axis (linear in mm, angular in deg). Limits are left +# unbounded by default -- see :class:`AxisLimits`. +AXIS_INFO: dict[Axis, AxisLimits] = { + Axis.SHX: AxisLimits(egu="mm"), + Axis.SHY: AxisLimits(egu="mm"), + Axis.SHZ: AxisLimits(egu="mm"), + Axis.OMEGA: AxisLimits(egu="deg"), + Axis.CHI: AxisLimits(egu="deg"), + Axis.PHI: AxisLimits(egu="deg"), + Axis.OX: AxisLimits(egu="mm"), + Axis.OY: AxisLimits(egu="mm"), + Axis.OZ: AxisLimits(egu="deg"), +} + + +@dataclass +class SmargopoloInfo: + """Identity / health snapshot read from ``/readbackMCS``.""" + + mode: Mode = Mode.UNINITIALIZED + serial_number: str = "" + corr_type: int = 0 + seq: int = 0 # loop counter; should advance if the ROS control loop is alive + rosout: str = "" # last message logged on /rosout (carries error detail) + motor_positions: dict[str, float] = field(default_factory=dict) + + +class SmargopoloTransport(abc.ABC): + """Control-plane interface to a SmarGon goniometer via smargopolo. + + A transport owns the HTTP session lifecycle, exposes the ``/mode`` state + machine (including referencing), drives the SCS axes (absolute and relative + targets), reads back SCS positions and raw motor positions, and applies BCS + nudges. + + All methods raise subclasses of :class:`SmargopoloTransportError` on failure. + Implementations must be safe to call from a single controlling thread; the + ophyd device serialises access behind a shared lock. + """ + + # -- session lifecycle ------------------------------------------------- + + @abc.abstractmethod + def connect(self, *, timeout: float | None = None) -> None: + """Open/verify the HTTP session with smargopolo. + + Idempotent. Does **not** reference the goniometer -- referencing is a + deliberate operator action (see :meth:`set_mode`). + """ + + @abc.abstractmethod + def close(self) -> None: + """Release the session. Idempotent and must never raise.""" + + @property + @abc.abstractmethod + def connected(self) -> bool: + """Whether the server is reachable (last contact succeeded).""" + + # -- state machine ----------------------------------------------------- + + @abc.abstractmethod + def get_mode(self) -> Mode: + """Return the current state-machine mode.""" + + @abc.abstractmethod + def set_mode(self, mode: Mode) -> None: + """Set the state-machine mode (``PUT /mode?mode=N``). + + Setting :attr:`Mode.REFERENCING` starts referencing, which physically + moves the goniometer to find its reference marks before transitioning to + :attr:`Mode.READY`. + """ + + @abc.abstractmethod + def get_info(self) -> SmargopoloInfo: + """Return an identity/health snapshot (mode, serial, rosout, motors).""" + + # -- motion ------------------------------------------------------------ + + @abc.abstractmethod + def get_position(self, axis: Axis) -> float: + """Return the current readback of *axis* in its SCS unit.""" + + @abc.abstractmethod + def get_positions(self) -> dict[Axis, float]: + """Return the full SCS readback as ``{Axis: value}``.""" + + @abc.abstractmethod + def move_absolute(self, targets: dict[Axis, float]) -> None: + """Command an absolute SCS move of one or more axes (non-blocking). + + Returns as soon as the target is accepted (smargopolo replies ``202``); + completion is observed by the caller polling :meth:`get_position`. + Requires :attr:`Mode.READY` -- otherwise raises + :class:`~csaxs_bec.devices.smargon.transport.errors.SmargopoloModeError`. + """ + + @abc.abstractmethod + def move_relative(self, deltas: dict[Axis, float]) -> None: + """Command a relative SCS move (``/targetSCS_rel``). Requires READY.""" + + @abc.abstractmethod + def stop(self, axes: list[Axis] | None = None) -> None: + """Halt *axes* (or all SCS axes) by retargeting them to their readback. + + smargopolo exposes no explicit stop endpoint; in ``Mode.READY`` ("Follow + Target") setting the target equal to the current readback brings the axis + to rest. Best-effort: must not raise during teardown. + """ + + # -- diagnostics ------------------------------------------------------- + + @abc.abstractmethod + def get_motor_positions(self) -> dict[str, float]: + """Return the raw ``q1..q6`` motor positions (read-only).""" + + # -- beamline-frame nudges -------------------------------------------- + + @abc.abstractmethod + def nudge_beamline( + self, deltas: dict[BeamlineAxis, float], *, modify_offset: bool = False + ) -> None: + """Nudge the sample along the beamline axes (``/nudgeBCS``). + + With ``modify_offset=True`` the O-vector is nudged instead + (``/nudgeOBCS``). Relative only. Requires READY. + """ diff --git a/csaxs_bec/devices/smargon/transport/errors.py b/csaxs_bec/devices/smargon/transport/errors.py new file mode 100644 index 00000000..8908d6b2 --- /dev/null +++ b/csaxs_bec/devices/smargon/transport/errors.py @@ -0,0 +1,37 @@ +"""Typed exceptions for the smargopolo transport layer. + +Defined in a tiny, dependency-free module so that both the transport +implementations and the ophyd device can import them without pulling in ophyd or +any networking stack. +""" + +from __future__ import annotations + + +class SmargopoloTransportError(Exception): + """Base class for all smargopolo transport errors.""" + + +class SmargopoloConnectionError(SmargopoloTransportError): + """Raised when the transport cannot reach the smargopolo server (HTTP failure).""" + + +class SmargopoloTimeoutError(SmargopoloTransportError): + """Raised when a request to the smargopolo server exceeds its timeout.""" + + +class SmargopoloModeError(SmargopoloTransportError): + """Raised when a command is attempted in the wrong state-machine mode. + + smargopolo only follows position targets in ``Mode.READY``. Commanding motion + while ``UNINITIALIZED``/``REFERENCING``/``ERROR`` raises this so the caller can + surface "reference the goniometer first" rather than silently doing nothing. + """ + + +class SmargopoloCommandError(SmargopoloTransportError): + """Raised when the server rejects a command or returns an error response.""" + + +class SmargopoloAxisError(SmargopoloCommandError): + """Raised when an axis name is unknown or a target is otherwise invalid.""" diff --git a/csaxs_bec/devices/smargon/transport/fake.py b/csaxs_bec/devices/smargon/transport/fake.py new file mode 100644 index 00000000..7fcc1e40 --- /dev/null +++ b/csaxs_bec/devices/smargon/transport/fake.py @@ -0,0 +1,225 @@ +"""In-memory fake smargopolo transport. + +:class:`FakeTransport` implements the full :class:`SmargopoloTransport` interface +with no network access. It backs two things: + +* **offline / simulation mode** -- set ``simulation: true`` in the device config + and the ophyd device loads without a real smargopolo server, and +* **the unit-test suite** -- tests inject it (or assert against it) instead of + touching hardware. + +It models the ``/mode`` state machine: it starts ``UNINITIALIZED`` and refuses +motion until referenced. ``set_mode(REFERENCING)`` transitions to ``READY`` +(instantly by default, or after ``reference_time`` of simulated travel). + +Motion is *instant* by default; pass ``gradual_motion=True`` to advance positions +toward target at a finite speed against the injected clock, for exercising the +device's polling / status path. +""" + +from __future__ import annotations + +from typing import Callable + +from .base import ( + AXIS_INFO, + MOTOR_CHANNELS, + Axis, + BeamlineAxis, + Mode, + SmargopoloInfo, + SmargopoloTransport, +) +from .errors import SmargopoloModeError + +# A plausible home pose (matches the example script's home target). +_DEFAULT_POSITIONS: dict[Axis, float] = { + Axis.SHX: 0.0, + Axis.SHY: 0.0, + Axis.SHZ: 18.0, + Axis.OMEGA: 0.0, + Axis.CHI: 0.0, + Axis.PHI: 0.0, + Axis.OX: 0.0, + Axis.OY: 0.0, + Axis.OZ: 180.0, +} + +# Default move speed (axis-units/s) when ``gradual_motion`` is on. +_DEFAULT_SPEED = 50.0 + + +class FakeTransport(SmargopoloTransport): + """A deterministic, hardware-free :class:`SmargopoloTransport` implementation.""" + + def __init__( + self, + *, + gradual_motion: bool = False, + reference_time: float = 0.0, + serial_number: str = "MCS2-FAKE0001", + time_func: Callable[[], float] | None = None, + ) -> None: + import time as _time + + self._time = time_func or _time.monotonic + self._gradual = gradual_motion + self._reference_time = float(reference_time) + self._serial = serial_number + + self._connected = False + self._mode = Mode.UNINITIALIZED + self._seq = 0 + self._rosout = "" + + self._positions = dict(_DEFAULT_POSITIONS) + self._targets = dict(_DEFAULT_POSITIONS) + self._move_start_pos: dict[Axis, float] = {} + self._move_start_t: dict[Axis, float] = {} + self._reference_done_t: float | None = None + + # Test hooks: count calls and optionally inject failures. + self.call_log: list[str] = [] + self.fail_next: dict[str, Exception] = {} + + # -- internal helpers -------------------------------------------------- + + def _record(self, name: str) -> None: + self._seq += 1 + self.call_log.append(name) + exc = self.fail_next.pop(name, None) + if exc is not None: + raise exc + + def _settle(self, axis: Axis) -> None: + """Advance gradual-motion state for *axis* to 'now'.""" + if axis not in self._move_start_t: + return + elapsed = max(0.0, self._time() - self._move_start_t[axis]) + start = self._move_start_pos[axis] + target = self._targets[axis] + travelled = _DEFAULT_SPEED * elapsed + if travelled >= abs(target - start): + self._positions[axis] = target + self._move_start_t.pop(axis, None) + self._move_start_pos.pop(axis, None) + else: + direction = 1.0 if target >= start else -1.0 + self._positions[axis] = start + direction * travelled + + def _settle_referencing(self) -> None: + if self._mode is not Mode.REFERENCING or self._reference_done_t is None: + return + if self._time() >= self._reference_done_t: + self._mode = Mode.READY + self._reference_done_t = None + + def _require_ready(self) -> None: + self._settle_referencing() + if self._mode is not Mode.READY: + raise SmargopoloModeError( + f"smargopolo is in {self._mode.name} (mode {int(self._mode)}); reference the " + "goniometer (set Mode.READY) before commanding motion" + ) + + def _start_move(self, axis: Axis, value: float) -> None: + self._targets[axis] = value + if not self._gradual: + self._positions[axis] = value + return + self._move_start_pos[axis] = self._positions[axis] + self._move_start_t[axis] = self._time() + + # -- session lifecycle ------------------------------------------------- + + def connect(self, *, timeout: float | None = None) -> None: + self._record("connect") + self._connected = True + + def close(self) -> None: + self._record("close") + self._connected = False + + @property + def connected(self) -> bool: + return self._connected + + # -- state machine ----------------------------------------------------- + + def get_mode(self) -> Mode: + self._settle_referencing() + return self._mode + + def set_mode(self, mode: Mode) -> None: + self._record(f"set_mode:{Mode(mode).name}") + mode = Mode(mode) + if mode is Mode.REFERENCING: + if self._reference_time > 0 and self._gradual: + self._mode = Mode.REFERENCING + self._reference_done_t = self._time() + self._reference_time + else: + self._mode = Mode.READY # instant referencing + else: + self._mode = mode + + def get_info(self) -> SmargopoloInfo: + return SmargopoloInfo( + mode=self.get_mode(), + serial_number=self._serial, + corr_type=1, + seq=self._seq, + rosout=self._rosout, + motor_positions=self.get_motor_positions(), + ) + + # -- motion ------------------------------------------------------------ + + def get_position(self, axis: Axis) -> float: + axis = Axis(axis) + if self._gradual: + self._settle(axis) + return self._positions[axis] + + def get_positions(self) -> dict[Axis, float]: + return {a: self.get_position(a) for a in Axis} + + def move_absolute(self, targets: dict[Axis, float]) -> None: + self._record("move_absolute") + self._require_ready() + for axis, value in targets.items(): + self._start_move(Axis(axis), float(value)) + + def move_relative(self, deltas: dict[Axis, float]) -> None: + self._record("move_relative") + self._require_ready() + for axis, delta in deltas.items(): + axis = Axis(axis) + self._start_move(axis, self.get_position(axis) + float(delta)) + + def stop(self, axes: list[Axis] | None = None) -> None: + self._record("stop") + chosen = list(axes) if axes is not None else list(Axis) + for axis in chosen: + axis = Axis(axis) + if self._gradual: + self._settle(axis) + self._targets[axis] = self._positions[axis] + self._move_start_t.pop(axis, None) + + # -- diagnostics ------------------------------------------------------- + + def get_motor_positions(self) -> dict[str, float]: + # No real kinematics in the fake; surface a stable, inspectable mapping. + return {q: 0.0 for q in MOTOR_CHANNELS} + + # -- beamline-frame nudges -------------------------------------------- + + def nudge_beamline( + self, deltas: dict[BeamlineAxis, float], *, modify_offset: bool = False + ) -> None: + self._record("nudge_obcs" if modify_offset else "nudge_bcs") + self._require_ready() + # The fake does not model the BCS->SCS kinematics; nudges are accepted as + # no-ops on the SCS readback so callers can exercise the code path. + for _axis in deltas: + BeamlineAxis(_axis) # validate the axis name diff --git a/csaxs_bec/devices/smargon/transport/rest.py b/csaxs_bec/devices/smargon/transport/rest.py new file mode 100644 index 00000000..507adffc --- /dev/null +++ b/csaxs_bec/devices/smargon/transport/rest.py @@ -0,0 +1,212 @@ +"""Real smargopolo RESTful-API transport. + +Standard-library only (``urllib``); imports neither ophyd nor any third-party +networking stack, mirroring the rest of the transport layer. Talks to the +smargopolo server documented in ``projects/smargon/resources`` -- e.g.:: + + GET http://smargopolo:3000/readbackSCS + PUT http://smargopolo:3000/targetSCS?SHX=0.1&SHZ=18 + PUT http://smargopolo:3000/mode?mode=1 + +PUTs return ``202 Accepted`` with the updated target (not a motion confirmation); +move completion is observed by the device polling :meth:`get_position`. +""" + +from __future__ import annotations + +import json +import socket +import urllib.error +import urllib.parse +import urllib.request + +from .base import ( + AXIS_INFO, + MOTOR_CHANNELS, + Axis, + BeamlineAxis, + Mode, + SmargopoloInfo, + SmargopoloTransport, +) +from .errors import ( + SmargopoloCommandError, + SmargopoloConnectionError, + SmargopoloModeError, + SmargopoloTimeoutError, +) + + +class RestTransport(SmargopoloTransport): + """:class:`SmargopoloTransport` backed by the smargopolo HTTP REST API.""" + + def __init__(self, *, host: str = "smargopolo", port: int = 3000, timeout: float = 5.0) -> None: + self._host = host + self._port = int(port) + self._timeout = float(timeout) + self._base = f"http://{host}:{self._port}" + self._connected = False + + # -- HTTP helpers ------------------------------------------------------ + + def _url(self, resource: str, params: dict | None = None) -> str: + url = f"{self._base}/{resource.lstrip('/')}" + if params: + # smargopolo expects bare ``?A=1&B=2`` query strings. + url += "?" + urllib.parse.urlencode(params) + return url + + def _request(self, resource: str, *, method: str = "GET", params: dict | None = None) -> dict: + url = self._url(resource, params) + req = urllib.request.Request(url, method=method) + try: + with urllib.request.urlopen(req, timeout=self._timeout) as resp: + body = resp.read().decode("utf-8") + except urllib.error.HTTPError as exc: + self._connected = True # we reached the server; it returned an error + raise SmargopoloCommandError( + f"{method} {url} -> HTTP {exc.code}: {exc.reason}" + ) from exc + except socket.timeout as exc: + self._connected = False + raise SmargopoloTimeoutError( + f"{method} {url} timed out after {self._timeout}s" + ) from exc + except urllib.error.URLError as exc: + self._connected = False + raise SmargopoloConnectionError(f"{method} {url} failed: {exc.reason}") from exc + self._connected = True + if not body: + return {} + try: + return json.loads(body) + except json.JSONDecodeError as exc: + raise SmargopoloCommandError(f"{method} {url} returned non-JSON: {body!r}") from exc + + def _get(self, resource: str) -> dict: + return self._request(resource, method="GET") + + def _put(self, resource: str, params: dict) -> dict: + return self._request(resource, method="PUT", params=params) + + # -- session lifecycle ------------------------------------------------- + + def connect(self, *, timeout: float | None = None) -> None: + # A successful readback confirms the server is up and the loop is running. + self._get("readbackMCS") + self._connected = True + + def close(self) -> None: + self._connected = False + + @property + def connected(self) -> bool: + return self._connected + + # -- state machine ----------------------------------------------------- + + def get_mode(self) -> Mode: + data = self._get("readbackMCS") + return self._coerce_mode(data.get("mode")) + + def set_mode(self, mode: Mode) -> None: + self._put("mode", {"mode": int(mode)}) + + def get_info(self) -> SmargopoloInfo: + data = self._get("readbackMCS") + rosout = data.get("rosout") or {} + position = data.get("position") or {} + return SmargopoloInfo( + mode=self._coerce_mode(data.get("mode")), + serial_number=data.get("serial_number", "") or "", + corr_type=int(data.get("corr_type", 0) or 0), + seq=int(data.get("seq", 0) or 0), + rosout=rosout.get("msg", "") or "", + motor_positions={q: float(position.get(q, 0.0) or 0.0) for q in MOTOR_CHANNELS}, + ) + + @staticmethod + def _coerce_mode(value) -> Mode: + try: + return Mode(int(value)) + except (TypeError, ValueError): + return Mode.ERROR + + # -- motion ------------------------------------------------------------ + + def get_position(self, axis: Axis) -> float: + return self.get_positions()[Axis(axis)] + + def get_positions(self) -> dict[Axis, float]: + data = self._get("readbackSCS") + out: dict[Axis, float] = {} + for axis in Axis: + if axis.value in data: + out[axis] = float(data[axis.value]) + return out + + def _require_ready(self) -> None: + mode = self.get_mode() + if mode is not Mode.READY: + raise SmargopoloModeError( + f"smargopolo is in {mode.name} (mode {int(mode)}); reference the goniometer " + "(set Mode.READY) before commanding motion" + ) + + def move_absolute(self, targets: dict[Axis, float]) -> None: + if not targets: + return + self._require_ready() + self._put("targetSCS", {Axis(a).value: float(v) for a, v in targets.items()}) + + def move_relative(self, deltas: dict[Axis, float]) -> None: + if not deltas: + return + self._require_ready() + self._put("targetSCS_rel", {Axis(a).value: float(v) for a, v in deltas.items()}) + + def stop(self, axes: list[Axis] | None = None) -> None: + # Retarget the chosen axes to their current readback: in "Follow Target" + # mode this halts them. Best-effort -- never raise. + try: + positions = self.get_positions() + except Exception: # pylint: disable=broad-except + return + chosen = list(axes) if axes is not None else list(positions) + targets = {a: positions[a] for a in chosen if a in positions} + if not targets: + return + try: + self._put("targetSCS", {Axis(a).value: float(v) for a, v in targets.items()}) + except Exception: # pylint: disable=broad-except + pass + + # -- diagnostics ------------------------------------------------------- + + def get_motor_positions(self) -> dict[str, float]: + return self.get_info().motor_positions + + # -- beamline-frame nudges -------------------------------------------- + + def nudge_beamline( + self, deltas: dict[BeamlineAxis, float], *, modify_offset: bool = False + ) -> None: + if not deltas: + return + self._require_ready() + resource = "nudgeOBCS" if modify_offset else "nudgeBCS" + self._put(resource, {BeamlineAxis(a).value: float(v) for a, v in deltas.items()}) + + # -- introspection ----------------------------------------------------- + + @property + def host(self) -> str: + return self._host + + @property + def port(self) -> int: + return self._port + + +# Re-exported for callers that want axis units without importing base directly. +__all__ = ["RestTransport", "AXIS_INFO"] diff --git a/tests/tests_devices/test_smargon_device.py b/tests/tests_devices/test_smargon_device.py new file mode 100644 index 00000000..64a1eda8 --- /dev/null +++ b/tests/tests_devices/test_smargon_device.py @@ -0,0 +1,162 @@ +"""Tests for the SmarGon ophyd device (simulation mode, no hardware).""" + +import pytest +from ophyd import Kind +from ophyd.utils import LimitError + +from csaxs_bec.devices.smargon import SmarGon +from csaxs_bec.devices.smargon.transport import FakeTransport, Mode +from csaxs_bec.devices.smargon.transport.errors import SmargopoloCommandError, SmargopoloModeError + + +@pytest.fixture +def smargon(): + device = SmarGon(name="smargon_test", simulation=True, has_omega=True) + device.wait_for_connection() + yield device + device.destroy() + + +def test_uses_fake_transport(smargon): + assert isinstance(smargon.transport, FakeTransport) + assert smargon.transport.connected + + +def test_v1_axes_are_positioners(smargon): + for ax in (smargon.shx, smargon.shy, smargon.shz, smargon.chi, smargon.phi): + assert hasattr(ax, "move") and hasattr(ax, "stop") + assert ax.kind != Kind.omitted + + +def test_refuses_move_until_referenced(smargon): + assert smargon.get_mode() == "UNINITIALIZED" + with pytest.raises(SmargopoloModeError): + smargon.shx.move(0.5) + + +def test_reference_then_move(smargon): + smargon.reference() + assert smargon.is_ready() + status = smargon.shx.move(0.5, wait=True) + assert status.done and status.success + assert smargon.shx.position == pytest.approx(0.5) + + +def test_set_alias_is_nonblocking(smargon): + smargon.reference() + status = smargon.chi.set(12.0) + status.wait(timeout=5) + assert status.success + assert smargon.chi.position == pytest.approx(12.0) + + +def test_omega_enabled_when_has_omega(smargon): + assert smargon.omega.kind != Kind.omitted + smargon.reference() + smargon.omega.move(45.0, wait=True) + assert smargon.omega.position == pytest.approx(45.0) + + +def test_omega_disabled_without_has_omega(): + device = SmarGon(name="no_omega", simulation=True, has_omega=False) + device.wait_for_connection() + try: + device.reference() + assert device.omega.kind == Kind.omitted + with pytest.raises(SmargopoloModeError): + device.omega.move(45.0) + finally: + device.destroy() + + +def test_soft_limits_enforced(): + device = SmarGon(name="lim", simulation=True, limits={"SHX": [-1.0, 1.0]}) + device.wait_for_connection() + try: + device.reference() + assert device.shx.limits == (-1.0, 1.0) + with pytest.raises(LimitError): + device.shx.move(2.0) + finally: + device.destroy() + + +def test_per_axis_tolerance_applied(): + device = SmarGon(name="tol", simulation=True, tolerance={"SHX": 0.05}) + device.wait_for_connection() + try: + assert device.shx._tolerance == pytest.approx(0.05) + finally: + device.destroy() + + +def test_move_scs_coordinated(smargon): + smargon.reference() + smargon.move_scs(SHX=0.3, SHZ=19.0, wait=True) + assert smargon.shx.position == pytest.approx(0.3) + assert smargon.shz.position == pytest.approx(19.0) + + +def test_stop_all_does_not_raise(smargon): + smargon.reference() + smargon.shx.set(0.2) + smargon.stop_all() + + +def test_motor_positions_diagnostic(smargon): + motors = smargon.motor_positions() + assert set(motors) == {"q1", "q2", "q3", "q4", "q5", "q6"} + + +def test_transport_info(smargon): + info = smargon.transport_info() + assert info["simulation"] is True + assert info["has_omega"] is True + assert info["connected"] is True + + +def test_move_fails_when_target_rejected_up_front(smargon): + """smargopolo refusing an out-of-range target -> immediate move failure.""" + smargon.reference() + smargon.transport.fail_next["move_absolute"] = SmargopoloCommandError("target out of range") + status = smargon.shx.set(0.5) + with pytest.raises(SmargopoloCommandError): + status.wait(timeout=5) + assert not status.success + + +def test_move_aborts_when_controller_faults_mid_move(): + """Target accepted (202) but a coupled-limit fault appears after partial motion.""" + device = SmarGon(name="faulter", simulation=True, gradual_motion=True) + device.wait_for_connection() + try: + device.reference() + status = device.shx.set(1e6) # far target: gradual motion keeps it in-flight + # Simulate smargopolo hitting a coupled hardware limit mid-move. + device.transport._mode = Mode.ERROR + with pytest.raises(SmargopoloModeError): + status.wait(timeout=5) + assert not status.success + assert device.shx.motor_is_moving.get() == 0 + finally: + device.destroy() + + +def test_move_scs_aborts_on_controller_fault(): + device = SmarGon(name="faulter2", simulation=True, gradual_motion=True) + device.wait_for_connection() + try: + device.reference() + device.transport._mode = Mode.ERROR + with pytest.raises(SmargopoloModeError): + device.move_scs(SHX=1e6, wait=True) + finally: + device.destroy() + + +def test_mode_signal_is_readonly(smargon): + from ophyd.utils import ReadOnlyError + + assert smargon.mode.get() == int(Mode.UNINITIALIZED) + with pytest.raises(ReadOnlyError): + smargon.mode.put(2) diff --git a/tests/tests_devices/test_smargon_transport.py b/tests/tests_devices/test_smargon_transport.py new file mode 100644 index 00000000..8aba62c6 --- /dev/null +++ b/tests/tests_devices/test_smargon_transport.py @@ -0,0 +1,82 @@ +"""Tests for the ophyd-free smargopolo transport layer (FakeTransport).""" + +import pytest + +from csaxs_bec.devices.smargon.transport import Axis, BeamlineAxis, FakeTransport, Mode +from csaxs_bec.devices.smargon.transport.errors import SmargopoloModeError + + +@pytest.fixture +def transport(): + t = FakeTransport() + t.connect() + return t + + +def test_starts_uninitialized_and_refuses_motion(transport): + assert transport.get_mode() is Mode.UNINITIALIZED + with pytest.raises(SmargopoloModeError): + transport.move_absolute({Axis.SHX: 0.5}) + + +def test_referencing_transitions_to_ready(transport): + transport.set_mode(Mode.REFERENCING) + assert transport.get_mode() is Mode.READY + + +def test_absolute_move_updates_readback(transport): + transport.set_mode(Mode.REFERENCING) + transport.move_absolute({Axis.SHX: 0.5, Axis.CHI: 10.0}) + assert transport.get_position(Axis.SHX) == pytest.approx(0.5) + assert transport.get_position(Axis.CHI) == pytest.approx(10.0) + + +def test_relative_move(transport): + transport.set_mode(Mode.REFERENCING) + start = transport.get_position(Axis.SHY) + transport.move_relative({Axis.SHY: 0.2}) + assert transport.get_position(Axis.SHY) == pytest.approx(start + 0.2) + + +def test_get_positions_covers_all_scs_axes(transport): + positions = transport.get_positions() + assert set(positions) == set(Axis) + + +def test_stop_retargets_to_readback(transport): + transport.set_mode(Mode.REFERENCING) + transport.move_absolute({Axis.SHZ: 20.0}) + transport.stop([Axis.SHZ]) + assert transport.get_position(Axis.SHZ) == pytest.approx(20.0) + + +def test_info_reports_serial_and_motors(transport): + info = transport.get_info() + assert info.serial_number == "MCS2-FAKE0001" + assert set(info.motor_positions) == {"q1", "q2", "q3", "q4", "q5", "q6"} + + +def test_seq_advances_on_commands(transport): + first = transport.get_info().seq + transport.set_mode(Mode.REFERENCING) + transport.move_absolute({Axis.SHX: 0.1}) + assert transport.get_info().seq > first + + +def test_nudge_requires_ready(transport): + with pytest.raises(SmargopoloModeError): + transport.nudge_beamline({BeamlineAxis.BX: 0.01}) + transport.set_mode(Mode.REFERENCING) + transport.nudge_beamline({BeamlineAxis.BX: 0.01}) # no raise once READY + + +def test_gradual_motion_advances_over_time(): + clock = {"t": 0.0} + t = FakeTransport(gradual_motion=True, time_func=lambda: clock["t"]) + t.connect() + t.set_mode(Mode.REFERENCING) + t.move_absolute({Axis.SHX: 100.0}) # 50 units/s default speed + clock["t"] = 1.0 + assert t.get_position(Axis.SHX) == pytest.approx(50.0) + clock["t"] = 5.0 + assert t.get_position(Axis.SHX) == pytest.approx(100.0)