feat(canon): add a guarded Stage 4 motion tool

The camera is mounted, so motion is unblocked. Stage 4 is the one
safety-critical stage -- it is the first time software moves hardware -- and
the axis scale being already confirmed makes a mistyped command genuinely
consequential: a 170-degree slew is one keystroke from a 1-degree one.

motion_check.py is therefore deliberately awkward to misuse:

* dry run is the default; nothing is sent without --go
* moves are relative and capped at --max-delta (5 units), so a typo cannot
  become a full-range slew without deliberately raising the cap
* the target is checked against the measured hardware range before sending
* 5 deg/s by default, against a 100 deg/s ceiling
* the axis is stopped and control yielded in a `finally` -- including on
  Ctrl-C and on any exception, so an aborted run never leaves the camera
  moving or the control privilege held

It polls c.1.<axis>.status and the readback, then classifies the outcome:
arrived as commanded, arrived but in the opposite direction (which would mean
negating the scale in _AXIS_ENCODING), moved by the wrong amount, or never
settled -- each with what to suspect. It deliberately does not claim to judge
*physical* direction: it prints a prompt to look at the live view, because
only a human can say whether "+1 degree" went the way that was intended.

Same three-route import as the other tools, so it runs from a copied canon/
directory on a bare python3.

Tests: 94. The six new ones verify the guards actually guard -- dry run sends
nothing, oversized and out-of-range moves are refused before any command, and
control is yielded and the axis stopped even when the move itself raises.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
menzelandClaude Opus 5 committed 2026-08-31 16:41:06 +02:00
1 parent 5c3f632def
commit a2887fcfe2
4 files changed
+340 -6

No files matched your search

+18 -3
View File
@@ -174,7 +174,7 @@ footprint behind one grab API. Default is **`jpeg_poll`**.
## 3. What has already been tested
**88 automated tests, no hardware required** (`pytest tests/tests_devices/test_canon_*.py
**94 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
@@ -354,7 +354,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 88 passed.* If not, stop — the environment is wrong, not the
--random-order`. *Expect 94 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 +411,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:** 88 green + the canon worktree is the live install + sim device usable in BEC +
**Gate:** 94 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**
@@ -562,6 +562,21 @@ behaves sensibly (errors or waits, not silent).
Do this through the transport first (tightest control), then through the ophyd device.
4.0 **Use the guarded tool, dry run first.** `motion_check.py` is the only tool here
that moves the camera. It refuses moves larger than `--max-delta`, checks the target
against the hardware range before sending, runs slowly by default, and always stops and
yields in a `finally` — including on Ctrl-C.
```bash
cd ~/scratch/csaxs_bec/csaxs_bec/devices/canon
python3 motion_check.py --host <ip> --axis pan --delta 1 # prints the plan only
python3 motion_check.py --host <ip> --axis pan --delta 1 --go # actually moves
```
It reports whether the axis arrived, by how much, and whether the direction matched the
command — and then tells you to look at the live view, because only a human can say
whether "+1°" went the way you intended.
4.1 **Smallest relative pan.** From a known position, command `+1°`. *Confirm* it moves
**~1° in the expected direction**. → wrong size ⇒ [R5](#r5) scale; wrong direction ⇒
[R5](#r5) sign.
+9 -3
View File
@@ -28,7 +28,8 @@ devices/canon/
├── positioner.py # CanonAxis — transport-backed ophyd positioner
├── canon_ptz.py # CanonPTZCamera — the BEC ophyd device
├── probe.py # operator tool: read-only first-contact probe
└── acquisition_check.py # operator tool: Stage 7 acquisition & bandwidth check
├── acquisition_check.py # operator tool: Stage 7 acquisition & bandwidth check
└── motion_check.py # operator tool: Stage 4 guarded motion (the only one that moves)
```
Key boundaries:
@@ -258,10 +259,15 @@ Two read-only operator tools, both standard-library only and runnable from a cop
`canon/` directory with nothing installed:
```bash
python3 probe.py --host <ip> # first contact: reachability, CGIs, session
python3 acquisition_check.py --host <ip> # Stage 7: establish, rate, bandwidth, grabs
python3 probe.py --host <ip> # first contact
python3 acquisition_check.py --host <ip> # Stage 7: rate, bandwidth, grabs
python3 motion_check.py --host <ip> --axis pan --delta 1 # Stage 4: dry run
```
The first two are read-only. **`motion_check.py` moves the camera** — it dry-runs unless
given `--go`, caps the step size, checks the target against the hardware range first, and
always stops the axis and yields control in a `finally`, including on Ctrl-C.
`acquisition_check.py` drives the production acquisition path — the same poll source,
worker and ring buffer the device uses — and reports establishment time, achieved rate,
measured bandwidth, grab latency and (with `--decode`) a real JPEG decode. It never sends
+223
View File
@@ -0,0 +1,223 @@
"""Commissioning Stage 4 -- the first motion, one guarded step at a time.
**This is the only tool here that moves the camera.** `probe.py` and
`acquisition_check.py` are read-only; this one is not. It exists so the first physical
moves are small, deliberate, bounded and always followed by a stop and a yield --
rather than a raw one-liner where a typo becomes a 170-degree slew.
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
Credentials come from ``CANON_PROBE_USER`` / ``CANON_PROBE_PASSWORD``.
Safety behaviour
----------------
* **Dry run is the default.** Nothing is sent without ``--go``.
* **Relative, small, bounded.** The move is *current + delta*; ``--delta`` is capped at
``--max-delta`` (5 units by default) unless you raise it deliberately.
* **Soft-limit checked before sending**, against the measured hardware range.
* **Slow by default** -- 5 deg/s for pan/tilt, well under the 100 deg/s ceiling.
* **Always stops and yields**, including on Ctrl-C or any exception, in a ``finally``.
* 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.
What Stage 4 is really testing
------------------------------
The *scale* is already confirmed from info.cgi (pan/tilt in hundredths of a degree,
+/-17000 and -3000..10000). What is unverified is:
1. whether writing ``c.1.<axis>`` on control.cgi commands a move at all;
2. the **direction sign** -- whether +1 deg goes the way you expect;
3. whether the move **completes** and the status flag returns to idle;
4. whether ``c.1.<axis>.speed`` is honoured.
Only a human watching the live view can answer (2) in physical terms, so the tool
prints a prompt to look rather than pretending to decide it.
"""
from __future__ import annotations
import argparse
import os
import sys
import time
_AXIS_UNITS = {"pan": "deg", "tilt": "deg", "zoom": "x", "focus": "norm"}
def _import_transport():
"""Same three-route import as probe.py, so a copied canon/ directory works."""
try:
from .transport.base import HARDWARE_LIMITS, Axis # pylint: disable=C0415
from .transport.xc import XCTransport # pylint: disable=C0415
return XCTransport, Axis, HARDWARE_LIMITS
except ImportError:
pass
try:
from csaxs_bec.devices.canon.transport.base import ( # pylint: disable=C0415
HARDWARE_LIMITS,
Axis,
)
from csaxs_bec.devices.canon.transport.xc import XCTransport # pylint: disable=C0415
return XCTransport, Axis, HARDWARE_LIMITS
except ImportError:
pass
here = os.path.dirname(os.path.abspath(__file__))
if here not in sys.path:
sys.path.insert(0, here)
# Resolved at runtime via the sys.path insert above.
from transport.base import HARDWARE_LIMITS, Axis # pylint: disable=C0415,import-error
from transport.xc import XCTransport # pylint: disable=C0415,import-error
return XCTransport, Axis, HARDWARE_LIMITS
def run(args, transport=None) -> dict: # pylint: disable=too-many-branches,too-many-statements
XCTransport, Axis, HARDWARE_LIMITS = _import_transport()
axis = Axis(args.axis)
limits = HARDWARE_LIMITS[axis]
unit = _AXIS_UNITS[args.axis]
report: dict = {"axis": args.axis, "delta": args.delta, "sent": False}
if abs(args.delta) > args.max_delta:
print(
f"REFUSED: |delta| {abs(args.delta)} exceeds --max-delta {args.max_delta}.\n"
" Raise it deliberately once small moves are proven."
)
report["refused"] = "delta too large"
return report
owned = transport is None
if transport is None:
transport = XCTransport(
host=args.host,
port=args.port,
username=os.environ.get("CANON_PROBE_USER"),
password=os.environ.get("CANON_PROBE_PASSWORD"),
timeout=args.timeout,
)
try:
transport.connect()
start = transport.get_position(axis)
target = start + args.delta
report.update({"start": start, "target": target})
print("=" * 72)
print(
f"Stage 4 -- {args.axis}: {start:+.3f} -> {target:+.3f} {unit} "
f"(delta {args.delta:+g})"
)
print(f"hardware range {limits.low:+g} .. {limits.high:+g} {unit}")
print("=" * 72)
if not limits.low <= target <= limits.high:
print(f"REFUSED: target {target:+.3f} is outside the hardware range.")
report["refused"] = "target outside hardware range"
return report
if not args.go:
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
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'})")
deadline = time.monotonic() + args.move_timeout
settled_at = None
while time.monotonic() < deadline:
pos = transport.get_position(axis)
moving = transport.is_moving(axis)
print(f" {pos:+9.3f} {unit} moving={moving}")
if not moving and abs(pos - target) <= args.tolerance:
settled_at = pos
break
time.sleep(args.poll)
final = transport.get_position(axis)
achieved = final - start
report.update(
{
"final": final,
"achieved_delta": achieved,
"settled": settled_at is not None,
"elapsed_s": args.move_timeout - max(0.0, deadline - time.monotonic()),
}
)
print("\n" + "-" * 72)
print(f"commanded delta : {args.delta:+.3f} {unit}")
print(f"achieved delta : {achieved:+.3f} {unit}")
print(f"final position : {final:+.3f} {unit} (target {target:+.3f})")
if settled_at is None:
print("\nRESULT: did NOT settle within the timeout.")
print(" - readback unchanged -> writing c.1.<axis> may not command a move")
print(" - readback moving -> raise --move-timeout or lower --speed")
print(" - readback overshoots -> check --tolerance against the axis scale")
elif abs(achieved - args.delta) <= args.tolerance:
print("\nRESULT: moved by the commanded amount, and the status flag returned")
print(" to idle. Scale and completion confirmed for this axis.")
elif abs(achieved + args.delta) <= args.tolerance:
print("\nRESULT: moved the commanded amount but in the OPPOSITE direction.")
print(" The sign convention is inverted -- negate the scale for this")
print(" axis in _AXIS_ENCODING before going any further.")
else:
print("\nRESULT: moved, but not by the commanded amount. Suspect the scale.")
print("\nNow LOOK at the live view: did it move the direction you intended?")
print("Only you can answer that; the numbers above only prove self-consistency.")
return report
finally:
try:
transport.stop(axis)
except Exception as exc: # pylint: disable=broad-except
print(f"warning: stop failed: {exc}")
try:
transport.yield_control()
except Exception as exc: # pylint: disable=broad-except
print(f"warning: yield failed: {exc}")
if owned:
transport.close()
print("stopped, control yielded, session closed")
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(
description="Stage 4 guarded motion check for the Canon CR-N300.",
epilog="Dry run unless --go is given. Always stops and yields on exit.",
)
parser.add_argument("--host", required=True)
parser.add_argument("--port", type=int, default=80)
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("--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)
parser.add_argument("--move-timeout", type=float, default=30.0)
parser.add_argument("--poll", type=float, default=0.5)
parser.add_argument("--timeout", type=float, default=5.0)
args = parser.parse_args(argv)
try:
report = run(args)
except KeyboardInterrupt:
print("\ninterrupted -- axis stopped and control yielded")
return 130
return 0 if report.get("sent") or report.get("dry_run") else 1
if __name__ == "__main__":
sys.exit(main())
@@ -0,0 +1,90 @@
"""Tests for the Stage 4 motion tool -- the only tool that moves the camera.
Everything here runs against FakeTransport, so the safety behaviour is verified
without hardware. The point of these is that the guards actually guard: a dry run
must send nothing, an oversized or out-of-range move must be refused, and control
must be yielded even when the move raises.
"""
from __future__ import annotations
import argparse
import pytest
from csaxs_bec.devices.canon.motion_check import run
from csaxs_bec.devices.canon.transport.base import Axis
from csaxs_bec.devices.canon.transport.errors import CanonCommandError
from csaxs_bec.devices.canon.transport.fake import FakeTransport
def _args(**kw):
base = dict(
host="cam.test",
port=80,
axis="pan",
delta=1.0,
go=False,
speed=5.0,
max_delta=5.0,
tolerance=0.2,
move_timeout=5.0,
poll=0.0,
timeout=5.0,
)
base.update(kw)
return argparse.Namespace(**base)
@pytest.fixture
def fake():
t = FakeTransport()
t.connect()
return t
def test_dry_run_sends_nothing(fake):
report = run(_args(), transport=fake)
assert report["dry_run"] is True
assert report["sent"] is False
assert not any(c.startswith("move_absolute") for c in fake.call_log)
assert fake.get_position(Axis.PAN) == 0.0
def test_oversized_delta_is_refused(fake):
report = run(_args(delta=90.0, go=True), transport=fake)
assert report["refused"] == "delta too large"
assert report["sent"] is False
assert not any(c.startswith("move_absolute") for c in fake.call_log)
def test_target_outside_hardware_range_is_refused(fake):
fake.claim_control()
fake.move_absolute(Axis.PAN, 169.0)
report = run(_args(delta=3.0, go=True), transport=fake)
assert report["refused"] == "target outside hardware range"
assert report["sent"] is False
def test_go_moves_and_reports_the_achieved_delta(fake):
report = run(_args(delta=1.0, go=True), transport=fake)
assert report["sent"] is True
assert report["settled"] is True
assert report["achieved_delta"] == pytest.approx(1.0)
assert report["final"] == pytest.approx(1.0)
def test_control_is_yielded_and_axis_stopped_even_on_failure(fake):
"""The finally block is the safety net: a raising move must still release."""
fake.fail_next["move_absolute:pan"] = CanonCommandError("boom")
with pytest.raises(CanonCommandError):
run(_args(go=True), transport=fake)
assert not fake.has_control
assert "stop:pan" in fake.call_log
def test_zoom_moves_without_a_speed_argument(fake):
"""Speed is a pan/tilt concept; zoom must not be handed one."""
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