From 2fbd2756b0e23136aed603eaff93178c612dfe39 Mon Sep 17 00:00:00 2001 From: x12sa Date: Sat, 7 Nov 2026 20:10:52 +0100 Subject: [PATCH] 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 Claude-Session: https://claude.ai/code/session_01M8zoDzRrboFFDabqB9bLNZ --- csaxs_bec/device_configs/smargon.yaml | 7 +- .../device_configs/smargon_sim_session.yaml | 2 +- csaxs_bec/devices/smargon/controller.py | 12 ++- csaxs_bec/devices/smargon/positioner.py | 45 ++++++++- tests/tests_devices/test_smargon_device.py | 94 +++++++++++++++++++ 5 files changed, 150 insertions(+), 10 deletions(-) diff --git a/csaxs_bec/device_configs/smargon.yaml b/csaxs_bec/device_configs/smargon.yaml index fe004d1e..a40932bd 100644 --- a/csaxs_bec/device_configs/smargon.yaml +++ b/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 diff --git a/csaxs_bec/device_configs/smargon_sim_session.yaml b/csaxs_bec/device_configs/smargon_sim_session.yaml index d04fed72..dca79d03 100644 --- a/csaxs_bec/device_configs/smargon_sim_session.yaml +++ b/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] diff --git a/csaxs_bec/devices/smargon/controller.py b/csaxs_bec/devices/smargon/controller.py index e2c00605..3359ad13 100644 --- a/csaxs_bec/devices/smargon/controller.py +++ b/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: diff --git a/csaxs_bec/devices/smargon/positioner.py b/csaxs_bec/devices/smargon/positioner.py index c48361ea..6921fd53 100644 --- a/csaxs_bec/devices/smargon/positioner.py +++ b/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() diff --git a/tests/tests_devices/test_smargon_device.py b/tests/tests_devices/test_smargon_device.py index d05f2fb5..f6da280b 100644 --- a/tests/tests_devices/test_smargon_device.py +++ b/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()