diff --git a/csaxs_bec/device_configs/smargon.yaml b/csaxs_bec/device_configs/smargon.yaml new file mode 100644 index 00000000..fe004d1e --- /dev/null +++ b/csaxs_bec/device_configs/smargon.yaml @@ -0,0 +1,86 @@ +# Example BEC device-config entries for the SmarGon goniometer (via smargopolo). +# +# Per-axis architecture (like the smaract motors): each SCS axis is its own top-level +# motor sharing ONE SmargopoloController (singleton per host:port), plus a thin +# `smargon` coordinator for whole-goniometer ops. All entries must use the SAME +# host/port so they share the controller. +# +# The motor NAME (the top-level key: sgx, sgchi, ...) is yours to choose. The +# `scs_axis` field binds it to the fixed smargopolo axis (SHX/SHY/SHZ/CHI/PHI/OMEGA). +# +# Referencing is a deliberate operator action: after loading, run once +# dev.smargon.reference() # (or dev.sgx.controller.reference()) + +# --- Coordinator (whole-goniometer ops: reference/mode/move_scs/diagnostics) ------ +smargon: + description: SmarGon coordinator (referencing, mode, coordinated moves) + deviceClass: csaxs_bec.devices.smargon.smargon.SmarGon + deviceConfig: + host: tt-smargopolo # cSAXS beamline smargopolo REST API + port: 3000 # REST API port (per maintainer). NB: :8080 is the web GUI, not the API + timeout: 5.0 + deviceTags: [goniometer, smargon] + enabled: true + onFailure: retry + readOnly: false + readoutPriority: baseline + softwareTrigger: false + +# --- SCS axis motors -------------------------------------------------------------- +# Limits from the MX-lab reference client (resources_2/Smargon.py) -- VERIFY per build. +# NOTE: SHZ is not centred on 0 (home ~18); PHI is continuous (client wraps to +# [-180,180) -- open question whether smargopolo wraps server-side). +sgx: + description: SmarGon SHX (sample X) + deviceClass: csaxs_bec.devices.smargon.positioner.SmargopoloMotor + deviceConfig: {scs_axis: SHX, host: tt-smargopolo, port: 3000, limits: [-2, 2], tolerance: 0.001} + deviceTags: [goniometer, smargon] + enabled: true + onFailure: retry + readoutPriority: baseline + +sgy: + description: SmarGon SHY (sample Y) + deviceClass: csaxs_bec.devices.smargon.positioner.SmargopoloMotor + deviceConfig: {scs_axis: SHY, host: tt-smargopolo, port: 3000, limits: [-2, 2], tolerance: 0.001} + deviceTags: [goniometer, smargon] + enabled: true + onFailure: retry + readoutPriority: baseline + +sgz: + description: SmarGon SHZ (sample Z) + deviceClass: csaxs_bec.devices.smargon.positioner.SmargopoloMotor + deviceConfig: {scs_axis: SHZ, host: tt-smargopolo, port: 3000, limits: [10, 22], tolerance: 0.001} + deviceTags: [goniometer, smargon] + enabled: true + onFailure: retry + readoutPriority: baseline + +sgchi: + description: SmarGon CHI (goniometer tilt) + deviceClass: csaxs_bec.devices.smargon.positioner.SmargopoloMotor + deviceConfig: {scs_axis: CHI, host: tt-smargopolo, port: 3000, limits: [0, 65], tolerance: 0.01} + deviceTags: [goniometer, smargon] + enabled: true + onFailure: retry + readoutPriority: baseline + +sgphi: + description: SmarGon PHI (goniometer rotation) + deviceClass: csaxs_bec.devices.smargon.positioner.SmargopoloMotor + deviceConfig: {scs_axis: PHI, host: tt-smargopolo, port: 3000, limits: [-180, 180], tolerance: 0.01} + deviceTags: [goniometer, smargon] + enabled: true + onFailure: retry + readoutPriority: baseline + +# --- Optional OMEGA rotation drive (only on builds that have it) ------------------- +# sgomega: +# description: SmarGon OMEGA (rotation drive) +# deviceClass: csaxs_bec.devices.smargon.positioner.SmargopoloMotor +# deviceConfig: {scs_axis: OMEGA, host: tt-smargopolo, port: 3000, tolerance: 0.01} +# deviceTags: [goniometer, smargon] +# enabled: true +# onFailure: retry +# readoutPriority: baseline diff --git a/csaxs_bec/device_configs/smargon_sim_session.yaml b/csaxs_bec/device_configs/smargon_sim_session.yaml new file mode 100644 index 00000000..d04fed72 --- /dev/null +++ b/csaxs_bec/device_configs/smargon_sim_session.yaml @@ -0,0 +1,124 @@ +# Self-contained laptop test session for the SmarGon device (SIMULATION). +# +# Per-axis motors (dev.sgx, dev.sgy, dev.sgz, dev.sgchi, dev.sgphi, dev.sgomega) + +# a `smargon` coordinator, all sharing one simulated controller, plus a sim monitor +# so scans record data. No smargopolo server / hardware needed. +# +# Load in the client: +# bec.config.update_session_with_file( +# '/Users/menzel/BEC_development/stack/csaxs_bec-smargon/csaxs_bec/device_configs/smargon_sim_session.yaml') +# Then: dev.smargon.reference(); umv(dev.sgx, 0.5); dev.wm('sg*') +# +# NOTE: update_session_with_file REPLACES the current session. PREREQUISITE: the +# running BEC must import csaxs_bec.devices.smargon (repoint the editable install at +# the smargon worktree; see RUNNING_BEC.md). + +smargon: + description: SmarGon coordinator (SIMULATION) + deviceClass: csaxs_bec.devices.smargon.smargon.SmarGon + deviceConfig: { simulation: true, gradual_motion: true } + deviceTags: [goniometer, smargon, simulation] + enabled: true + onFailure: retry + readoutPriority: baseline + +sgx: + description: SmarGon SHX (sim) + deviceClass: csaxs_bec.devices.smargon.positioner.SmargopoloMotor + deviceConfig: + { + scs_axis: SHX, + simulation: true, + gradual_motion: true, + limits: [-2, 2], + tolerance: 0.001, + } + deviceTags: [goniometer, smargon, simulation] + enabled: true + onFailure: retry + readoutPriority: baseline + +sgy: + description: SmarGon SHY (sim) + deviceClass: csaxs_bec.devices.smargon.positioner.SmargopoloMotor + deviceConfig: + { + scs_axis: SHY, + simulation: true, + gradual_motion: true, + limits: [-2, 2], + tolerance: 0.001, + } + deviceTags: [goniometer, smargon, simulation] + enabled: true + onFailure: retry + readoutPriority: baseline + +sgz: + description: SmarGon SHZ (sim) + deviceClass: csaxs_bec.devices.smargon.positioner.SmargopoloMotor + deviceConfig: + { + scs_axis: SHZ, + simulation: true, + gradual_motion: true, + limits: [10, 22], + tolerance: 0.001, + } + deviceTags: [goniometer, smargon, simulation] + enabled: true + onFailure: retry + readoutPriority: baseline + +sgchi: + description: SmarGon CHI (sim) + deviceClass: csaxs_bec.devices.smargon.positioner.SmargopoloMotor + deviceConfig: + { + scs_axis: CHI, + simulation: true, + gradual_motion: true, + limits: [0, 65], + tolerance: 0.01, + } + deviceTags: [goniometer, smargon, simulation] + enabled: true + onFailure: retry + readoutPriority: baseline + +sgphi: + description: SmarGon PHI (sim) + deviceClass: csaxs_bec.devices.smargon.positioner.SmargopoloMotor + deviceConfig: + { + scs_axis: PHI, + simulation: true, + gradual_motion: true, + limits: [-180, 180], + tolerance: 0.01, + } + deviceTags: [goniometer, smargon, simulation] + enabled: true + onFailure: retry + readoutPriority: baseline + +# sgomega: +# description: SmarGon OMEGA (sim) +# deviceClass: csaxs_bec.devices.smargon.positioner.SmargopoloMotor +# deviceConfig: {scs_axis: OMEGA, simulation: true, gradual_motion: true, tolerance: 0.01} +# deviceTags: [goniometer, smargon, simulation] +# enabled: true +# onFailure: retry +# readoutPriority: baseline + +sim_mon: + description: Simulated 1D monitor (readout for test scans over SmarGon axes) + deviceClass: ophyd_devices.SimMonitor + deviceConfig: + sim_init: + model: "GaussianModel" + deviceTags: [monitor, simulation] + enabled: true + onFailure: retry + readoutPriority: monitored + softwareTrigger: false diff --git a/csaxs_bec/devices/smargon/README.md b/csaxs_bec/devices/smargon/README.md new file mode 100644 index 00000000..53a74a2b --- /dev/null +++ b/csaxs_bec/devices/smargon/README.md @@ -0,0 +1,113 @@ +# SmarGon goniometer (BEC integration via smargopolo) + +BEC integration 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). + +## Architecture (per-axis, like the `smaract` device) + +Each SCS axis is its **own top-level BEC motor** sharing a **singleton +`SmargopoloController`** (one per `host:port`), plus a thin `SmarGon` **coordinator** +for whole-goniometer operations. This makes `dev.wm`, `umv`, limits and the GUI motor +widgets work natively per axis. + +``` +smargon/ + transport/ # ophyd-free control plane (unit-testable, no BEC needed) + base.py # SmargopoloTransport ABC + Axis/BeamlineAxis/Mode + AXIS_INFO + rest.py # RestTransport -- real urllib client against :3000 + fake.py # FakeTransport -- in-memory sim (offline mode + tests) + errors.py # typed exceptions + controller.py # SmargopoloController(OphydObject) -- singleton per host:port; + # owns transport + lock + mode; referencing / move_scs / nudge / q1..q6 + positioner.py # SmargopoloMotor(Device, PositionerBase) -- one SCS axis, top-level motor + smargon.py # SmarGon(PSIDeviceBase) -- thin coordinator (dev.smargon), delegates +``` + +The `transport/` layer is shared with (and unchanged from) the original design and is +structured like the Canon CR-N300 device (HTTP transport + in-memory fake + threaded +poll-to-tolerance moves), since both wrap a non-EPICS, poll-to-confirm HTTP backend. + +## Devices & naming + +- **One motor per axis**, top-level: the YAML key is the BEC name (**you choose it**: + `sgx`, `sgchi`, …); a `scs_axis:` field binds it to the fixed smargopolo axis name + (`SHX/SHY/SHZ/CHI/PHI/OMEGA`). OMEGA is optional hardware — add a motor entry only on + builds that have it. +- **`dev.smargon`** — the coordinator: `reference()`, `get_mode()`, `is_ready()`, + `wait_until_ready()`, `move_scs(...)`, `nudge_beamline(...)`, `motor_positions()`, + `stop_all()`, `transport_info()`, plus read-only `mode`/`rosout` signals. +- The same whole-goniometer ops are reachable from any axis as + `dev..controller.()` (the motors expose `USER_ACCESS = ["controller"]`). + +Because the motors expose `low_limit_travel`/`high_limit_travel` signals and alias +their readback to the device name, `dev.wm(dev.sgchi)` shows real readback/limits +(`low == high` means unbounded, EpicsMotor convention). + +## 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 +dev.smargon.get_mode() # 'UNINITIALIZED' on a fresh server +dev.smargon.reference() # PUT /mode?mode=1, waits until READY (moves hardware!) +umv(dev.sgx, 0.5) # now honoured +``` + +Any axis move while not READY raises `SmargopoloModeError` with a "reference first" +message. `dev.smargon.mode` / `.rosout` surface the state machine and last ROS message. + +## Usage + +```python +umv(dev.sgx, 0.5) # single-axis move (mm) +dev.wm('sg*') # readback/setpoint/limits table +dev.smargon.move_scs(SHX=0.1, SHZ=18) # coordinated move in one /targetSCS call +dev.smargon.nudge_beamline(BX=0.01) # beamline-frame nudge (/nudgeBCS) +scans.grid_scan(dev.sgx, -1, 1, 11, dev.sgchi, 0, 60, 7, exp_time=0.1, relative=False) +``` + +## Configuration + +See `csaxs_bec/device_configs/smargon.yaml` (real) and `smargon_sim_session.yaml` +(laptop simulation). All entries for one goniometer must share `host`/`port` so they +share the controller. Motor `deviceConfig` keys: `scs_axis`, `host`, `port`, `limits` +`[low, high]` (**verify per build**), `tolerance`, `poll_interval`, `move_timeout`. +Simulation: `simulation: true` (+ optional `gradual_motion: true`) uses `FakeTransport`, +no server required. + +## 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 axes (like a hexapod) and can't be reliably +declared in advance. smargopolo owns the real limits. Both failure modes are handled: + +- **Rejected up front** — 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` after partial motion, the move waiters detect it on the next poll and + fail the status with the `rosout` detail (see `SmargopoloController.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). + +## 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. + +## 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. +- **PHI wrap**: the MX-lab reference client wraps PHI to `[-180,180)` before sending. + Confirm whether smargopolo wraps server-side, or whether the motor should. +- *(resolved)* OMEGA is drivable via `/targetSCS` wherever it physically exists. diff --git a/csaxs_bec/devices/smargon/__init__.py b/csaxs_bec/devices/smargon/__init__.py new file mode 100644 index 00000000..da631e22 --- /dev/null +++ b/csaxs_bec/devices/smargon/__init__.py @@ -0,0 +1,37 @@ +"""SmarGon goniometer integration for BEC (via the smargopolo REST API). + +Per-axis architecture (mirrors the ``smaract`` device): each SCS axis is a top-level +:class:`SmargopoloMotor` sharing a singleton :class:`SmargopoloController` per +smargopolo server; :class:`SmarGon` is a thin coordinator (``dev.smargon``) for +whole-goniometer operations (referencing, mode, coordinated moves, diagnostics). + +The control transport (:mod:`.transport`) is ophyd-free and independently importable. +""" + +from __future__ import annotations + +from .controller import SmargopoloController +from .positioner import SmargopoloMotor +from .smargon import SmarGon +from .transport import ( + Axis, + BeamlineAxis, + FakeTransport, + Mode, + RestTransport, + SmargopoloInfo, + SmargopoloTransport, +) + +__all__ = [ + "SmarGon", + "SmargopoloController", + "SmargopoloMotor", + "Axis", + "BeamlineAxis", + "Mode", + "SmargopoloTransport", + "SmargopoloInfo", + "FakeTransport", + "RestTransport", +] diff --git a/csaxs_bec/devices/smargon/controller.py b/csaxs_bec/devices/smargon/controller.py new file mode 100644 index 00000000..d5a7ca88 --- /dev/null +++ b/csaxs_bec/devices/smargon/controller.py @@ -0,0 +1,284 @@ +"""Shared controller for the SmarGon goniometer (smargopolo backend). + +``SmargopoloController`` is a **singleton per ``(host, port, simulation)``** (the same +pattern as :class:`ophyd_devices.utils.controller.Controller`): every +:class:`~csaxs_bec.devices.smargon.positioner.SmargopoloMotor` and the thin +:class:`~csaxs_bec.devices.smargon.smargon.SmarGon` coordinator that point at the same +smargopolo server share **one** controller instance -- one transport, one lock, one +``/mode`` state machine. + +It owns the whole-goniometer operations that don't belong to a single axis: +referencing / the mode state machine, the coordinated multi-axis move +(``/targetSCS`` in one request), beamline nudges, and the raw ``q1..q6`` diagnostics. +All hardware access goes through the injected +:class:`~csaxs_bec.devices.smargon.transport.base.SmargopoloTransport`, serialised +behind ``self._lock``. +""" + +from __future__ import annotations + +import threading +import time + +from ophyd import OphydObject + +from .transport.base import Axis, BeamlineAxis, Mode, SmargopoloTransport +from .transport.errors import SmargopoloModeError +from .transport.fake import FakeTransport +from .transport.rest import RestTransport + +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__) + + +class SmargopoloController(OphydObject): + """One shared control plane for a smargopolo server (singleton per host/port). + + Reachable from any axis as ``dev..controller`` (the motors set + ``USER_ACCESS = ["controller"]``) and mirrored by the ``dev.smargon`` + coordinator. + """ + + _controller_instances: dict[str, "SmargopoloController"] = {} + _initialized = False + + SUB_CONNECTION_CHANGE = "connection_change" + + USER_ACCESS = [ + "reference", + "is_ready", + "wait_until_ready", + "get_mode", + "move_scs", + "nudge_beamline", + "motor_positions", + "stop_all", + "transport_info", + ] + + def __new__(cls, *, host: str = "smargopolo", port: int = 3000, simulation: bool = False, **_): + key = f"{host}:{port}:{'sim' if simulation else 'real'}" + if key not in cls._controller_instances: + cls._controller_instances[key] = object.__new__(cls) + return cls._controller_instances[key] + + def __init__( + self, + *, + host: str = "smargopolo", + port: int = 3000, + simulation: bool = False, + timeout: float = 5.0, + gradual_motion: bool = False, + device_manager=None, + name: str = "smargopolo_controller", + parent=None, + **kwargs, + ) -> None: + if self._initialized: + # Singleton: a later constructor may pass a device_manager the first + # one lacked -- adopt it, but do not rebuild the transport. + if device_manager is not None and getattr(self, "device_manager", None) is None: + self.device_manager = device_manager + return + super().__init__(name=name, parent=parent, **kwargs) + self._lock = threading.RLock() + self._host = host + self._port = port + self._simulation = simulation + self.device_manager = device_manager + if simulation: + self.transport: SmargopoloTransport = FakeTransport(gradual_motion=gradual_motion) + else: + self.transport = RestTransport(host=host, port=port, timeout=timeout) + self._axes: dict[Axis, object] = {} # scs_axis -> SmargopoloMotor (registered by motors) + self._connected = False + self._initialized = True + + # -- axis registry ----------------------------------------------------- + + def register_axis(self, motor) -> None: + """Called by each :class:`SmargopoloMotor` so the controller can find it.""" + self._axes[Axis(motor.scs_axis)] = motor + + # -- connection lifecycle ---------------------------------------------- + + def connect(self, timeout: float = 10.0) -> None: + with self._lock: + if self._connected: + return + self.transport.connect(timeout=timeout) + self._connected = True + + def close(self) -> None: + with self._lock: + try: + self.transport.close() + except Exception as exc: # pylint: disable=broad-except + logger.warning(f"{self.name}: transport close failed: {exc}") + finally: + self._connected = False + + @property + def connected(self) -> bool: + return self._connected + + # -- locked transport passthroughs ------------------------------------- + + def get_position(self, axis: Axis) -> float: + with self._lock: + return self.transport.get_position(Axis(axis)) + + def get_positions(self) -> dict: + with self._lock: + return self.transport.get_positions() + + def move_absolute(self, targets: dict) -> None: + with self._lock: + self.transport.move_absolute({Axis(a): float(v) for a, v in targets.items()}) + + def move_relative(self, deltas: dict) -> None: + with self._lock: + self.transport.move_relative({Axis(a): float(v) for a, v in deltas.items()}) + + def stop(self, axes: list | None = None) -> None: + with self._lock: + self.transport.stop([Axis(a) for a in axes] if axes is not None else None) + + def get_info(self): + with self._lock: + return self.transport.get_info() + + def mode(self) -> Mode: + """Current state-machine mode (enum) -- used internally by axes/coordinator.""" + with self._lock: + return self.transport.get_mode() + + # -- state machine / referencing (whole-goniometer ops) ---------------- + + def require_ready(self) -> None: + """Raise :class:`SmargopoloModeError` unless the goniometer is READY.""" + m = self.mode() + if m is not Mode.READY: + raise SmargopoloModeError( + f"{self.name} is in {m.name} (mode {int(m)}); call reference() and wait for " + "READY before moving" + ) + + def raise_if_error(self) -> None: + """Raise if smargopolo went to ERROR (used to abort in-flight moves). + + A target accepted with ``202`` can still fault after partial motion when a + coupled hardware limit is hit; move waiters poll this so the move fails with + the controller's ``rosout`` detail instead of running to 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'}" + ) + + def reference(self, wait: bool = True, timeout: float = 120.0) -> str: + """Reference the goniometer (``REFERENCING`` -> ``READY``). + + WARNING: this physically moves the goniometer to find its reference marks. + Deliberate operator action. Returns the resulting mode name. + """ + with self._lock: + self.transport.set_mode(Mode.REFERENCING) + logger.info(f"{self.name}: referencing started") + if wait: + m = self.wait_until_ready(timeout=timeout) + logger.info(f"{self.name}: referencing finished -> {m.name}") + return m.name + return self.mode().name + + def is_ready(self) -> bool: + return self.mode() is Mode.READY + + def wait_until_ready(self, timeout: float = 120.0, poll: float = 0.5) -> Mode: + deadline = time.monotonic() + timeout + while True: + m = self.mode() + if m is Mode.READY: + return m + if m 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 {m.name})") + time.sleep(poll) + + def get_mode(self) -> str: + """State-machine mode name, for the user console.""" + return self.mode().name + + # -- coordinated / diagnostic ops -------------------------------------- + + def move_scs(self, wait: bool = True, **targets: float) -> None: + """Move several SCS axes at once in a single ``/targetSCS`` request. + + Keys are SCS axis names, e.g. ``move_scs(SHX=0.1, SHZ=18)``. Each targeted + axis is limit-checked against its registered motor (if any) before sending. + """ + self.require_ready() + parsed = {Axis(k): float(v) for k, v in targets.items()} + for ax, val in parsed.items(): + motor = self._axes.get(ax) + if motor is not None: + motor.check_value(val) + self.move_absolute(parsed) + if wait: + for ax, val in parsed.items(): + self._wait_axis(ax, val) + + def _wait_axis(self, axis: Axis, target: float, timeout: float | None = None) -> None: + motor = self._axes.get(axis) + tol = getattr(motor, "_tolerance", 1e-3) + poll = getattr(motor, "_poll_interval", 0.1) + deadline = time.monotonic() + (timeout or getattr(motor, "_move_timeout", 60.0)) + while True: + pos = self.get_position(axis) + if motor is not None: + motor._update_readback(pos) # noqa: SLF001 + if abs(pos - target) <= tol: + return + self.raise_if_error() + if time.monotonic() > deadline: + raise TimeoutError(f"{axis.value}: move to {target} timed out") + time.sleep(poll) + + def nudge_beamline(self, modify_offset: bool = False, **deltas: float) -> None: + """Nudge the sample along beamline axes (``BX``/``BY``/``BZ``).""" + self.require_ready() + parsed = {BeamlineAxis(k): float(v) for k, v in deltas.items()} + with self._lock: + self.transport.nudge_beamline(parsed, modify_offset=modify_offset) + + def stop_all(self) -> None: + """Halt all SCS axes.""" + self.stop(None) + + def motor_positions(self) -> dict: + """Raw ``q1..q6`` motor positions (read-only diagnostics).""" + return self.get_info().motor_positions + + def transport_info(self) -> dict: + info = self.get_info() if self.connected else None + return { + "connected": self.connected, + "simulation": self._simulation, + "host": self._host, + "port": self._port, + "mode": info.mode.name if info else None, + "serial_number": getattr(info, "serial_number", None), + "seq": getattr(info, "seq", None), + "axes": sorted(a.value for a in self._axes), + } diff --git a/csaxs_bec/devices/smargon/positioner.py b/csaxs_bec/devices/smargon/positioner.py new file mode 100644 index 00000000..032e1079 --- /dev/null +++ b/csaxs_bec/devices/smargon/positioner.py @@ -0,0 +1,238 @@ +"""Per-axis SmarGon motor -- a top-level BEC device for one SCS axis. + +``SmargopoloMotor`` is an independent ophyd positioner (``Device`` + +``PositionerBase``): each SCS axis (SHX/SHY/SHZ/CHI/PHI/OMEGA) is its own top-level +BEC motor, exactly like the ``smaract`` motors. It talks to hardware only through a +shared :class:`~csaxs_bec.devices.smargon.controller.SmargopoloController` (singleton +per smargopolo server), so ``dev.sgx`` and ``dev.sgchi`` pointing at the same host +share one transport, one lock and one mode state machine. + +Two conventions matter for BEC's motor UX to work natively (``dev.wm``, ``umv``, the +GUI motor widgets): + +* the readback signal is aliased to the device name, so ``read()`` yields a key equal + to the device name (what ``dev.wm`` looks up), and +* soft limits are published via ``low_limit_travel`` / ``high_limit_travel`` signals, + which the device server broadcasts as the device's limits (``low == high`` means + unbounded, following the EpicsMotor convention). +""" + +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 .controller import SmargopoloController +from .transport.base import AXIS_INFO, Axis + + +class SmargopoloMotor(Device, PositionerBase): + """One SCS axis of a SmarGon goniometer, as a standalone BEC motor. + + Parameters + ---------- + scs_axis: + Which smargopolo SCS axis this motor drives (``"SHX"``, ``"CHI"``, ...). + host / port / simulation / timeout / gradual_motion: + Select/parameterise the shared :class:`SmargopoloController`. + limits: + Optional ``(low, high)`` soft limits in SCS units. ``low == high`` (the + default ``(0, 0)``) means unbounded -- the controller still enforces the + real coupled hardware limits. + tolerance / poll_interval / move_timeout: + Move-completion tolerance and polling cadence / timeout. + egu: + Engineering unit; defaults to the axis' natural unit (mm or deg). + """ + + 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) + low_limit_travel = Cpt(Signal, value=0.0, kind=Kind.config) + high_limit_travel = Cpt(Signal, value=0.0, kind=Kind.config) + + SUB_READBACK = "readback" + _default_sub = SUB_READBACK + + USER_ACCESS = ["controller"] + + def __init__( + self, + prefix: str = "", + *, + name: str, + scs_axis: str, + host: str = "smargopolo", + port: int = 3000, + simulation: bool = False, + timeout: float = 5.0, + gradual_motion: bool = False, + limits: tuple[float, float] | list | None = None, + tolerance: float = 0.001, + poll_interval: float = 0.1, + move_timeout: float = 60.0, + egu: str | None = None, + device_manager=None, + kind=None, + parent=None, + **kwargs, + ) -> None: + self.scs_axis = Axis(scs_axis) + self._tolerance = tolerance + self._poll_interval = poll_interval + self._move_timeout = move_timeout + self._egu = egu if egu is not None else AXIS_INFO[self.scs_axis].egu + self._stopped = False + self._move_thread: threading.Thread | None = None + # Optional extra gate (parent/tests). The controller READY-gate is always applied. + self.pre_move_hook: Callable[[float], None] | None = None + + # Shared, singleton controller for this smargopolo server. + self.controller = SmargopoloController( + host=host, + port=port, + simulation=simulation, + timeout=timeout, + gradual_motion=gradual_motion, + device_manager=device_manager, + ) + + super().__init__(prefix, name=name, parent=parent, kind=kind, **kwargs) + + # Alias readback to the device name so dev.wm() / BEC find it under `name`. + self.readback.name = self.name + + if limits is not None: + self.low_limit_travel.put(float(limits[0])) + self.high_limit_travel.put(float(limits[1])) + + self.controller.register_axis(self) + + # -- positioner interface ---------------------------------------------- + + @property + def limits(self) -> tuple[float, float]: + return (self.low_limit_travel.get(), self.high_limit_travel.get()) + + @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: + low, high = self.limits + if low != high and not low <= value <= high: + raise LimitError(f"{self.name}: position {value} outside limits {(low, high)}") + + # -- connection lifecycle ---------------------------------------------- + + # pylint: disable=arguments-differ + def wait_for_connection(self, all_signals: bool = False, timeout: float = 10.0) -> None: + self.controller.connect(timeout=timeout) + try: + self._update_readback(self.controller.get_position(self.scs_axis)) + except Exception as exc: # pylint: disable=broad-except + self.log.warning("%s: initial readback failed: %s", self.name, exc) + super().wait_for_connection(all_signals=all_signals, timeout=timeout) + + # -- motion ------------------------------------------------------------ + + 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 _poll_until_target(self, target: float, status: DeviceStatus) -> None: + deadline = time.monotonic() + self._move_timeout + try: + while not self._stopped: + pos = self.controller.get_position(self.scs_axis) + self._update_readback(pos) + if abs(pos - target) <= self._tolerance: + break + self.controller.raise_if_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) + self._update_readback(self.controller.get_position(self.scs_axis)) + 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 to *position*; return a status that completes at target.""" + self.check_value(position) + self.controller.require_ready() + 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.controller.move_absolute({self.scs_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.controller.stop([self.scs_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..528a26b3 --- /dev/null +++ b/csaxs_bec/devices/smargon/smargon.py @@ -0,0 +1,182 @@ +"""SmarGon coordinator -- a thin top-level device for whole-goniometer operations. + +In the per-axis architecture the SCS axes are independent top-level +:class:`~csaxs_bec.devices.smargon.positioner.SmargopoloMotor` devices +(``dev.sgx``, ``dev.sgchi``, ...). ``SmarGon`` is a small companion device +(``dev.smargon``) that exposes the operations that belong to the *whole* goniometer +rather than one axis -- referencing / the mode state machine, the coordinated +multi-axis move, beamline nudges and the raw ``q1..q6`` diagnostics. + +It owns no axes and no motion logic of its own: it holds a reference to the same +shared :class:`~csaxs_bec.devices.smargon.controller.SmargopoloController` the motors +use (singleton per smargopolo server) and delegates to it. The same operations are +also reachable from any axis as ``dev..controller.()``. +""" + +from __future__ import annotations + +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 .controller import SmargopoloController +from .transport.base import Mode + +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__) + + +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.controller.mode()) + 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.controller.get_info().rosout + 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): + """Thin coordinator for the whole SmarGon goniometer (``dev.smargon``). + + Delegates every operation to the shared :class:`SmargopoloController`; holds no + axes. Point it at the same ``host``/``port`` as the axis motors. + """ + + USER_ACCESS = [ + "reference", + "is_ready", + "wait_until_ready", + "get_mode", + "move_scs", + "nudge_beamline", + "motor_positions", + "stop_all", + "transport_info", + "controller", + ] + + 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, + gradual_motion: bool = False, + scan_info: "ScanInfo | None" = None, + device_manager=None, + **kwargs, + ) -> None: + self.controller = SmargopoloController( + host=host, + port=port, + simulation=simulation, + timeout=timeout, + gradual_motion=gradual_motion, + device_manager=device_manager, + ) + super().__init__(name=name, scan_info=scan_info, device_manager=device_manager, **kwargs) + + # -- connection lifecycle ---------------------------------------------- + + # pylint: disable=arguments-differ + def wait_for_connection(self, all_signals: bool = False, timeout: float = 10.0) -> None: + self.controller.connect(timeout=timeout) + try: + info = self.controller.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}") + super().wait_for_connection(all_signals=all_signals, timeout=timeout) + + def on_stop(self) -> None: + """Halt all SCS axes.""" + self.controller.stop_all() + + # -- delegated whole-goniometer operations ----------------------------- + + def reference(self, wait: bool = True, timeout: float = 120.0) -> str: + """Reference the goniometer (READY). WARNING: physically moves it.""" + return self.controller.reference(wait=wait, timeout=timeout) + + def is_ready(self) -> bool: + return self.controller.is_ready() + + def wait_until_ready(self, timeout: float = 120.0, poll: float = 0.5) -> str: + return self.controller.wait_until_ready(timeout=timeout, poll=poll).name + + def get_mode(self) -> str: + return self.controller.get_mode() + + def move_scs(self, wait: bool = True, **targets: float) -> None: + """Coordinated multi-axis move in one ``/targetSCS`` request.""" + self.controller.move_scs(wait=wait, **targets) + + def nudge_beamline(self, modify_offset: bool = False, **deltas: float) -> None: + self.controller.nudge_beamline(modify_offset=modify_offset, **deltas) + + def motor_positions(self) -> dict: + return self.controller.motor_positions() + + def stop_all(self) -> None: + self.controller.stop_all() + + def transport_info(self) -> dict: + return self.controller.transport_info() 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..8f10b5c5 --- /dev/null +++ b/tests/tests_devices/test_smargon_device.py @@ -0,0 +1,198 @@ +"""Tests for the per-axis SmarGon devices (controller + motor + coordinator). + +All run against the in-memory FakeTransport (no hardware/server). +""" + +import pytest +from ophyd.utils import LimitError, ReadOnlyError + +from csaxs_bec.devices.smargon import SmarGon, SmargopoloController, SmargopoloMotor +from csaxs_bec.devices.smargon.transport import FakeTransport, Mode +from csaxs_bec.devices.smargon.transport.errors import SmargopoloCommandError, SmargopoloModeError + + +@pytest.fixture(autouse=True) +def _reset_controller_singletons(): + """Each test gets fresh controllers (the singleton registry is class-level).""" + SmargopoloController._controller_instances.clear() + yield + SmargopoloController._controller_instances.clear() + + +def _motor(name, scs_axis, **cfg): + m = SmargopoloMotor(name=name, scs_axis=scs_axis, simulation=True, **cfg) + m.wait_for_connection() + return m + + +def _coordinator(**cfg): + c = SmarGon(name="smargon", simulation=True, **cfg) + c.wait_for_connection() + return c + + +# -- controller singleton --------------------------------------------------- + + +def test_axes_and_coordinator_share_one_controller(): + coord = _coordinator() + sgx = _motor("sgx", "SHX") + sgy = _motor("sgy", "SHY") + assert coord.controller is sgx.controller is sgy.controller + assert coord.controller.transport is sgx.controller.transport + + +def test_real_and_sim_are_distinct_controllers(): + sim = SmargopoloMotor(name="a", scs_axis="SHX", simulation=True) + real = SmargopoloMotor(name="b", scs_axis="SHX", simulation=False, host="h", port=1) + assert sim.controller is not real.controller + assert isinstance(sim.controller.transport, FakeTransport) + + +# -- motor as a BEC-native positioner --------------------------------------- + + +def test_motor_is_positioner_with_config_limits(): + sgx = _motor("sgx", "SHX", limits=[-2, 2]) + assert sgx.limits == (-2.0, 2.0) + assert sgx.low_limit_travel.get() == -2.0 and sgx.high_limit_travel.get() == 2.0 + assert hasattr(sgx, "move") and hasattr(sgx, "stop") + + +def test_readback_is_aliased_to_device_name(): + """dev.wm relies on the readback key equalling the device name.""" + sgx = _motor("sgx", "SHX") + assert "sgx" in sgx.read() # not 'sgx_readback' + assert "sgx_setpoint" in sgx.read() + + +def test_default_egu_from_axis(): + assert _motor("sgx", "SHX").egu == "mm" + assert _motor("sgchi", "CHI").egu == "deg" + + +def test_refuses_move_until_referenced(): + sgx = _motor("sgx", "SHX") + assert sgx.controller.get_mode() == "UNINITIALIZED" + with pytest.raises(SmargopoloModeError): + sgx.move(0.5) + + +def test_reference_then_move(): + coord = _coordinator() + sgx = _motor("sgx", "SHX", limits=[-2, 2]) + coord.reference() # shared controller -> sgx is now movable + assert sgx.controller.is_ready() + st = sgx.move(0.5, wait=True) + assert st.done and st.success + assert sgx.position == pytest.approx(0.5) + + +def test_set_is_nonblocking(): + coord = _coordinator() + sgchi = _motor("sgchi", "CHI") + coord.reference() + st = sgchi.set(12.0) + st.wait(timeout=5) + assert st.success and sgchi.position == pytest.approx(12.0) + + +def test_soft_limits_enforced(): + coord = _coordinator() + sgx = _motor("sgx", "SHX", limits=[-1, 1]) + coord.reference() + with pytest.raises(LimitError): + sgx.move(2.0) + + +def test_zero_zero_limits_mean_unbounded(): + coord = _coordinator() + sgo = _motor("sgo", "OMEGA") # no limits configured -> (0, 0) + coord.reference() + assert sgo.limits == (0.0, 0.0) + sgo.move(123.0, wait=True) # not rejected + assert sgo.position == pytest.approx(123.0) + + +# -- coordinator ------------------------------------------------------------ + + +def test_coordinator_reference_and_mode(): + coord = _coordinator() + assert coord.get_mode() == "UNINITIALIZED" + coord.reference() + assert coord.is_ready() and coord.get_mode() == "READY" + + +def test_move_scs_coordinated(): + coord = _coordinator() + sgx = _motor("sgx", "SHX", limits=[-2, 2]) + sgz = _motor("sgz", "SHZ", limits=[10, 22]) + coord.reference() + coord.move_scs(SHX=0.3, SHZ=19.0, wait=True) + assert sgx.position == pytest.approx(0.3) + assert sgz.position == pytest.approx(19.0) + + +def test_move_scs_respects_limits(): + coord = _coordinator() + _motor("sgz", "SHZ", limits=[10, 22]) + coord.reference() + with pytest.raises(LimitError): + coord.move_scs(SHZ=0.0) # below the SHZ range + + +def test_motor_positions_and_transport_info(): + coord = _coordinator() + assert set(coord.motor_positions()) == {"q1", "q2", "q3", "q4", "q5", "q6"} + info = coord.transport_info() + assert info["simulation"] is True and info["connected"] is True + + +def test_mode_signal_is_readonly(): + coord = _coordinator() + assert coord.mode.get() == int(Mode.UNINITIALIZED) + with pytest.raises(ReadOnlyError): + coord.mode.put(2) + + +def test_controller_reachable_from_axis(): + coord = _coordinator() + sgx = _motor("sgx", "SHX") + # dev..controller.reference() path + sgx.controller.reference() + assert coord.is_ready() + + +# -- fault handling --------------------------------------------------------- + + +def test_move_fails_when_target_rejected_up_front(): + coord = _coordinator() + sgx = _motor("sgx", "SHX", limits=[-2, 2]) + coord.reference() + coord.controller.transport.fail_next["move_absolute"] = SmargopoloCommandError("nope") + st = sgx.set(0.5) + with pytest.raises(SmargopoloCommandError): + st.wait(timeout=5) + assert not st.success + + +def test_move_aborts_when_controller_faults_mid_move(): + coord = _coordinator(gradual_motion=True) + sgx = _motor("sgx", "SHX", gradual_motion=True) # shares the gradual controller + coord.reference() + st = sgx.set(1e6) # far target: gradual motion keeps it in-flight + coord.controller.transport._mode = Mode.ERROR + with pytest.raises(SmargopoloModeError): + st.wait(timeout=5) + assert not st.success and sgx.motor_is_moving.get() == 0 + + +def test_move_scs_aborts_on_controller_fault(): + coord = _coordinator(gradual_motion=True) + _motor("sgx", "SHX", gradual_motion=True) + coord.reference() + coord.controller.transport._mode = Mode.ERROR + with pytest.raises(SmargopoloModeError): + coord.move_scs(SHX=1e6, wait=True) 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)