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

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

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.<axis>.controller.<op>()` (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.

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

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

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)

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.<axis>.controller.<op>()``.
+"""
+
+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()

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,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.<axis>.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)

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)