feat(canon): verify stop() actually halts an axis before trusting it (R9)
Motion works, which makes the next question urgent: does stop() stop? stop() sends `control.cgi?stop=<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 probably not something this firmware understands. On its own that would be a mild problem. What makes it the most dangerous open item is that **this server answers HTTP 200 for CGIs it does not implement**, 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 be quietly doing nothing, and the first time anyone found out would be the first time they needed it. motion_check.py --test-stop settles it empirically: command a move, interrupt it partway, then watch whether the axis froze or carried on to the target. It is bounded either way -- worst case is the move it was already going to make -- and it distinguishes "froze" from "kept going" from "the move finished before we could interrupt it" (reported as inconclusive rather than as a pass). COMMISSIONING: R9 rewritten with the evidence and the command. Stage 4.5 (jog) is now gated on R9 passing -- jog starts *continuous* motion, so a non-working stop means the axis runs to its limit. Noted there that the jog encoding is itself a guess, and that the measured data suggests c.1.pan.speed is a setting used by moves rather than a jog command, so jog most likely does nothing at all. Tests: 98. The new ones cover the axis freezing, the expected hardware failure (a stop that is accepted and ignored -- exercised with a transport whose stop() is a no-op, which is what a silent 200 looks like), the inconclusive case, and the dry run. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
a2887fcfe2
commit
a84555bb84
3 files changed
+173
-8
No files matched your search
@@ -174,7 +174,7 @@ footprint behind one grab API. Default is **`jpeg_poll`**.
|
||||
|
||||
## 3. What has already been tested
|
||||
|
||||
**94 automated tests, no hardware required** (`pytest tests/tests_devices/test_canon_*.py
|
||||
**98 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,8 +308,25 @@ 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.** `control.cgi?stop=all|<axis>`. *Symptom:* stop
|
||||
doesn't halt motion. *Fix:* `_PARAM["stop"]` and `stop()`.
|
||||
<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.
|
||||
|
||||
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.
|
||||
|
||||
*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.
|
||||
|
||||
<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
|
||||
@@ -354,7 +371,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 94 passed.* If not, stop — the environment is wrong, not the
|
||||
--random-order`. *Expect 98 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`
|
||||
@@ -411,7 +428,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:** 94 green + the canon worktree is the live install + sim device usable in BEC +
|
||||
**Gate:** 98 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**
|
||||
@@ -592,8 +609,13 @@ move `Status` completes. → never completes ⇒ [R5](#r5) (scale so tolerance n
|
||||
4.4 **Speed.** Move pan with `pan_speed` set low vs. high; confirm the camera honours it.
|
||||
→ ignored ⇒ [R5](#r5) speed param name.
|
||||
|
||||
4.5 **Jog + stop.** `dev.ptz_cam.jog_pan(5)`, watch it pan, then `dev.ptz_cam.jog_pan(0)`
|
||||
or `stop_all()`. *Confirm* it starts and **stops promptly**. → won't stop ⇒ [R9](#r9).
|
||||
4.5 **Jog + stop.** ⚠️ **Do not attempt until [R9](#r9) is verified** — jog starts
|
||||
*continuous* motion, so a non-working stop means the axis runs to its limit. The jog
|
||||
encoding is itself a guess (a signed speed with no target), and the measured data
|
||||
suggests `c.1.pan.speed` is a *setting* used by moves rather than a jog command, so the
|
||||
likely outcome is that jog does nothing at all. Once stop is proven:
|
||||
`dev.ptz_cam.jog_pan(5)`, watch it pan, then `dev.ptz_cam.jog_pan(0)` or `stop_all()`.
|
||||
*Confirm* it starts and **stops promptly**.
|
||||
|
||||
4.6 **Through BEC.** Repeat 4.2 and 4.5 as `dev.ptz_cam.pan.move(...)` / `stop_all()` in a
|
||||
BEC session. *Confirm* identical behaviour and that `read()` reflects positions.
|
||||
|
||||
@@ -10,6 +10,9 @@ Usage -- always dry-run first::
|
||||
python3 motion_check.py --host <ip> --axis pan --delta 1 # prints the plan
|
||||
python3 motion_check.py --host <ip> --axis pan --delta 1 --go # actually moves
|
||||
|
||||
# verify stop() actually halts the axis -- do this early (see below)
|
||||
python3 motion_check.py --host <ip> --axis pan --delta 5 --test-stop --go
|
||||
|
||||
Credentials come from ``CANON_PROBE_USER`` / ``CANON_PROBE_PASSWORD``.
|
||||
|
||||
Safety behaviour
|
||||
@@ -23,6 +26,24 @@ Safety behaviour
|
||||
* Polls ``c.1.<axis>.status`` and the readback, and reports whether the axis actually
|
||||
arrived, how far it went, and whether the direction matched the command.
|
||||
|
||||
Verify stop() early -- it is a guess, and it fails silently
|
||||
-----------------------------------------------------------
|
||||
``stop()`` sends ``control.cgi?stop=<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 probably not a
|
||||
thing this firmware understands.
|
||||
|
||||
That would be a mild problem on its own. What makes it worth testing before anything
|
||||
else is that **this server answers HTTP 200 for every request**, including ones it does
|
||||
not implement: a no-op stop looks exactly like a successful stop. Everything that
|
||||
depends on halting an axis in a hurry -- ``stop_all()``, the device's ``on_stop()``
|
||||
abort hook, this tool's own cleanup -- would be quietly doing nothing.
|
||||
|
||||
``--test-stop`` commands a move, interrupts it partway with ``stop()``, and then watches
|
||||
whether the axis actually froze or carried on to the target. It is bounded either way:
|
||||
the worst case is the move it was already going to make.
|
||||
|
||||
What Stage 4 is really testing
|
||||
------------------------------
|
||||
The *scale* is already confirmed from info.cgi (pan/tilt in hundredths of a degree,
|
||||
@@ -76,6 +97,53 @@ def _import_transport():
|
||||
return XCTransport, Axis, HARDWARE_LIMITS
|
||||
|
||||
|
||||
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}
|
||||
transport.claim_control()
|
||||
print("\ncontrol claimed")
|
||||
speed = args.speed if axis.value in ("pan", "tilt") else None
|
||||
transport.move_absolute(axis, target, speed=speed)
|
||||
report["sent"] = True
|
||||
print(f"move commanded (speed {speed if speed is not None else 'default'})")
|
||||
print(f"interrupting after {args.stop_after} s ...")
|
||||
|
||||
time.sleep(args.stop_after)
|
||||
at_stop = transport.get_position(axis)
|
||||
transport.stop(axis)
|
||||
print(f" stop() sent with the axis at {at_stop:+.3f} {unit}")
|
||||
|
||||
time.sleep(args.settle)
|
||||
after = transport.get_position(axis)
|
||||
drift = after - at_stop
|
||||
remaining = abs(target - at_stop)
|
||||
report.update({"pos_at_stop": at_stop, "pos_after": after, "drift_after_stop": drift})
|
||||
print(f" {args.settle:.1f} s later : {after:+.3f} {unit} (drift {drift:+.3f})")
|
||||
|
||||
print("\n" + "-" * 72)
|
||||
if remaining <= args.tolerance:
|
||||
print("INCONCLUSIVE: the move had already finished before stop() was sent.")
|
||||
print(" Use a larger --delta or a lower --speed so there is travel to interrupt.")
|
||||
report["stop_works"] = None
|
||||
elif abs(drift) <= args.tolerance:
|
||||
print("RESULT: the axis froze where it was -- stop() WORKS.")
|
||||
print(f" It halted {remaining:.3f} {unit} short of the target.")
|
||||
report["stop_works"] = True
|
||||
elif abs(after - target) <= args.tolerance:
|
||||
print("RESULT: the axis carried on to the target -- **stop() DID NOTHING**.")
|
||||
print(" Expected: no bare 'stop' field exists in info.cgi. The camera")
|
||||
print(" uses '<namespace>.action' instead, so _PARAM['stop'] and")
|
||||
print(" XCTransport.stop() need correcting (risk item R9) before")
|
||||
print(" anything relies on halting an axis -- including stop_all(),")
|
||||
print(" the device's on_stop() abort hook, and this tool's cleanup.")
|
||||
report["stop_works"] = False
|
||||
else:
|
||||
print("RESULT: the axis kept moving but did not reach the target -- unclear.")
|
||||
print(" Re-run; if it is repeatable, treat stop() as unreliable.")
|
||||
report["stop_works"] = False
|
||||
return report
|
||||
|
||||
|
||||
def run(args, transport=None) -> dict: # pylint: disable=too-many-branches,too-many-statements
|
||||
XCTransport, Axis, HARDWARE_LIMITS = _import_transport()
|
||||
axis = Axis(args.axis)
|
||||
@@ -121,12 +189,21 @@ def run(args, transport=None) -> dict: # pylint: disable=too-many-branches,too-
|
||||
return report
|
||||
|
||||
if not args.go:
|
||||
print("\nDRY RUN -- nothing sent. Re-run with --go to move.")
|
||||
if 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.")
|
||||
else:
|
||||
print("\nDRY RUN -- nothing sent. Re-run with --go to move.")
|
||||
print("Before you do: confirm the camera has clearance in that direction,")
|
||||
print("and be ready to cut power.")
|
||||
report["dry_run"] = True
|
||||
return report
|
||||
|
||||
if args.test_stop:
|
||||
report.update(_run_stop_test(transport, axis, args, unit, start, target))
|
||||
return report
|
||||
|
||||
transport.claim_control()
|
||||
print("\ncontrol claimed")
|
||||
speed = args.speed if axis.value in ("pan", "tilt") else None
|
||||
@@ -203,6 +280,13 @@ def main(argv: list[str] | None = None) -> int:
|
||||
parser.add_argument("--axis", required=True, choices=sorted(_AXIS_UNITS))
|
||||
parser.add_argument("--delta", type=float, required=True, help="relative move, axis units")
|
||||
parser.add_argument("--go", action="store_true", help="actually send the move")
|
||||
parser.add_argument(
|
||||
"--test-stop",
|
||||
action="store_true",
|
||||
help="interrupt the move with stop() and check the axis really halts (R9)",
|
||||
)
|
||||
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")
|
||||
parser.add_argument("--max-delta", type=float, default=5.0, help="refuse larger moves")
|
||||
parser.add_argument("--tolerance", type=float, default=0.2)
|
||||
|
||||
@@ -31,6 +31,9 @@ def _args(**kw):
|
||||
move_timeout=5.0,
|
||||
poll=0.0,
|
||||
timeout=5.0,
|
||||
test_stop=False,
|
||||
stop_after=0.1,
|
||||
settle=0.1,
|
||||
)
|
||||
base.update(kw)
|
||||
return argparse.Namespace(**base)
|
||||
@@ -88,3 +91,59 @@ def test_zoom_moves_without_a_speed_argument(fake):
|
||||
report = run(_args(axis="zoom", delta=1.0, go=True), transport=fake)
|
||||
assert report["sent"] is True
|
||||
assert "move_absolute:zoom" in fake.call_log
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# --test-stop: does stop() actually halt the axis? (risk item R9) #
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def _moving_fake():
|
||||
"""A fake that travels at a finite speed, so a move can be interrupted."""
|
||||
t = FakeTransport(gradual_motion=True)
|
||||
t.connect()
|
||||
return t
|
||||
|
||||
|
||||
def test_stop_test_reports_success_when_the_axis_freezes():
|
||||
fake = _moving_fake()
|
||||
report = run(_args(delta=1.0, speed=2.0, go=True, test_stop=True), transport=fake)
|
||||
assert report["stop_test"] is True
|
||||
assert report["stop_works"] is True
|
||||
# It halted short of the target, and did not drift afterwards.
|
||||
assert abs(report["pos_at_stop"] - report["target"]) > _args().tolerance
|
||||
assert abs(report["drift_after_stop"]) <= _args().tolerance
|
||||
|
||||
|
||||
def test_stop_test_detects_a_stop_that_does_nothing():
|
||||
"""The expected hardware failure: stop() is accepted, and has no effect.
|
||||
|
||||
The camera answers HTTP 200 for unimplemented CGIs, so a no-op stop is
|
||||
indistinguishable from a working one at the protocol level. Only watching the
|
||||
axis afterwards tells them apart -- which is what this mode is for.
|
||||
"""
|
||||
|
||||
class DeafTransport(FakeTransport):
|
||||
def stop(self, axis=None): # accepted, ignored -- like a 200 with no effect
|
||||
self._record("stop_ignored")
|
||||
|
||||
fake = DeafTransport(gradual_motion=True)
|
||||
fake.connect()
|
||||
report = run(_args(delta=1.0, speed=2.0, go=True, test_stop=True, settle=0.8), transport=fake)
|
||||
assert report["stop_works"] is False
|
||||
assert report["pos_after"] == pytest.approx(report["target"], abs=0.05)
|
||||
|
||||
|
||||
def test_stop_test_says_inconclusive_when_the_move_finishes_first():
|
||||
"""Instant motion leaves nothing to interrupt -- do not claim stop() works."""
|
||||
fake = FakeTransport() # not gradual: moves complete immediately
|
||||
fake.connect()
|
||||
report = run(_args(delta=1.0, go=True, test_stop=True), transport=fake)
|
||||
assert report["stop_works"] is None
|
||||
|
||||
|
||||
def test_stop_test_dry_run_sends_nothing():
|
||||
fake = _moving_fake()
|
||||
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)
|
||||
Reference in new issue
Block a user