diff --git a/csaxs_bec/devices/canon/COMMISSIONING.md b/csaxs_bec/devices/canon/COMMISSIONING.md
index ee3b8a37..1bfca996 100644
--- a/csaxs_bec/devices/canon/COMMISSIONING.md
+++ b/csaxs_bec/devices/canon/COMMISSIONING.md
@@ -174,7 +174,7 @@ footprint behind one grab API. Default is **`jpeg_poll`**.
## 3. What has already been tested
-**98 automated tests, no hardware required** (`pytest tests/tests_devices/test_canon_*.py
+**104 automated tests, no hardware required** (`pytest tests/tests_devices/test_canon_*.py
--random-order`, run under `OPHYD_CONTROL_LAYER=dummy`). They prove the **Python logic**
using the in-memory `FakeTransport` and a **mocked HTTP opener** — i.e. they confirm the
device builds the requests we *intend* and parses the response format we *assume*. They
@@ -308,25 +308,24 @@ tightening tolerance or adding settle time.
**R3 — Preset CGI.** `preset.cgi?preset=`. *Symptom:* preset recall
errors or no-op. *Fix:* `_CGI["preset"]` / `_PARAM["preset"]`.
-**R9 — Stop command. UNVERIFIED, and it fails silently.**
-`stop()` sends `control.cgi?stop=all|`, but **no bare `stop` field appears anywhere
-in the camera's 32 kB info.cgi response**. The camera expresses actions as
-`.action` with an allowed list — `p.action.list:=stop`,
-`c.1.focus.action.list:=far,near,one_shot,stop` — so `stop=pan` is likely not a thing
-this firmware understands.
+**R9 — Stop command. CONFIRMED BROKEN as shipped; now fixed.**
+`stop()` used to send `control.cgi?stop=`. **Verified against hardware on
+[[2026-08-18]]: it did nothing.** No bare `stop` field exists on this camera — actions
+are `.action` — and because the server answers HTTP 200 for unimplemented
+commands, the failure was completely silent. `stop_all()`, the device's `on_stop()`
+abort hook and the motion tool's own cleanup were all quietly no-ops.
-What makes this the most dangerous open item rather than a cosmetic one: **the server
-answers HTTP 200 for unimplemented CGIs**, so a no-op stop is indistinguishable from a
-working one at the protocol level. `stop_all()`, the device's `on_stop()` abort hook and
-`motion_check.py`'s own cleanup would all quietly do nothing.
+*Fixed* by not sending a stop command at all: `stop()` now halts each axis by
+**commanding a move to where that axis currently is**, built on the one motion
+primitive proven to work here. It is protocol-independent, and samples all positions in
+one info.cgi rather than one per axis. Two properties to know: the axis decelerates on
+its ramp rather than dead-stopping, and the info.cgi read (~32 kB) precedes the halt, so
+there is a round-trip of latency on the path taken when things go wrong.
-*Test it directly* — `motion_check.py --test-stop` commands a move, interrupts it, and
-watches whether the axis froze or carried on:
-```bash
-python3 motion_check.py --host --axis pan --delta 5 --speed 2 --test-stop --go
-```
-*Fix if it fails:* `_PARAM["stop"]` and `stop()` — most likely an `.action`-style
-command. Do this **before** Stage 4.5 (jog), which starts continuous motion.
+*Optional refinement:* `motion_check.py --find-stop` hunts for a native one-request stop
+(`c.1.pan.action=stop` and similar) by interrupting a series of moves with each
+candidate. If one halts the axis, wire it in as the fast path and keep the re-target as
+the fallback — a silent no-op is exactly what the fallback exists to protect against.
**R10 — Image endpoints (stream *and* still).** Two separate SPEC
constants in `xc.py`, both defaulting to `image.cgi`; on real firmware they may differ
@@ -371,7 +370,7 @@ the bench**; Stage 1+ need it on the network.
### Stage 0 — Pre-hardware (no camera needed) — do this first
0.1 **Run the suite.** `OPHYD_CONTROL_LAYER=dummy pytest tests/tests_devices/test_canon_*.py
---random-order`. *Expect 98 passed.* If not, stop — the environment is wrong, not the
+--random-order`. *Expect 104 passed.* If not, stop — the environment is wrong, not the
camera.
0.2 **Point the editable install at *this* worktree — before starting BEC.** `csaxs_bec`
@@ -428,7 +427,7 @@ cp csaxs_bec/devices/canon/probe.py /tmp/ && (cd /tmp && python3 probe.py --host
`# SPEC:` tables in `transport/xc.py` you can already confirm from the document. *Every
correction here saves a hardware round-trip.*
-**Gate:** 98 green + the canon worktree is the live install + sim device usable in BEC +
+**Gate:** 104 green + the canon worktree is the live install + sim device usable in BEC +
probe rehearsal shows a full session + SPEC tables reviewed. → proceed.
### Stage 0.5 — First contact (hardware just arrived) — **start here**
diff --git a/csaxs_bec/devices/canon/motion_check.py b/csaxs_bec/devices/canon/motion_check.py
index 4449a1d0..96fc9dbb 100644
--- a/csaxs_bec/devices/canon/motion_check.py
+++ b/csaxs_bec/devices/canon/motion_check.py
@@ -97,6 +97,73 @@ def _import_transport():
return XCTransport, Axis, HARDWARE_LIMITS
+# Candidate native stop commands, in decreasing order of evidence. The camera
+# expresses actions as ".action" with an allowed list, and the only
+# "stop" values seen in info.cgi are p.action.list and c.1.focus.action.list --
+# neither of which is obviously the PTZ stop, hence the hunt.
+_STOP_CANDIDATES = [
+ ("{axis}.action=stop", lambda key, ax: {f"{key}.action": "stop"}),
+ ("c.1.action=stop", lambda key, ax: {"c.1.action": "stop"}),
+ ("p.action=stop", lambda key, ax: {"p.action": "stop"}),
+ ("c.1.ptz.action=stop", lambda key, ax: {"c.1.ptz.action": "stop"}),
+ ("{axis}.speed=0", lambda key, ax: {f"{key}.speed": 0}),
+]
+
+
+def _find_native_stop(transport, axis, args, unit) -> dict:
+ """Hunt for a stop command that halts the axis in one request.
+
+ ``stop()`` works by re-targeting the current position, which is reliable but must
+ read info.cgi (~32 kB) before it can act -- at 100 deg/s that round-trip is real
+ extra travel on the path taken when something is wrong. A native one-request stop
+ would remove it. Each candidate gets its own move to interrupt.
+ """
+ from_ = _import_transport()[1]
+ key = {"pan": "c.1.pan", "tilt": "c.1.tilt", "zoom": "c.1.zoom"}[axis.value]
+ results = {}
+ print(f"\nHunting a native stop for {axis.value} -- {len(_STOP_CANDIDATES)} candidates,")
+ print("each gets its own move to interrupt.\n")
+
+ for label, build in _STOP_CANDIDATES:
+ name = label.format(axis=key)
+ start = transport.get_position(axis)
+ target = start + args.delta
+ limits = _import_transport()[2][axis]
+ if not limits.low <= target <= limits.high:
+ args.delta = -args.delta # bounce off the limit and keep going
+ target = start + args.delta
+ speed = args.speed if axis.value in ("pan", "tilt") else None
+ transport.move_absolute(axis, target, speed=speed)
+ time.sleep(args.stop_after)
+ at_stop = transport.get_position(axis)
+ try:
+ transport._request("control", build(key, axis)) # noqa: SLF001
+ sent = True
+ except Exception as exc: # pylint: disable=broad-except
+ print(f" {name:<24} request failed: {exc}")
+ sent = False
+ time.sleep(args.settle)
+ after = transport.get_position(axis)
+ drift = after - at_stop
+ halted = sent and abs(drift) <= args.tolerance and abs(at_stop - target) > args.tolerance
+ results[name] = {"halted": halted, "drift": drift}
+ print(f" {name:<24} drift {drift:+7.3f} {unit} -> {'HALTS' if halted else 'no effect'}")
+ # Leave the axis where it is before the next candidate.
+ transport.stop(axis)
+ time.sleep(0.3)
+
+ winners = [n for n, r in results.items() if r["halted"]]
+ print("\n" + "-" * 72)
+ if winners:
+ print(f"RESULT: native stop found -> {winners[0]}")
+ print(" Wire it into XCTransport.stop() as the fast path, keeping the")
+ print(" position re-target as the fallback.")
+ else:
+ print("RESULT: none of the candidates halted the axis.")
+ print(" Keep the position re-target -- it is proven, just not instant.")
+ return {"stop_test": True, "sent": True, "candidates": results, "native_stop": winners}
+
+
def _run_stop_test(transport, axis, args, unit, start, target) -> dict:
"""Command a move, interrupt it with stop(), and see whether it really halted."""
report: dict = {"stop_test": True, "start": start, "target": target, "sent": False}
@@ -189,7 +256,11 @@ def run(args, transport=None) -> dict: # pylint: disable=too-many-branches,too-
return report
if not args.go:
- if args.test_stop:
+ if args.find_stop:
+ print("\nDRY RUN (stop hunt) -- nothing sent. Re-run with --go.")
+ print(f"Would make {len(_STOP_CANDIDATES)} moves of {args.delta:+g} {unit},")
+ print("interrupting each with a different candidate command.")
+ elif args.test_stop:
print("\nDRY RUN (stop test) -- nothing sent. Re-run with --go.")
print(f"Would move toward {target:+.3f} {unit}, send stop() after")
print(f"{args.stop_after} s, then watch for {args.settle} s.")
@@ -200,6 +271,12 @@ def run(args, transport=None) -> dict: # pylint: disable=too-many-branches,too-
report["dry_run"] = True
return report
+ if args.find_stop:
+ transport.claim_control()
+ print("\ncontrol claimed")
+ report.update(_find_native_stop(transport, axis, args, unit))
+ return report
+
if args.test_stop:
report.update(_run_stop_test(transport, axis, args, unit, start, target))
return report
@@ -285,6 +362,11 @@ def main(argv: list[str] | None = None) -> int:
action="store_true",
help="interrupt the move with stop() and check the axis really halts (R9)",
)
+ parser.add_argument(
+ "--find-stop",
+ action="store_true",
+ help="hunt for a native one-request stop command (moves the axis repeatedly)",
+ )
parser.add_argument("--stop-after", type=float, default=1.0, help="seconds before stop()")
parser.add_argument("--settle", type=float, default=2.0, help="seconds to watch after stop()")
parser.add_argument("--speed", type=float, default=5.0, help="deg/s for pan/tilt")
diff --git a/csaxs_bec/devices/canon/transport/xc.py b/csaxs_bec/devices/canon/transport/xc.py
index 27b61740..c1fe677a 100644
--- a/csaxs_bec/devices/canon/transport/xc.py
+++ b/csaxs_bec/devices/canon/transport/xc.py
@@ -115,7 +115,6 @@ _PARAM = {
"focus_mode": f"{_CAM}.focus",
"face_detect": f"{_CAM}.focus.detect",
"auto_track": f"{_CAM}.focus.auto.track",
- "stop": "stop",
}
# Camera parameters exposed as ophyd signals, mapped to their measured names.
@@ -501,10 +500,48 @@ class XCTransport(CameraTransport):
return flag not in (None, "0", "false", "")
def stop(self, axis: Axis | None = None) -> None:
+ """Halt *axis*, or all axes when *axis* is ``None``. Requires control.
+
+ MEASURED 2026-08-18: the shipped ``control.cgi?stop=`` **did nothing**.
+ No bare ``stop`` field exists on this camera -- it expresses actions as
+ ``.action`` -- and because the server answers HTTP 200 for
+ unimplemented commands, the failure was completely silent.
+
+ So this does not send a stop command at all. It halts each axis by
+ **commanding a move to where that axis currently is**, which is built on the
+ one motion primitive proven to work on this hardware. Properties worth knowing:
+
+ * It is protocol-independent: any camera that accepts absolute moves can be
+ stopped this way, whatever it calls its stop command.
+ * The axis decelerates on its normal ramp rather than dead-stopping, and may
+ drift a fraction past the sampled position before settling -- it is a halt,
+ not a freeze.
+ * Position is sampled **once** for all axes (one info.cgi), not once per axis,
+ because this is the path taken when something is going wrong and it must not
+ be slow.
+
+ If the camera's native stop is ever identified (``motion_check.py --find-stop``
+ hunts for it), sending that first would be a refinement -- but this must remain
+ the fallback, since a silent no-op is exactly what it is protecting against.
+ """
if not self._has_control:
raise CanonControlPrivilegeError("control privilege required to stop")
- target = "all" if axis is None else axis.value
- self._request("control", {_PARAM["stop"]: target})
+ axes = list(Axis) if axis is None else [axis]
+ kv = self._parse_kv(self._request("info").text)
+ failures = []
+ for ax in axes:
+ key = _PARAM[ax.value]
+ raw = kv.get(key)
+ if raw is None:
+ failures.append(f"{ax.value}: {key} absent from info.cgi")
+ continue
+ try:
+ # Re-issue the native value verbatim: no unit round-trip to get wrong.
+ self._request("control", {key: int(round(float(raw)))})
+ except CanonTransportError as exc:
+ failures.append(f"{ax.value}: {exc}")
+ if failures:
+ raise CanonCommandError("stop failed for " + "; ".join(failures))
# -- focus mode --------------------------------------------------------
diff --git a/tests/tests_devices/test_canon_motion_check.py b/tests/tests_devices/test_canon_motion_check.py
index 9bbbedf0..383b4759 100644
--- a/tests/tests_devices/test_canon_motion_check.py
+++ b/tests/tests_devices/test_canon_motion_check.py
@@ -32,6 +32,7 @@ def _args(**kw):
poll=0.0,
timeout=5.0,
test_stop=False,
+ find_stop=False,
stop_after=0.1,
settle=0.1,
)
@@ -147,3 +148,38 @@ def test_stop_test_dry_run_sends_nothing():
report = run(_args(delta=1.0, test_stop=True), transport=fake)
assert report["dry_run"] is True
assert not any(c.startswith("move_absolute") for c in fake.call_log)
+
+
+def test_find_stop_dry_run_sends_nothing():
+ fake = _moving_fake()
+ report = run(_args(delta=1.0, find_stop=True), transport=fake)
+ assert report["dry_run"] is True
+ assert not any(c.startswith("move_absolute") for c in fake.call_log)
+
+
+def test_find_stop_tries_every_candidate_and_names_a_winner():
+ """FakeTransport halts on any control write, so all candidates should 'work'.
+
+ The point of the test is the mechanics: each candidate gets its own move, the
+ result is classified, and a winner is reported.
+ """
+ from csaxs_bec.devices.canon.motion_check import _STOP_CANDIDATES
+
+ class RecordingFake(FakeTransport):
+ def __init__(self, **kw):
+ super().__init__(**kw)
+ self.control_writes = []
+
+ def _request(self, cgi_key, params=None, *, timeout=None):
+ self.control_writes.append(params)
+ self.stop() # a working native stop would halt the axis
+
+ fake = RecordingFake(gradual_motion=True)
+ fake.connect()
+ report = run(
+ _args(delta=1.0, speed=2.0, go=True, find_stop=True, stop_after=0.05, settle=0.05),
+ transport=fake,
+ )
+ assert len(report["candidates"]) == len(_STOP_CANDIDATES)
+ assert len(fake.control_writes) == len(_STOP_CANDIDATES)
+ assert report["native_stop"], "a halting candidate should be named"
diff --git a/tests/tests_devices/test_canon_transport.py b/tests/tests_devices/test_canon_transport.py
index a3adf55b..9aa02bb5 100644
--- a/tests/tests_devices/test_canon_transport.py
+++ b/tests/tests_devices/test_canon_transport.py
@@ -452,3 +452,66 @@ def test_transport_still_capture_is_optional_not_mandatory():
t.connect()
with pytest.raises(CanonCommandError, match="does not support single-frame capture"):
t.get_still_jpeg()
+
+
+# --------------------------------------------------------------------------- #
+# stop() -- MEASURED broken 2026-08-18, rewritten to re-target the position #
+# --------------------------------------------------------------------------- #
+
+
+def test_stop_halts_by_commanding_the_current_position():
+ """The shipped `control.cgi?stop=pan` did nothing on hardware.
+
+ stop() now re-issues each axis's *current* position as a move target, which is
+ built on the one motion primitive proven to work on this camera.
+ """
+ opener = RecordingOpener(
+ bodies={"open.cgi": REAL_OPEN_BODY, "info.cgi": "c.1.pan:=-1559\nc.1.tilt:=-2226\n"}
+ )
+ t = make_xc(opener)
+ t.connect()
+ t.claim_control()
+ opener.requests.clear()
+ t.stop(Axis.PAN)
+
+ control = [u for u in opener.requests if "control.cgi" in u]
+ assert len(control) == 1
+ assert "c.1.pan=-1559" in control[0], "must re-target the position it is at"
+ # The old, broken form must be gone for good.
+ assert not any("stop=" in u for u in opener.requests)
+
+
+def test_stop_all_samples_positions_once_not_once_per_axis():
+ """This is the panic path; it must not make four 32 kB round-trips."""
+ body = "c.1.pan:=100\nc.1.tilt:=200\nc.1.zoom:=1406\nc.1.focus.value:=1114\n"
+ opener = RecordingOpener(bodies={"open.cgi": REAL_OPEN_BODY, "info.cgi": body})
+ t = make_xc(opener)
+ t.connect()
+ t.claim_control()
+ opener.requests.clear()
+ t.stop()
+
+ assert len([u for u in opener.requests if "info.cgi" in u]) == 1
+ control = [u for u in opener.requests if "control.cgi" in u]
+ assert len(control) == 4
+ joined = " ".join(control)
+ for expected in ("c.1.pan=100", "c.1.tilt=200", "c.1.zoom=1406", "c.1.focus.value=1114"):
+ assert expected in joined
+
+
+def test_stop_requires_control_privilege():
+ opener = RecordingOpener(bodies={"open.cgi": REAL_OPEN_BODY, "info.cgi": "c.1.pan:=0\n"})
+ t = make_xc(opener)
+ t.connect()
+ with pytest.raises(CanonControlPrivilegeError):
+ t.stop(Axis.PAN)
+
+
+def test_stop_reports_an_axis_it_cannot_read_rather_than_failing_silently():
+ """Silent failure is the whole bug this replaced; do not reintroduce it."""
+ opener = RecordingOpener(bodies={"open.cgi": REAL_OPEN_BODY, "info.cgi": "c.1.tilt:=0\n"})
+ t = make_xc(opener)
+ t.connect()
+ t.claim_control()
+ with pytest.raises(CanonCommandError, match="pan"):
+ t.stop(Axis.PAN)