feat(smargon): scaffold SmarGon goniometer device (DRAFT, untested vs hardware)

Initial BEC ophyd integration for the SmarGon goniometer via the smargopolo
RESTful API. Controls the virtual SCS axes (the server runs the kinematics; the
underlying q1..q6 stages are never commanded directly). Structure mirrors the
Canon CR-N300 device: an ophyd-free transport (real RestTransport over urllib +
in-memory FakeTransport) with threaded poll-to-tolerance positioners and a
PSIDeviceBase parent.

Scope / decisions:
- v1 movable axes: SHX SHY SHZ CHI PHI; OMEGA optional via has_omega (YAML).
- Referencing is a deliberate operator action (reference()); moves refuse unless
  Mode.READY.
- Soft limits are user-set; smargopolo owns the coupled hardware limits. Both
  fault paths handled: up-front PUT rejection fails the move immediately, and a
  mid-move fault (mode -> 99 ERROR) aborts the in-flight move with the rosout
  detail (SmarGon._raise_if_error).
- Move completion = readback within (user-set) tolerance, done on first
  in-tolerance sample (robust to active position-hold dither).

Status: DRAFT. 27 unit tests pass against FakeTransport, but NOTHING has been
tested against a real smargopolo server or hardware. Do not commission as-is.

Open questions (to confirm with the smargopolo maintainer, W. Glettig):
- Stop semantics: no explicit stop endpoint; we abort by retargeting an axis to
  its current readback ("Follow Target" halt). Confirm this is the intended way.
- Mid-move fault signalling: we rely on /mode -> 99 (ERROR) with the reason in
  /readbackMCS rosout.msg. Confirm an out-of-range / coupled-limit violation
  reliably drives this, so the abort path is dependable.
- Per-poll cost: the mid-move check adds a /readbackMCS GET alongside the
  /readbackSCS position read (~20 req/s per moving axis at 0.1s). Revisit once we
  know whether /readbackSCS also carries mode, or once the per-motor `state`
  strings can give a truer moving/settled flag.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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

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`.

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",
+]

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)

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),
+        }

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",
+]

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://<host>:<port>``), 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.
+        """

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."""

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

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"]

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)

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)