diff --git a/csaxs_bec/devices/canon/COMMISSIONING.md b/csaxs_bec/devices/canon/COMMISSIONING.md index 320941c7..ee3b8a37 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 -**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. **R3 — Preset CGI.** `preset.cgi?preset=`. *Symptom:* preset recall errors or no-op. *Fix:* `_CGI["preset"]` / `_PARAM["preset"]`. -**R9 — Stop command.** `control.cgi?stop=all|`. *Symptom:* stop -doesn't halt motion. *Fix:* `_PARAM["stop"]` and `stop()`. +**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. + +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 --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. **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. diff --git a/csaxs_bec/devices/canon/motion_check.py b/csaxs_bec/devices/canon/motion_check.py index 202664f3..4449a1d0 100644 --- a/csaxs_bec/devices/canon/motion_check.py +++ b/csaxs_bec/devices/canon/motion_check.py @@ -10,6 +10,9 @@ Usage -- always dry-run first:: python3 motion_check.py --host --axis pan --delta 1 # prints the plan python3 motion_check.py --host --axis pan --delta 1 --go # actually moves + # verify stop() actually halts the axis -- do this early (see below) + python3 motion_check.py --host --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..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=``, 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 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 '.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) diff --git a/tests/tests_devices/test_canon_motion_check.py b/tests/tests_devices/test_canon_motion_check.py index 2d8c7014..9bbbedf0 100644 --- a/tests/tests_devices/test_canon_motion_check.py +++ b/tests/tests_devices/test_canon_motion_check.py @@ -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)