feat(smargon): modulo option for endless rotation axes; use it for PHI

With `modulo`, requested positions are taken modulo this value and the
axis always moves the shortest way (350 -> 10 moves +20, not -340). The
readback is shown wrapped into [-modulo/2, modulo/2) while the raw
smargopolo position may grow freely.

sgphi uses `modulo: 360` instead of `limits: [-180, 180]` in smargon.yaml
and smargon_sim_session.yaml.

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

csaxs_bec/device_configs/smargon.yaml

@@ -28,8 +28,9 @@ smargon:
 
 # --- 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).
+# NOTE: SHZ is not centred on 0 (home ~18). PHI is continuous: `modulo: 360` makes
+# sgphi take requested angles mod 360 and always move the shortest way (350 -> 10 is
+# +20); the readback is shown wrapped into [-180,180), the raw smargopolo PHI may grow.
 sgx:
   description: SmarGon SHX (sample X)
   deviceClass: csaxs_bec.devices.smargon.positioner.SmargopoloMotor
@@ -69,7 +70,7 @@ sgchi:
 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}
+  deviceConfig: {scs_axis: PHI, host: tt-smargopolo, port: 3000, modulo: 360, tolerance: 0.01}
   deviceTags: [goniometer, smargon]
   enabled: true
   onFailure: retry

csaxs_bec/device_configs/smargon_sim_session.yaml

@@ -94,7 +94,7 @@ sgphi:
       scs_axis: PHI,
       simulation: true,
       gradual_motion: true,
-      limits: [-180, 180],
+      modulo: 360,
       tolerance: 0.01,
     }
   deviceTags: [goniometer, smargon, simulation]

csaxs_bec/devices/smargon/controller.py

@@ -167,7 +167,10 @@ class SmargopoloController(OphydObject):
                 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:
+                if value is None:
+                    continue
+                user_value = motor._user_position(value)  # pylint: disable=protected-access
+                if abs(user_value - motor.readback.get()) > MONITOR_DEADBAND:
                     motor._update_readback(value)  # pylint: disable=protected-access
 
     # -- connection lifecycle ----------------------------------------------
@@ -290,7 +293,8 @@ class SmargopoloController(OphydObject):
         """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.
+        axis is limit-checked against its registered motor (if any) before sending;
+        axes whose motor has a ``modulo`` take the shortest way.
         """
         self.require_ready()
         parsed = {Axis(k): float(v) for k, v in targets.items()}
@@ -298,6 +302,7 @@ class SmargopoloController(OphydObject):
             motor = self._axes.get(ax)
             if motor is not None:
                 motor.check_value(val)
+                parsed[ax] = motor._resolve_target(val)[1]  # pylint: disable=protected-access
         self.move_absolute(parsed)
         if wait:
             for ax, val in parsed.items():
@@ -312,7 +317,8 @@ class SmargopoloController(OphydObject):
             pos = self.get_position(axis)
             if motor is not None:
                 motor._update_readback(pos)  # noqa: SLF001
-            if abs(pos - target) <= tol:
+            dist = motor._distance(pos, target) if motor is not None else abs(pos - target)
+            if dist <= tol:
                 return
             self.raise_if_error()
             if time.monotonic() > deadline:

csaxs_bec/devices/smargon/positioner.py

@@ -52,6 +52,13 @@ class SmargopoloMotor(Device, PositionerBase):
         share one monitor; the value of the last configured axis applies.
     egu:
         Engineering unit; defaults to the axis' natural unit (mm or deg).
+    modulo:
+        Optional period (e.g. ``360`` for an endless rotation such as PHI). When set,
+        requested positions are taken modulo this value and the axis always moves the
+        shortest way (350 -> 10 moves +20, not -340). The readback is shown wrapped into
+        ``[-modulo/2, modulo/2)`` while the raw smargopolo position may grow freely.
+        Note: this applies to every absolute request, so a relative move of more than
+        half a period (``umvr(sgphi, 270)``) also takes the short way (-90).
     """
 
     readback = Cpt(Signal, value=0.0, kind=Kind.hinted)
@@ -82,6 +89,7 @@ class SmargopoloMotor(Device, PositionerBase):
         move_timeout: float = 60.0,
         monitor_interval: float = 0.5,
         egu: str | None = None,
+        modulo: float | None = None,
         device_manager=None,
         kind=None,
         parent=None,
@@ -92,6 +100,7 @@ class SmargopoloMotor(Device, PositionerBase):
         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._modulo = float(modulo) if modulo else None
         self._stopped = False
         self._move_thread: threading.Thread | None = None
         # Optional extra gate (parent/tests). The controller READY-gate is always applied.
@@ -149,10 +158,38 @@ class SmargopoloMotor(Device, PositionerBase):
         return self.readback.get()
 
     def check_value(self, value: float) -> None:
+        value = self._wrap(value)
         low, high = self.limits
         if low != high and not low <= value <= high:
             raise LimitError(f"{self.name}: position {value} outside limits {(low, high)}")
 
+    # -- modulo (endless rotation) helpers ------------------------------------
+
+    def _wrap(self, value: float) -> float:
+        """Wrap *value* into ``[-modulo/2, modulo/2)``; identity without modulo."""
+        if self._modulo is None:
+            return value
+        half = self._modulo / 2
+        return ((value + half) % self._modulo) - half
+
+    def _user_position(self, raw: float) -> float:
+        """User-facing position for a raw smargopolo position."""
+        return self._wrap(raw)
+
+    def _distance(self, pos: float, target: float) -> float:
+        """Distance pos -> target (modulo-aware, so a server-side wrap still matches)."""
+        return abs(self._wrap(pos - target))
+
+    def _resolve_target(self, position: float) -> tuple[float, float]:
+        """Return ``(user_target, raw_target)`` for a requested *position*.
+
+        With modulo, raw_target is the raw position reached by the shortest move.
+        """
+        if self._modulo is None:
+            return position, position
+        raw_now = self.controller.get_position(self.scs_axis)
+        return self._wrap(position), raw_now + self._wrap(position - raw_now)
+
     # -- connection lifecycle ----------------------------------------------
 
     # pylint: disable=arguments-differ
@@ -167,6 +204,7 @@ class SmargopoloMotor(Device, PositionerBase):
     # -- motion ------------------------------------------------------------
 
     def _update_readback(self, value: float) -> None:
+        value = self._user_position(value)
         old = self.readback.get()
         self.readback.put(value)
         self._set_position(value)
@@ -180,7 +218,7 @@ class SmargopoloMotor(Device, PositionerBase):
             while not self._stopped:
                 pos = self.controller.get_position(self.scs_axis)
                 self._update_readback(pos)
-                if abs(pos - target) <= self._tolerance:
+                if self._distance(pos, target) <= self._tolerance:
                     break
                 self.controller.raise_if_error()
                 if time.monotonic() > deadline:
@@ -209,6 +247,7 @@ class SmargopoloMotor(Device, PositionerBase):
         if self.pre_move_hook is not None:
             self.pre_move_hook(position)
         self._stopped = False
+        position, raw_target = self._resolve_target(position)
         self.setpoint.put(position)
 
         status = DeviceStatus(self, timeout=timeout or self._move_timeout)
@@ -217,14 +256,14 @@ class SmargopoloMotor(Device, PositionerBase):
         self.motor_is_moving.put(1)
 
         try:
-            self.controller.move_absolute({self.scs_axis: position})
+            self.controller.move_absolute({self.scs_axis: raw_target})
         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
+            target=self._poll_until_target, args=(raw_target, status), daemon=True
         )
         self._move_thread.start()
 

tests/tests_devices/test_smargon_device.py

@@ -266,3 +266,97 @@ def test_monitor_stops_with_the_last_axis_and_can_be_disabled():
     off = _motor("sgz", "SHZ", monitor_interval=0)
     assert off.controller._monitor_thread is None
     off.destroy()
+
+
+# -- modulo (endless rotation, shortest path) ---------------------------------
+
+
+def _phi(**cfg):
+    coord = _coordinator()
+    sgphi = _motor("sgphi", "PHI", limits=[-180, 180], modulo=360, tolerance=0.01, **cfg)
+    coord.reference()
+    return coord, sgphi
+
+
+def test_modulo_moves_shortest_way_across_the_wrap():
+    from csaxs_bec.devices.smargon.transport.base import Axis
+
+    _, sgphi = _phi()
+    sgphi.move(170, wait=True)
+    sgphi.move(-170, wait=True)  # +20 through 180, not -340
+    assert sgphi.controller.transport._positions[Axis.PHI] == pytest.approx(190)
+    assert sgphi.position == pytest.approx(-170)
+    assert sgphi.setpoint.get() == pytest.approx(-170)
+
+
+def test_modulo_request_is_taken_mod_360():
+    from csaxs_bec.devices.smargon.transport.base import Axis
+
+    _, sgphi = _phi()
+    sgphi.move(350, wait=True)  # == -10: short way from 0 is -10
+    assert sgphi.controller.transport._positions[Axis.PHI] == pytest.approx(-10)
+    assert sgphi.position == pytest.approx(-10)
+    sgphi.move(10, wait=True)  # +20
+    assert sgphi.controller.transport._positions[Axis.PHI] == pytest.approx(10)
+    sgphi.move(370, wait=True)  # == 10: no motion
+    assert sgphi.controller.transport._positions[Axis.PHI] == pytest.approx(10)
+
+
+def test_modulo_readback_wraps_after_several_turns():
+    from csaxs_bec.devices.smargon.transport.base import Axis
+
+    _, sgphi = _phi()
+    for target in (120, -120, 0, 120, -120, 0, 120):  # always +120
+        sgphi.move(target, wait=True)
+    assert sgphi.controller.transport._positions[Axis.PHI] == pytest.approx(840)
+    assert sgphi.position == pytest.approx(120)
+
+
+def test_modulo_limit_check_uses_wrapped_value():
+    _, sgphi = _phi()
+    sgphi.check_value(350)  # == -10, inside [-180, 180]
+    plain = _motor("sgchi", "CHI", limits=[-180, 180])
+    with pytest.raises(LimitError):
+        plain.check_value(350)
+
+
+def test_modulo_done_check_works_if_server_wraps():
+    from csaxs_bec.devices.smargon.transport.base import Axis
+
+    _, sgphi = _phi()
+    transport = sgphi.controller.transport
+    raw_get = transport.get_position
+
+    def wrapped_get(axis):
+        value = raw_get(axis)
+        return ((value + 180) % 360) - 180 if Axis(axis) is Axis.PHI else value
+
+    transport.get_position = wrapped_get
+    sgphi.move(170, wait=True)
+    st = sgphi.move(-170, wait=True)  # raw target 190, server reports -170
+    assert st.success and sgphi.position == pytest.approx(-170)
+
+
+def test_modulo_move_scs_takes_shortest_way():
+    from csaxs_bec.devices.smargon.transport.base import Axis
+
+    coord, sgphi = _phi()
+    coord.move_scs(PHI=170, wait=True)
+    coord.move_scs(PHI=-170, wait=True)
+    assert sgphi.controller.transport._positions[Axis.PHI] == pytest.approx(190)
+    assert sgphi.position == pytest.approx(-170)
+
+
+def test_modulo_monitor_does_not_republish_wrapped_position():
+    import time
+
+    from csaxs_bec.devices.smargon.transport.base import Axis
+
+    _, sgphi = _phi(monitor_interval=0.02)
+    sgphi.controller.transport._positions[Axis.PHI] = 370.0  # raw beyond the window
+    assert _wait_for(lambda: sgphi.readback.get() == pytest.approx(10))
+    published = []
+    sgphi.subscribe(lambda value, **_: published.append(value), event_type="readback", run=False)
+    time.sleep(0.15)
+    assert published == []
+    sgphi.destroy()