From e71f2df284dbf47f1e79cc141b62135bbcf1c06f Mon Sep 17 00:00:00 2001 From: menzel Date: Mon, 6 Jul 2026 14:19:24 +0200 Subject: [PATCH] refactor(smargon): per-axis motors + shared controller (smaract pattern) Expose each SCS axis as its own top-level BEC motor sharing a singleton SmargopoloController per host:port, plus a thin `dev.smargon` coordinator for whole-goniometer ops. Replaces the single-device-with-axis-subcomponents design. Why: BEC's motor UX (dev.wm, umv, limits, GUI motor widgets) is keyed on top-level devices. With sub-components, dev.wm showed `readback N/A` (it looks up a key equal to the device name) and `limits [0,0]` (the client reads limits from the root device's low/high_limit_travel signals). General, not a sim artifact. - controller.py: SmargopoloController(OphydObject), singleton per (host,port,sim); owns transport + lock + mode; referencing / move_scs / nudge / q1..q6 diagnostics. Reachable as dev..controller.() (USER_ACCESS). - positioner.py: SmargopoloMotor(Device, PositionerBase) -- top-level per-axis motor. scs_axis field binds the (user-chosen) BEC name to the fixed SCS axis. Exposes low/high_limit_travel signals and aliases readback to the device name, so dev.wm shows real per-axis readback + limits (low==high => unbounded). egu auto from axis. - smargon.py: SmarGon(PSIDeviceBase) reduced to a thin coordinator delegating to the shared controller; keeps read-only mode/rosout signals. - YAML: smargon.yaml + smargon_sim_session.yaml rewritten to per-axis entries (sgx/sgy/sgz/sgchi/sgphi[/sgomega]) + coordinator. Names user-configurable; SCS binding explicit via scs_axis. - transport/ layer + fake unchanged. Tests rewritten: 29 pass vs FakeTransport (singleton reset fixture; covers sharing, wm readback aliasing, limits, fault paths). Still DRAFT / untested against hardware. Co-Authored-By: Claude Opus 4.8 (1M context) --- csaxs_bec/device_configs/smargon.yaml | 126 ++++--- .../device_configs/smargon_sim_session.yaml | 89 +++++ csaxs_bec/devices/smargon/README.md | 114 ++++--- csaxs_bec/devices/smargon/__init__.py | 15 +- csaxs_bec/devices/smargon/controller.py | 284 ++++++++++++++++ csaxs_bec/devices/smargon/positioner.py | 161 ++++----- csaxs_bec/devices/smargon/smargon.py | 315 +++--------------- tests/tests_devices/test_smargon_device.py | 280 +++++++++------- 8 files changed, 810 insertions(+), 574 deletions(-) create mode 100644 csaxs_bec/device_configs/smargon_sim_session.yaml create mode 100644 csaxs_bec/devices/smargon/controller.py diff --git a/csaxs_bec/device_configs/smargon.yaml b/csaxs_bec/device_configs/smargon.yaml index b1382e8..bcbe889 100644 --- a/csaxs_bec/device_configs/smargon.yaml +++ b/csaxs_bec/device_configs/smargon.yaml @@ -1,70 +1,86 @@ # Example BEC device-config entries for the SmarGon goniometer (via smargopolo). # -# The device controls the virtual SCS axes; smargopolo runs the kinematics and -# drives the underlying q1..q6 stages. Referencing is a deliberate operator action: -# after loading, call `smargon.reference()` once smargopolo is up. +# 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()) -# --- Real goniometer (smargopolo REST API) --------------------------------------- +# --- Coordinator (whole-goniometer ops: reference/mode/move_scs/diagnostics) ------ smargon: - description: SmarGon goniometer (SCS axes via smargopolo REST API) + description: SmarGon coordinator (referencing, mode, coordinated moves) deviceClass: csaxs_bec.devices.smargon.smargon.SmarGon deviceConfig: - host: smargopolo # smargopolo server hostname -- SET TO THE ACTUAL SERVER - # (the MX-lab reference client used "mxlab1-smargopolo"; - # use the hostname reachable from the beamline network) - port: 3000 # REST API port - timeout: 5.0 # per-request transport timeout [s] - has_omega: false # set true on builds with the OMEGA rotation drive - # Per-move completion tolerance: a float (all axes) or per-axis mapping. - tolerance: - SHX: 0.001 # mm - SHY: 0.001 - SHZ: 0.001 - CHI: 0.01 # deg - PHI: 0.01 - # Soft limits in SCS units [low, high]. These are typically USER-set (per - # experiment), not hardware ranges: the true per-axis travel of a SmarGon is - # coupled across all SCS axes and cannot be reliably pre-declared. smargopolo - # owns the real hardware limits; if a target violates them the move fails (the - # PUT is rejected, or -- if the fault only shows after partial motion -- the - # device aborts the move on the controller's ERROR mode). Omit an axis here to - # leave it unbounded in BEC and rely on the controller. - # - # Values below are taken from the MX-lab reference client (resources_2/Smargon.py) - # -- a good grounded starting point, but VERIFY against this build. Note SHZ is - # NOT centred on 0 (home is SHZ~18), and PHI is a continuous axis the reference - # client wraps to [-180, 180) before sending (see the phi-wrap note in the docs). - limits: - SHX: [-2, 2] - SHY: [-2, 2] - SHZ: [10, 22] - CHI: [0, 65] - PHI: [-180, 180] - poll_interval: 0.1 # readback polling cadence while moving [s] - move_timeout: 60.0 # per-move timeout [s] - deviceTags: - - goniometer - - smargon + host: smargopolo # SET to the actual server (ref client used "mxlab1-smargopolo") + port: 3000 + timeout: 5.0 + deviceTags: [goniometer, smargon] enabled: true onFailure: retry readOnly: false readoutPriority: baseline softwareTrigger: false -# --- Simulation / offline (no server; uses FakeTransport) ------------------------ -smargon_sim: - description: SmarGon goniometer (simulation, no smargopolo server required) - deviceClass: csaxs_bec.devices.smargon.smargon.SmarGon - deviceConfig: - simulation: true - has_omega: true # exercise OMEGA in simulation - gradual_motion: true # simulate finite-speed travel (exercise move status) - deviceTags: - - goniometer - - smargon - - simulation +# --- 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: smargopolo, port: 3000, limits: [-2, 2], tolerance: 0.001} + deviceTags: [goniometer, smargon] enabled: true onFailure: retry - readOnly: false readoutPriority: baseline - softwareTrigger: false + +sgy: + description: SmarGon SHY (sample Y) + deviceClass: csaxs_bec.devices.smargon.positioner.SmargopoloMotor + deviceConfig: {scs_axis: SHY, host: 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: 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: 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: 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: smargopolo, port: 3000, tolerance: 0.01} +# deviceTags: [goniometer, smargon] +# enabled: true +# onFailure: retry +# readoutPriority: baseline diff --git a/csaxs_bec/device_configs/smargon_sim_session.yaml b/csaxs_bec/device_configs/smargon_sim_session.yaml new file mode 100644 index 0000000..61b98f1 --- /dev/null +++ b/csaxs_bec/device_configs/smargon_sim_session.yaml @@ -0,0 +1,89 @@ +# 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 diff --git a/csaxs_bec/devices/smargon/README.md b/csaxs_bec/devices/smargon/README.md index 53e02a4..53a74a2 100644 --- a/csaxs_bec/devices/smargon/README.md +++ b/csaxs_bec/devices/smargon/README.md @@ -1,37 +1,50 @@ # SmarGon goniometer (BEC integration via smargopolo) -BEC ophyd device for the [SmarGon](https://www.smaract.com/en/smargon) goniometer. +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). -## Layout +## 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/Mode enums + AXIS_INFO - rest.py # RestTransport -- real urllib client against :3000 - fake.py # FakeTransport -- in-memory sim (offline mode + tests) + 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 - positioner.py # SmargonAxis(Device, PositionerBase) -- one SCS axis - smargon.py # SmarGon(PSIDeviceBase) -- parent: transport, axes, mode + 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 mirrors the Canon CR-N300 device's structure (HTTP transport + an -in-memory fake + threaded polling positioners), since both wrap a non-EPICS, -poll-to-confirm HTTP backend. +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. -## Axes (v1) +## Devices & naming -Movable positioners: `shx shy shz` (mm) and `chi phi` (deg). `omega` (deg) is -**optional hardware** — enable it per deployment with `has_omega: true`; otherwise -it is `Kind.omitted` and refuses to move. The O-vector (`OX/OY/OZ`) and beamline -nudges (`BX/BY/BZ`) are reachable through the transport but not exposed as v1 -positioners; `nudge_beamline()` provides the BCS nudge path. +- **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..controller.()` (the motors expose `USER_ACCESS = ["controller"]`). -Raw `q1..q6` motor positions are read-only diagnostics via `motor_positions()`. +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) @@ -39,33 +52,46 @@ smargopolo only follows targets in `Mode.READY`. **Referencing physically moves goniometer**, so it is a deliberate operator action — never automatic: ```python -smargon.get_mode() # 'UNINITIALIZED' on a fresh server -smargon.reference() # PUT /mode?mode=1, waits until READY (moves hardware!) -smargon.is_ready() # True -smargon.shx.move(0.5) # now honoured +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 rather than silently doing nothing. `mode` and `rosout` signals surface the -state machine and last ROS message (error detail) for diagnostics. +message. `dev.smargon.mode` / `.rosout` surface the state machine and last ROS message. ## Usage ```python -smargon.shx.move(0.5) # single-axis move (mm) -smargon.move_scs(SHX=0.1, SHZ=18) # coordinated move in one /targetSCS call -smargon.nudge_beamline(BX=0.01) # beamline-frame nudge (/nudgeBCS) -smargon.stop_all() # halt all enabled axes -smargon.transport_info() # connection / identity diagnostics +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` for real + simulation entries. -Key `deviceConfig` keys: `host`, `port`, `has_omega`, `tolerance` (float or per-axis -map), `limits` (per-axis `[low, high]` soft limits — **verify against the build**), -`poll_interval`, `move_timeout`. Simulation: `simulation: true` (+ optional -`gradual_motion: true`) uses `FakeTransport`, no server required. +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 @@ -75,23 +101,6 @@ pytest tests/tests_devices/test_smargon_transport.py tests/tests_devices/test_sm All run against `FakeTransport` — no hardware or server needed. -## Limits & coupled-axis faults - -SCS soft limits are **user-set** (per experiment), not hardware ranges: a SmarGon's -true per-axis travel is coupled across all SCS axes (like a hexapod) and can't be -reliably declared in advance. smargopolo owns the real limits. The device handles -both failure modes: - -- **Rejected up front** — if smargopolo refuses an out-of-range target, the PUT - returns an error and the move status fails immediately. -- **Faulted mid-move** — if a target is accepted (`202`) but the controller goes to - `Mode.ERROR` only after partial motion, the move waiters detect it on the next - poll and fail the status with the controller's `rosout` detail (rather than - running to the move timeout). See `SmarGon._raise_if_error`. - -Move completion is by **readback tolerance** (user-set; done on the first in-tolerance -sample, so active position-hold dithering doesn't block completion). - ## Open items / to confirm with the smargopolo maintainer (Wayne Glettig) - **Stop semantics**: no explicit stop endpoint; we retarget axes to their readback @@ -99,5 +108,6 @@ sample, so active position-hold dithering doesn't block completion). - **Mid-move fault signal**: we infer faults from `Mode.ERROR`. Confirm an out-of-range / coupled-limit violation reliably drives `/mode` to 99 (and that `rosout` carries the reason), so the abort path is dependable. -- *(resolved)* OMEGA is drivable via `/targetSCS` wherever it physically exists — - handled by `has_omega`. +- **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. diff --git a/csaxs_bec/devices/smargon/__init__.py b/csaxs_bec/devices/smargon/__init__.py index e869ba6..da631e2 100644 --- a/csaxs_bec/devices/smargon/__init__.py +++ b/csaxs_bec/devices/smargon/__init__.py @@ -1,13 +1,17 @@ """SmarGon goniometer integration for BEC (via the smargopolo REST API). -Public entry point is :class:`SmarGon`. The control transport -(:mod:`.transport`) is ophyd-free and independently importable, so the REST client -and its in-memory fake can be used and tested without the BEC stack. +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 .positioner import SmargonAxis +from .controller import SmargopoloController +from .positioner import SmargopoloMotor from .smargon import SmarGon from .transport import ( Axis, @@ -21,7 +25,8 @@ from .transport import ( __all__ = [ "SmarGon", - "SmargonAxis", + "SmargopoloController", + "SmargopoloMotor", "Axis", "BeamlineAxis", "Mode", diff --git a/csaxs_bec/devices/smargon/controller.py b/csaxs_bec/devices/smargon/controller.py new file mode 100644 index 0000000..d5a7ca8 --- /dev/null +++ b/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..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), + } diff --git a/csaxs_bec/devices/smargon/positioner.py b/csaxs_bec/devices/smargon/positioner.py index 18a0327..032e107 100644 --- a/csaxs_bec/devices/smargon/positioner.py +++ b/csaxs_bec/devices/smargon/positioner.py @@ -1,16 +1,20 @@ -"""Transport-backed positioner for a single SmarGon SCS axis. +"""Per-axis SmarGon motor -- a top-level BEC device for one SCS axis. -``SmargonAxis`` is a real ophyd positioner (``Device`` + ``PositionerBase``) so BEC -treats each SCS axis (SHX/SHY/SHZ/CHI/PHI/OMEGA) as a motor: ``move()``/``set()`` -return a ``DeviceStatus`` that completes when the axis reaches target (or on -timeout/stop), plus ``stop()``. It follows the threaded-move pattern of the Canon -``CanonAxis`` but reads/writes through the parent device's injected -:class:`~csaxs_bec.devices.smargon.transport.base.SmargopoloTransport`. +``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. -Because smargopolo runs the kinematics, each axis simply PUTs its own SCS target -and polls its own SCS readback; the coupling to the q1..q6 motors is the server's -job, not ours. The axis never opens its own connection: it borrows the parent's -transport and a shared lock so concurrent axis commands serialise onto one session. +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 @@ -23,88 +27,97 @@ from ophyd import Component as Cpt from ophyd import Device, DeviceStatus, Kind, PositionerBase, Signal from ophyd.utils import LimitError -from .transport.base import Axis, SmargopoloTransport +from .controller import SmargopoloController +from .transport.base import AXIS_INFO, Axis -class SmargonAxis(Device, PositionerBase): - """One SCS axis of the SmarGon goniometer. +class SmargopoloMotor(Device, PositionerBase): + """One SCS axis of a SmarGon goniometer, as a standalone BEC motor. Parameters ---------- - axis: - Which :class:`Axis` this positioner drives. - egu: - Engineering units for the readback (``"mm"`` or ``"deg"``). + 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 soft ``(low, high)`` limits. ``None`` (default) means unbounded - -- the SCS travel ranges are build-specific and should be set in config. - tolerance: - Absolute readback tolerance for declaring a move complete. - poll_interval: - Seconds between readback polls while moving. - move_timeout: - Seconds before a move fails with ``TimeoutError``. + 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, - axis: Axis, - egu: str = "", - limits: tuple[float, float] | None = None, + 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, - parent=None, + egu: str | None = None, + device_manager=None, kind=None, + parent=None, **kwargs, ) -> None: - self._axis = Axis(axis) - self._egu = egu + self.scs_axis = Axis(scs_axis) self._tolerance = tolerance self._poll_interval = poll_interval self._move_timeout = move_timeout - self._limits = tuple(limits) if limits is not None else None + self._egu = egu if egu is not None else AXIS_INFO[self.scs_axis].egu self._stopped = False self._move_thread: threading.Thread | None = None - self._fallback_lock = threading.RLock() - # Optional gate set by the parent (e.g. refuse to move unless READY). + # 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) - # -- transport access -------------------------------------------------- + # Alias readback to the device name so dev.wm() / BEC find it under `name`. + self.readback.name = self.name - @property - def _transport(self) -> SmargopoloTransport: - transport = getattr(self.parent, "transport", None) - if transport is None: - raise RuntimeError(f"{self.name}: parent has no transport") - return transport + if limits is not None: + self.low_limit_travel.put(float(limits[0])) + self.high_limit_travel.put(float(limits[1])) - @property - def _lock(self) -> threading.RLock: - return getattr(self.parent, "_transport_lock", None) or self._fallback_lock - - def _call(self, fn): - """Run a transport call under the shared lock.""" - with self._lock: - return fn() + self.controller.register_axis(self) # -- positioner interface ---------------------------------------------- @property def limits(self) -> tuple[float, float]: - return self._limits if self._limits is not None else (0.0, 0.0) + return (self.low_limit_travel.get(), self.high_limit_travel.get()) @property def low_limit(self) -> float: @@ -123,11 +136,22 @@ class SmargonAxis(Device, PositionerBase): return self.readback.get() def check_value(self, value: float) -> None: - if self._limits is None: - return - low, high = self._limits + low, high = self.limits if low != high and not low <= value <= high: - raise LimitError(f"{self.name}: position {value} outside limits {self._limits}") + 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() @@ -137,35 +161,21 @@ class SmargonAxis(Device, PositionerBase): sub_type=self.SUB_READBACK, old_value=old, value=value, timestamp=time.time() ) - def _check_controller_error(self) -> None: - """Raise if the parent reports smargopolo went to ERROR mid-move. - - Coupled hardware limits (the underlying stages' true ranges depend on all - SCS axes and are not reliably known to BEC) mean a target accepted with a - ``202`` can still fault *after* partial motion -- like a hexapod hitting a - joint limit. The parent installs this so the move fails loudly with the - controller's ``rosout`` detail rather than silently timing out. - """ - check = getattr(self.parent, "_raise_if_error", None) - if check is not None: - self._call(check) - def _poll_until_target(self, target: float, status: DeviceStatus) -> None: deadline = time.monotonic() + self._move_timeout try: while not self._stopped: - pos = self._call(lambda: self._transport.get_position(self._axis)) + pos = self.controller.get_position(self.scs_axis) self._update_readback(pos) if abs(pos - target) <= self._tolerance: break - self._check_controller_error() + 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) - pos = self._call(lambda: self._transport.get_position(self._axis)) - self._update_readback(pos) + 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: @@ -180,8 +190,9 @@ class SmargonAxis(Device, PositionerBase): status.set_finished() def move(self, position: float, wait: bool = True, timeout=None, moved_cb=None) -> DeviceStatus: - """Move the axis to *position*; return a status that completes at target.""" + """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 @@ -193,7 +204,7 @@ class SmargonAxis(Device, PositionerBase): self.motor_is_moving.put(1) try: - self._call(lambda: self._transport.move_absolute({self._axis: position})) + 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) @@ -216,7 +227,7 @@ class SmargonAxis(Device, PositionerBase): """Halt this axis immediately (best-effort: never raises during teardown).""" self._stopped = True try: - self._call(lambda: self._transport.stop([self._axis])) + 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: diff --git a/csaxs_bec/devices/smargon/smargon.py b/csaxs_bec/devices/smargon/smargon.py index b5d2ee7..528a26b 100644 --- a/csaxs_bec/devices/smargon/smargon.py +++ b/csaxs_bec/devices/smargon/smargon.py @@ -1,30 +1,20 @@ -"""SmarGon goniometer -- BEC ophyd device (via the smargopolo REST API). +"""SmarGon coordinator -- a thin top-level device for whole-goniometer operations. -This device controls the *virtual* SmarGon Coordinate System (SCS) axes. The -smargopolo server runs the kinematics and drives the underlying q1..q6 stages; we -never command those directly (that is the separate low-level ``smaract`` device). +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. -Architecture ------------- -* ``self.transport`` -- a :class:`SmargopoloTransport` (real :class:`RestTransport` - or, when ``simulation: true``, :class:`FakeTransport`). All control goes through - it, serialised behind ``self._transport_lock``. -* ``shx/shy/shz/chi/phi`` (+ optional ``omega``) -- :class:`SmargonAxis` - positioners sharing the transport; they look like motors to BEC. -* ``mode`` / ``rosout`` -- read-only signals proxying the state machine + last ROS - message for diagnostics. - -State machine / referencing ---------------------------- -smargopolo only follows targets in ``Mode.READY``. Referencing physically moves -the goniometer, so it is a **deliberate operator action** here -- call -:meth:`reference`. Axes refuse to move unless READY (a pre-move guard surfaces a -clear "reference first" error instead of silently doing nothing). +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..controller.()``. """ from __future__ import annotations -import threading import time from typing import TYPE_CHECKING @@ -33,11 +23,8 @@ from ophyd import Kind, Signal from ophyd.utils import ReadOnlyError from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase -from .positioner import SmargonAxis -from .transport.base import Axis, BeamlineAxis, Mode, SmargopoloTransport -from .transport.errors import SmargopoloModeError -from .transport.fake import FakeTransport -from .transport.rest import RestTransport +from .controller import SmargopoloController +from .transport.base import Mode if TYPE_CHECKING: # pragma: no cover from bec_lib.devicemanager import ScanInfo @@ -52,11 +39,6 @@ except Exception: # pragma: no cover logger = logging.getLogger(__name__) -# Axes exposed as movable positioners in v1. OMEGA is appended at runtime when the -# deployment has the rotation drive (``has_omega: true``). -_V1_AXES: tuple[Axis, ...] = (Axis.SHX, Axis.SHY, Axis.SHZ, Axis.CHI, Axis.PHI) - - class SmargopoloModeSignal(Signal): """Read-only signal exposing the smargopolo state-machine mode (as an int).""" @@ -66,7 +48,7 @@ class SmargopoloModeSignal(Signal): # pylint: disable=arguments-differ def get(self, **kwargs): - value = int(self.parent._get_mode()) # noqa: SLF001 + value = int(self.parent.controller.mode()) self._readback = value return value @@ -87,7 +69,7 @@ class SmargopoloRosoutSignal(Signal): # pylint: disable=arguments-differ def get(self, **kwargs): - value = self.parent._get_info().rosout # noqa: SLF001 + value = self.parent.controller.get_info().rosout self._readback = value return value @@ -100,9 +82,10 @@ class SmargopoloRosoutSignal(Signal): class SmarGon(PSIDeviceBase): - """BEC ophyd device for the SmarGon goniometer driven through smargopolo. + """Thin coordinator for the whole SmarGon goniometer (``dev.smargon``). - See the module docstring for architecture and the referencing contract. + Delegates every operation to the shared :class:`SmargopoloController`; holds no + axes. Point it at the same ``host``/``port`` as the axis motors. """ USER_ACCESS = [ @@ -110,23 +93,14 @@ class SmarGon(PSIDeviceBase): "is_ready", "wait_until_ready", "get_mode", - "stop_all", - "motor_positions", "move_scs", "nudge_beamline", + "motor_positions", + "stop_all", "transport_info", + "controller", ] - # -- motion axes (look like motors to BEC) ----------------------------- - shx = Cpt(SmargonAxis, axis=Axis.SHX, egu="mm", kind=Kind.normal) - shy = Cpt(SmargonAxis, axis=Axis.SHY, egu="mm", kind=Kind.normal) - shz = Cpt(SmargonAxis, axis=Axis.SHZ, egu="mm", kind=Kind.normal) - chi = Cpt(SmargonAxis, axis=Axis.CHI, egu="deg", kind=Kind.normal) - phi = Cpt(SmargonAxis, axis=Axis.PHI, egu="deg", kind=Kind.normal) - # Optional rotation drive -- omitted unless ``has_omega`` is set (see __init__). - omega = Cpt(SmargonAxis, axis=Axis.OMEGA, egu="deg", kind=Kind.normal) - - # -- diagnostics ------------------------------------------------------- mode = Cpt(SmargopoloModeSignal, value=int(Mode.UNINITIALIZED), kind=Kind.normal) rosout = Cpt(SmargopoloRosoutSignal, value="", kind=Kind.config) @@ -138,131 +112,28 @@ class SmarGon(PSIDeviceBase): port: int = 3000, simulation: bool = False, timeout: float = 5.0, - has_omega: bool = False, - limits: dict | None = None, - tolerance: float | dict | None = None, - poll_interval: float = 0.1, - move_timeout: float = 60.0, gradual_motion: bool = False, scan_info: "ScanInfo | None" = None, device_manager=None, **kwargs, ) -> None: - """See class/module docstring for the architecture. - - Args: - host/port: smargopolo server address (ignored when ``simulation``). - simulation: use :class:`FakeTransport` -- no server needed. - timeout: per-request transport timeout in seconds. - has_omega: enable the OMEGA rotation axis (build-dependent). - limits: optional ``{axis_name: [low, high]}`` soft limits (SCS units). - tolerance: per-move completion tolerance; a float (all axes) or a - ``{axis_name: value}`` mapping. Defaults to 1e-3. - poll_interval/move_timeout: readback polling cadence / move timeout. - gradual_motion: simulate finite-speed travel in the fake transport. - """ - self._simulation = simulation - self._has_omega = has_omega - self._transport_lock = threading.RLock() - - # Build the transport BEFORE super().__init__ so axis components can reach it. - if simulation: - self.transport: SmargopoloTransport = FakeTransport(gradual_motion=gradual_motion) - else: - self.transport = RestTransport(host=host, port=port, timeout=timeout) - + 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) - # Configure the enabled axes; disable OMEGA unless the build has it. - enabled = set(_V1_AXES) | ({Axis.OMEGA} if has_omega else set()) - limits = limits or {} - for ax in self._all_axes(): - if ax._axis not in enabled: # noqa: SLF001 - ax.kind = Kind.omitted - ax.pre_move_hook = self._make_disabled_hook(ax._axis) # noqa: SLF001 - continue - ax.pre_move_hook = self._ready_guard - ax._poll_interval = poll_interval # noqa: SLF001 - ax._move_timeout = move_timeout # noqa: SLF001 - lim = limits.get(ax._axis.value) # noqa: SLF001 - if lim is not None: - ax._limits = (float(lim[0]), float(lim[1])) # noqa: SLF001 - tol = self._axis_tolerance(tolerance, ax._axis) # noqa: SLF001 - if tol is not None: - ax._tolerance = float(tol) # noqa: SLF001 - - # -- axis helpers ------------------------------------------------------ - - def _all_axes(self) -> tuple[SmargonAxis, ...]: - return (self.shx, self.shy, self.shz, self.chi, self.phi, self.omega) - - def _enabled_axes(self) -> tuple[SmargonAxis, ...]: - return tuple(ax for ax in self._all_axes() if ax.kind != Kind.omitted) - - @staticmethod - def _axis_tolerance(tolerance, axis: Axis): - if tolerance is None: - return None - if isinstance(tolerance, dict): - return tolerance.get(axis.value) - return tolerance - - @staticmethod - def _make_disabled_hook(axis: Axis): - def _hook(_position: float) -> None: - raise SmargopoloModeError( - f"axis {axis.value} is not enabled on this SmarGon " - "(set has_omega: true in the device config to enable OMEGA)" - ) - - return _hook - - def _ready_guard(self, _position: float) -> None: - mode = self._get_mode() - if mode is not Mode.READY: - raise SmargopoloModeError( - f"{self.name} is in {mode.name} (mode {int(mode)}); call reference() and wait " - "for READY before moving" - ) - - # -- transport routing ------------------------------------------------- - - def _get_mode(self) -> Mode: - with self._transport_lock: - return self.transport.get_mode() - - def _get_info(self): - with self._transport_lock: - return self.transport.get_info() - - def _raise_if_error(self) -> None: - """Raise if smargopolo is in ERROR (used to abort in-flight moves). - - smargopolo accepts a target with ``202`` even when the coupled hardware - limits will be violated, then faults once motion hits the real boundary. - Move waiters poll this so a mid-move fault surfaces the controller's - ``rosout`` message instead of running until the move timeout. - """ - info = self._get_info() - if info.mode is Mode.ERROR: - raise SmargopoloModeError( - f"{self.name}: smargopolo entered ERROR during motion " - f"(likely a hardware/coupled-axis limit): {info.rosout or 'no detail'}" - ) - # -- connection lifecycle ---------------------------------------------- # pylint: disable=arguments-differ def wait_for_connection(self, all_signals: bool = False, timeout: float = 10.0) -> None: - """Open the transport session, then wait for signals.""" - with self._transport_lock: - self.transport.connect(timeout=timeout) - super().wait_for_connection(all_signals=all_signals, timeout=timeout) - - def on_connected(self) -> None: - """Log identity and seed enabled-axis readbacks.""" + self.controller.connect(timeout=timeout) try: - info = self._get_info() + info = self.controller.get_info() logger.info( f"{self.name} connected to smargopolo (S/N {info.serial_number!r}, " f"mode {info.mode.name})" @@ -273,125 +144,39 @@ class SmarGon(PSIDeviceBase): ) except Exception as exc: # pylint: disable=broad-except logger.warning(f"{self.name}: could not read smargopolo info: {exc}") - for ax in self._enabled_axes(): - try: - with self._transport_lock: - ax._update_readback(self.transport.get_position(ax._axis)) # noqa: SLF001 - except Exception as exc: # pylint: disable=broad-except - logger.warning(f"{self.name}: initial readback for {ax.name} failed: {exc}") + super().wait_for_connection(all_signals=all_signals, timeout=timeout) def on_stop(self) -> None: - """Halt all enabled axes.""" - self.stop_all() + """Halt all SCS axes.""" + self.controller.stop_all() - def on_destroy(self) -> None: - """Close the transport session.""" - try: - self.transport.close() - except Exception as exc: # pylint: disable=broad-except - logger.warning(f"{self.name}: transport close failed: {exc}") + # -- delegated whole-goniometer operations ----------------------------- - # -- control convenience (USER_ACCESS) --------------------------------- - - def reference(self, wait: bool = True, timeout: float = 120.0) -> Mode: - """Reference the goniometer (``Mode.REFERENCING`` -> ``READY``). - - WARNING: this physically moves the goniometer to find its reference marks. - It is a deliberate operator action. Returns the resulting mode. - """ - with self._transport_lock: - self.transport.set_mode(Mode.REFERENCING) - logger.info(f"{self.name}: referencing started") - if wait: - mode = self.wait_until_ready(timeout=timeout) - logger.info(f"{self.name}: referencing finished -> {mode.name}") - return mode - return self._get_mode() + def 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: - """Whether smargopolo is in ``Mode.READY`` (follow-target).""" - return self._get_mode() is Mode.READY + return self.controller.is_ready() - def wait_until_ready(self, timeout: float = 120.0, poll: float = 0.5) -> Mode: - """Block until the mode is READY (or ERROR), or *timeout* elapses.""" - deadline = time.monotonic() + timeout - while True: - mode = self._get_mode() - if mode is Mode.READY: - return mode - if mode is Mode.ERROR: - info = self._get_info() - raise SmargopoloModeError(f"{self.name}: smargopolo in ERROR: {info.rosout}") - if time.monotonic() > deadline: - raise TimeoutError(f"{self.name}: not READY after {timeout}s (mode {mode.name})") - time.sleep(poll) + def 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 the current state-machine mode name (for the user console).""" - return self._get_mode().name - - def stop_all(self) -> None: - """Halt every enabled axis immediately.""" - for ax in self._enabled_axes(): - ax.stop() + return self.controller.get_mode() def move_scs(self, wait: bool = True, **targets: float) -> None: - """Move several SCS axes at once in a single ``/targetSCS`` call. - - More efficient than moving axes one-by-one for coordinated moves, e.g. - ``smargon.move_scs(SHX=0.1, SHZ=18)``. Keys are SCS axis names. - """ - self._ready_guard(0.0) - parsed = {Axis(k): float(v) for k, v in targets.items()} - for ax in parsed: - self._axis_for(ax).check_value(parsed[ax]) - with self._transport_lock: - self.transport.move_absolute(parsed) - if wait: - for ax_enum, val in parsed.items(): - self._wait_axis(self._axis_for(ax_enum), val) - - def _wait_axis(self, axis: SmargonAxis, target: float, timeout: float | None = None) -> None: - deadline = time.monotonic() + (timeout or axis._move_timeout) # noqa: SLF001 - while True: - with self._transport_lock: - pos = self.transport.get_position(axis._axis) # noqa: SLF001 - axis._update_readback(pos) # noqa: SLF001 - if abs(pos - target) <= axis._tolerance: # noqa: SLF001 - return - self._raise_if_error() - if time.monotonic() > deadline: - raise TimeoutError(f"{axis.name}: move to {target} timed out") - time.sleep(axis._poll_interval) # noqa: SLF001 - - def _axis_for(self, axis: Axis) -> SmargonAxis: - for ax in self._all_axes(): - if ax._axis is axis: # noqa: SLF001 - return ax - raise KeyError(axis) + """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: - """Nudge the sample along beamline axes (``BX``/``BY``/``BZ``). - - With ``modify_offset=True`` the O-vector is nudged (``/nudgeOBCS``). - """ - self._ready_guard(0.0) - parsed = {BeamlineAxis(k): float(v) for k, v in deltas.items()} - with self._transport_lock: - self.transport.nudge_beamline(parsed, modify_offset=modify_offset) + self.controller.nudge_beamline(modify_offset=modify_offset, **deltas) def motor_positions(self) -> dict: - """Return the raw ``q1..q6`` motor positions (read-only diagnostics).""" - return self._get_info().motor_positions + return self.controller.motor_positions() + + def stop_all(self) -> None: + self.controller.stop_all() def transport_info(self) -> dict: - """Return connection/identity info for diagnostics.""" - info = self._get_info() if self.transport.connected else None - return { - "connected": self.transport.connected, - "simulation": self._simulation, - "has_omega": self._has_omega, - "mode": info.mode.name if info else None, - "serial_number": getattr(info, "serial_number", None), - "seq": getattr(info, "seq", None), - } + return self.controller.transport_info() diff --git a/tests/tests_devices/test_smargon_device.py b/tests/tests_devices/test_smargon_device.py index 64a1eda..8f10b5c 100644 --- a/tests/tests_devices/test_smargon_device.py +++ b/tests/tests_devices/test_smargon_device.py @@ -1,162 +1,198 @@ -"""Tests for the SmarGon ophyd device (simulation mode, no hardware).""" +"""Tests for the per-axis SmarGon devices (controller + motor + coordinator). + +All run against the in-memory FakeTransport (no hardware/server). +""" import pytest -from ophyd import Kind -from ophyd.utils import LimitError +from ophyd.utils import LimitError, ReadOnlyError -from csaxs_bec.devices.smargon import SmarGon +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 -def smargon(): - device = SmarGon(name="smargon_test", simulation=True, has_omega=True) - device.wait_for_connection() - yield device - device.destroy() +@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 test_uses_fake_transport(smargon): - assert isinstance(smargon.transport, FakeTransport) - assert smargon.transport.connected +def _motor(name, scs_axis, **cfg): + m = SmargopoloMotor(name=name, scs_axis=scs_axis, simulation=True, **cfg) + m.wait_for_connection() + return m -def test_v1_axes_are_positioners(smargon): - for ax in (smargon.shx, smargon.shy, smargon.shz, smargon.chi, smargon.phi): - assert hasattr(ax, "move") and hasattr(ax, "stop") - assert ax.kind != Kind.omitted +def _coordinator(**cfg): + c = SmarGon(name="smargon", simulation=True, **cfg) + c.wait_for_connection() + return c -def test_refuses_move_until_referenced(smargon): - assert smargon.get_mode() == "UNINITIALIZED" +# -- 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): - smargon.shx.move(0.5) + sgx.move(0.5) -def test_reference_then_move(smargon): - smargon.reference() - assert smargon.is_ready() - status = smargon.shx.move(0.5, wait=True) - assert status.done and status.success - assert smargon.shx.position == pytest.approx(0.5) +def test_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_alias_is_nonblocking(smargon): - smargon.reference() - status = smargon.chi.set(12.0) - status.wait(timeout=5) - assert status.success - assert smargon.chi.position == pytest.approx(12.0) - - -def test_omega_enabled_when_has_omega(smargon): - assert smargon.omega.kind != Kind.omitted - smargon.reference() - smargon.omega.move(45.0, wait=True) - assert smargon.omega.position == pytest.approx(45.0) - - -def test_omega_disabled_without_has_omega(): - device = SmarGon(name="no_omega", simulation=True, has_omega=False) - device.wait_for_connection() - try: - device.reference() - assert device.omega.kind == Kind.omitted - with pytest.raises(SmargopoloModeError): - device.omega.move(45.0) - finally: - device.destroy() +def test_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(): - device = SmarGon(name="lim", simulation=True, limits={"SHX": [-1.0, 1.0]}) - device.wait_for_connection() - try: - device.reference() - assert device.shx.limits == (-1.0, 1.0) - with pytest.raises(LimitError): - device.shx.move(2.0) - finally: - device.destroy() + coord = _coordinator() + sgx = _motor("sgx", "SHX", limits=[-1, 1]) + coord.reference() + with pytest.raises(LimitError): + sgx.move(2.0) -def test_per_axis_tolerance_applied(): - device = SmarGon(name="tol", simulation=True, tolerance={"SHX": 0.05}) - device.wait_for_connection() - try: - assert device.shx._tolerance == pytest.approx(0.05) - finally: - device.destroy() +def test_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) -def test_move_scs_coordinated(smargon): - smargon.reference() - smargon.move_scs(SHX=0.3, SHZ=19.0, wait=True) - assert smargon.shx.position == pytest.approx(0.3) - assert smargon.shz.position == pytest.approx(19.0) +# -- coordinator ------------------------------------------------------------ -def test_stop_all_does_not_raise(smargon): - smargon.reference() - smargon.shx.set(0.2) - smargon.stop_all() +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_motor_positions_diagnostic(smargon): - motors = smargon.motor_positions() - assert set(motors) == {"q1", "q2", "q3", "q4", "q5", "q6"} +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_transport_info(smargon): - info = smargon.transport_info() - assert info["simulation"] is True - assert info["has_omega"] is True - assert info["connected"] is True +def test_move_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_move_fails_when_target_rejected_up_front(smargon): - """smargopolo refusing an out-of-range target -> immediate move failure.""" - smargon.reference() - smargon.transport.fail_next["move_absolute"] = SmargopoloCommandError("target out of range") - status = smargon.shx.set(0.5) +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..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): - status.wait(timeout=5) - assert not status.success + st.wait(timeout=5) + assert not st.success def test_move_aborts_when_controller_faults_mid_move(): - """Target accepted (202) but a coupled-limit fault appears after partial motion.""" - device = SmarGon(name="faulter", simulation=True, gradual_motion=True) - device.wait_for_connection() - try: - device.reference() - status = device.shx.set(1e6) # far target: gradual motion keeps it in-flight - # Simulate smargopolo hitting a coupled hardware limit mid-move. - device.transport._mode = Mode.ERROR - with pytest.raises(SmargopoloModeError): - status.wait(timeout=5) - assert not status.success - assert device.shx.motor_is_moving.get() == 0 - finally: - device.destroy() + 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(): - device = SmarGon(name="faulter2", simulation=True, gradual_motion=True) - device.wait_for_connection() - try: - device.reference() - device.transport._mode = Mode.ERROR - with pytest.raises(SmargopoloModeError): - device.move_scs(SHX=1e6, wait=True) - finally: - device.destroy() - - -def test_mode_signal_is_readonly(smargon): - from ophyd.utils import ReadOnlyError - - assert smargon.mode.get() == int(Mode.UNINITIALIZED) - with pytest.raises(ReadOnlyError): - smargon.mode.put(2) + 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)