feat(smargon): readback monitor follows moves made outside BEC

The motors only updated their readback at connect and during BEC moves, so a
move in the smargopolo web interface never reached BEC (a read returned the
stored value). The shared controller now polls /readbackSCS (all axes, one
request) every monitor_interval (motor config, default 0.5 s, 0 = off) and
publishes the axes that changed; it skips while BEC moves an axis, logs a
failure once, and stops when the last axis is destroyed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01M8zoDzDfsCTY2o5bT3J7Wv

csaxs_bec/devices/smargon/README.md

@@ -79,6 +79,15 @@ share the controller. Motor `deviceConfig` keys: `scs_axis`, `host`, `port`, `li
 Simulation: `simulation: true` (+ optional `gradual_motion: true`) uses `FakeTransport`,
 no server required.
 
+## Readback monitor (moves made outside BEC)
+
+The shared controller polls the SCS readback of all axes (`/readbackSCS`, one request)
+every `monitor_interval` seconds (motor config, default 0.5 s, `0` disables) and
+publishes the axes that changed. Moves made in the smargopolo web interface therefore
+show up in BEC (`wm`, motor widgets, scan baselines) within about one interval. The
+monitor skips a cycle while BEC itself moves an axis (that move publishes its own
+readback), logs a connection failure once, and stops with the last destroyed axis.
+
 ## Limits & coupled-axis faults
 
 SCS soft limits are **user-set** (per experiment), not hardware ranges: a SmarGon's

csaxs_bec/devices/smargon/controller.py

@@ -36,6 +36,11 @@ except Exception:  # pragma: no cover
 
     logger = logging.getLogger(__name__)
 
+# Readback monitor: polling interval (s) and the change below which an axis is not
+# republished (SCS units, mm / deg)
+DEFAULT_MONITOR_INTERVAL = 0.5
+MONITOR_DEADBAND = 1e-6
+
 
 class SmargopoloController(OphydObject):
     """One shared control plane for a smargopolo server (singleton per host/port).
@@ -99,6 +104,10 @@ class SmargopoloController(OphydObject):
             self.transport = RestTransport(host=host, port=port, timeout=timeout)
         self._axes: dict[Axis, object] = {}  # scs_axis -> SmargopoloMotor (registered by motors)
         self._connected = False
+        # readback monitor: follows moves made outside BEC (smargopolo web interface)
+        self.monitor_interval = DEFAULT_MONITOR_INTERVAL
+        self._monitor_thread: threading.Thread | None = None
+        self._monitor_stop = threading.Event()
         self._initialized = True
 
     # -- axis registry -----------------------------------------------------
@@ -107,16 +116,71 @@ class SmargopoloController(OphydObject):
         """Called by each :class:`SmargopoloMotor` so the controller can find it."""
         self._axes[Axis(motor.scs_axis)] = motor
 
+    def unregister_axis(self, motor) -> None:
+        """Called when a motor is destroyed; the monitor stops with the last motor."""
+        axis = Axis(motor.scs_axis)
+        if self._axes.get(axis) is motor:
+            del self._axes[axis]
+        if not self._axes:
+            self.stop_monitor()
+
+    # -- readback monitor --------------------------------------------------
+
+    def start_monitor(self) -> None:
+        """
+        Poll the SCS readback of all axes every monitor_interval seconds and publish the
+        axes that changed, so moves made outside BEC (smargopolo web interface) show up
+        in BEC. One REST call per cycle. monitor_interval 0 disables it.
+        """
+        if self.monitor_interval <= 0:
+            return
+        if self._monitor_thread is not None and self._monitor_thread.is_alive():
+            return
+        self._monitor_stop.clear()
+        self._monitor_thread = threading.Thread(
+            target=self._monitor_loop, name=f"{self.name}_monitor", daemon=True
+        )
+        self._monitor_thread.start()
+
+    def stop_monitor(self) -> None:
+        self._monitor_stop.set()
+        thread, self._monitor_thread = self._monitor_thread, None
+        if thread is not None and thread is not threading.current_thread():
+            thread.join(timeout=2.0)
+
+    def _monitor_loop(self) -> None:
+        failing = False
+        while not self._monitor_stop.wait(self.monitor_interval):
+            motors = dict(self._axes)
+            # a BEC move publishes its own readback while it polls to the target
+            if not motors or any(m.motor_is_moving.get() for m in motors.values()):
+                continue
+            try:
+                positions = self.get_positions()
+            except Exception as exc:  # pylint: disable=broad-except
+                if not failing:
+                    logger.warning(f"{self.name}: readback monitor failed: {exc}")
+                failing = True
+                continue
+            if failing:
+                logger.info(f"{self.name}: readback monitor recovered.")
+                failing = False
+            for axis, motor in motors.items():
+                value = positions.get(axis)
+                if value is not None and abs(value - motor.readback.get()) > MONITOR_DEADBAND:
+                    motor._update_readback(value)  # pylint: disable=protected-access
+
     # -- 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
+            if not self._connected:
+                self.transport.connect(timeout=timeout)
+                self._connected = True
+        self.start_monitor()
 
     def close(self) -> None:
+        self.stop_monitor()
         with self._lock:
             try:
                 self.transport.close()

csaxs_bec/devices/smargon/positioner.py

@@ -46,6 +46,10 @@ class SmargopoloMotor(Device, PositionerBase):
         real coupled hardware limits.
     tolerance / poll_interval / move_timeout:
         Move-completion tolerance and polling cadence / timeout.
+    monitor_interval:
+        Interval (s) of the shared controller's readback monitor, which publishes
+        moves made outside BEC (smargopolo web interface). 0 disables it. All axes
+        share one monitor; the value of the last configured axis applies.
     egu:
         Engineering unit; defaults to the axis' natural unit (mm or deg).
     """
@@ -76,6 +80,7 @@ class SmargopoloMotor(Device, PositionerBase):
         tolerance: float = 0.001,
         poll_interval: float = 0.1,
         move_timeout: float = 60.0,
+        monitor_interval: float = 0.5,
         egu: str | None = None,
         device_manager=None,
         kind=None,
@@ -111,8 +116,16 @@ class SmargopoloMotor(Device, PositionerBase):
             self.low_limit_travel.put(float(limits[0]))
             self.high_limit_travel.put(float(limits[1]))
 
+        self.controller.monitor_interval = float(monitor_interval)
         self.controller.register_axis(self)
 
+    def destroy(self) -> None:
+        """Unregister from the shared controller (its monitor stops with the last axis)."""
+        try:
+            self.controller.unregister_axis(self)
+        finally:
+            super().destroy()
+
     # -- positioner interface ----------------------------------------------
 
     @property

tests/tests_devices/test_smargon_device.py

@@ -16,6 +16,8 @@ def _reset_controller_singletons():
     """Each test gets fresh controllers (the singleton registry is class-level)."""
     SmargopoloController._controller_instances.clear()
     yield
+    for controller in SmargopoloController._controller_instances.values():
+        controller.stop_monitor()
     SmargopoloController._controller_instances.clear()
 
 
@@ -196,3 +198,71 @@ def test_move_scs_aborts_on_controller_fault():
     coord.controller.transport._mode = Mode.ERROR
     with pytest.raises(SmargopoloModeError):
         coord.move_scs(SHX=1e6, wait=True)
+
+
+# -- readback monitor (moves made outside BEC) --------------------------------
+
+
+def _wait_for(condition, timeout=2.0):
+    import time
+
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        if condition():
+            return True
+        time.sleep(0.01)
+    return False
+
+
+def test_monitor_publishes_moves_made_outside_bec():
+    from csaxs_bec.devices.smargon.transport.base import Axis
+
+    sgx = _motor("sgx", "SHX", monitor_interval=0.02)
+    published = []
+    sgx.subscribe(lambda value, **_: published.append(value), event_type="readback", run=False)
+    # the smargopolo web interface moves the axis
+    sgx.controller.transport._positions[Axis.SHX] = 1.234
+    assert _wait_for(lambda: sgx.readback.get() == pytest.approx(1.234))
+    assert published and published[-1] == pytest.approx(1.234)
+    sgx.destroy()
+
+
+def test_monitor_does_not_republish_unchanged_positions():
+    sgx = _motor("sgx", "SHX", monitor_interval=0.02)
+    published = []
+    sgx.subscribe(lambda value, **_: published.append(value), event_type="readback", run=False)
+    import time
+
+    time.sleep(0.15)
+    assert published == []
+    sgx.destroy()
+
+
+def test_monitor_skips_while_bec_moves():
+    from csaxs_bec.devices.smargon.transport.base import Axis
+
+    sgx = _motor("sgx", "SHX", monitor_interval=0.02)
+    sgx.motor_is_moving.put(1)
+    sgx.controller.transport._positions[Axis.SHX] = 0.5
+    import time
+
+    time.sleep(0.15)
+    assert sgx.readback.get() != pytest.approx(0.5)
+    sgx.motor_is_moving.put(0)
+    assert _wait_for(lambda: sgx.readback.get() == pytest.approx(0.5))
+    sgx.destroy()
+
+
+def test_monitor_stops_with_the_last_axis_and_can_be_disabled():
+    sgx = _motor("sgx", "SHX", monitor_interval=0.02)
+    sgy = _motor("sgy", "SHY", monitor_interval=0.02)
+    controller = sgx.controller
+    sgx.destroy()
+    assert controller._monitor_thread is not None and controller._monitor_thread.is_alive()
+    sgy.destroy()
+    assert controller._monitor_thread is None
+
+    SmargopoloController._controller_instances.clear()
+    off = _motor("sgz", "SHZ", monitor_interval=0)
+    assert off.controller._monitor_thread is None
+    off.destroy()