fix(canon): make stop() actually stop -- it was a silent no-op on hardware
CI for csaxs_bec / test (push) Failing after 25s

Verified against the camera on 2026-08-18: `control.cgi?stop=<axis>` does
nothing. The suspicion recorded yesterday was right -- no bare `stop` field
exists on this camera, it expresses actions as `<namespace>.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 doing nothing.

stop() no longer sends a stop command. 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:

* protocol-independent -- any camera accepting absolute moves stops this way,
  whatever it calls its stop command, so this survives firmware differences;
* positions are sampled **once** for all axes rather than once per axis,
  because this is the panic path and must not be slow;
* the axis decelerates on its normal ramp rather than dead-stopping, and may
  drift slightly past the sampled position before settling. It is a halt, not
  a freeze, and the docstring says so.

_PARAM["stop"] is gone rather than left as a misleading dead entry.

Also adds `motion_check.py --find-stop`, which hunts for a *native*
one-request stop by interrupting a series of moves with each candidate
(`c.1.pan.action=stop`, `c.1.action=stop`, `p.action=stop`, ...). Worth having
because the re-target must read ~32 kB of info.cgi before it can act, and at
100 deg/s that round-trip is real extra travel on the path taken when
something is already wrong. If a candidate halts the axis it becomes the fast
path -- with the re-target kept as the fallback, since a silent no-op is
precisely what the fallback protects against.

Tests: 104. Coverage for re-targeting the sampled position, sampling info.cgi
once for stop_all rather than four times, requiring control privilege,
raising rather than silently skipping an axis it cannot read (the original
bug's failure mode), and the candidate hunt's mechanics.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-08-18 08:44:12 +02:00
co-authored by Claude Opus 5
parent aacc9e659b
commit 5d6f2374ff
5 changed files with 241 additions and 24 deletions
+19 -20
View File
@@ -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.
<a name="r3"></a>**R3 — Preset CGI.** `preset.cgi?preset=<id>`. *Symptom:* preset recall
errors or no-op. *Fix:* `_CGI["preset"]` / `_PARAM["preset"]`.
<a name="r9"></a>**R9 — Stop command. UNVERIFIED, and it fails silently.**
`stop()` sends `control.cgi?stop=all|<axis>`, but **no bare `stop` field appears anywhere
in the camera's 32 kB info.cgi response**. The camera expresses actions as
`<namespace>.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.
<a name="r9"></a>**R9 — Stop command. CONFIRMED BROKEN as shipped; now fixed.**
`stop()` used to send `control.cgi?stop=<axis>`. **Verified against hardware on
[[2026-08-18]]: it did nothing.** No bare `stop` field exists on this camera — actions
are `<namespace>.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 <ip> --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.
<a name="r10"></a>**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**
+83 -1
View File
@@ -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 "<namespace>.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")
+40 -3
View File
@@ -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=<axis>`` **did nothing**.
No bare ``stop`` field exists on this camera -- it expresses actions as
``<namespace>.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 --------------------------------------------------------
@@ -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"
@@ -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)