diff --git a/csaxs_bec/device_configs/canon_crn300.yaml b/csaxs_bec/device_configs/canon_crn300.yaml new file mode 100644 index 0000000..b3bdd7d --- /dev/null +++ b/csaxs_bec/device_configs/canon_crn300.yaml @@ -0,0 +1,51 @@ +# Example BEC device-config entries for the Canon CR-N300 PTZ camera. +# +# Include these in a session by merging this file into your active config, or copy an +# entry into one of the bl_*.yaml configs. Credentials are NOT stored here: the device +# reads CANON__USER / CANON__PASSWORD from the environment at connect +# (NAME = the device name, upper-cased; e.g. CANON_PTZ_CAM_USER for ptz_cam). + +# --- Real camera (Canon XC Control Protocol over HTTP) --------------------------- +ptz_cam: + description: Canon CR-N300 PTZ remote camera (endstation overview) + deviceClass: csaxs_bec.devices.canon.canon_ptz.CanonPTZCamera + deviceConfig: + host: 192.168.0.100 # camera IP / hostname + port: 80 # XC HTTP port (default 80) + timeout: 5.0 # per-request transport timeout [s] + # cgi_prefix: "-wvhttp-01-" # XC CGI path prefix (verify vs BPE-7216-005) + # username_env: CANON_PTZ_CAM_USER # override the default env-var names + # password_env: CANON_PTZ_CAM_PASSWORD + pan_speed: 60 # optional default move speed [deg/s] + tilt_speed: 60 + # --- Phase-2 streaming (safe to leave at defaults for control-only use) --- + stream_start: "off" # off | connect | stage (when to open the stream) + buffer_maxlen: 64 # ring-buffer depth (compressed frames) + # stream_url: http://192.168.0.100/-wvhttp-01-/image.cgi # MJPEG URL override + deviceTags: + - camera + - ptz + enabled: true + onFailure: buffer + readOnly: false + readoutPriority: on_request + softwareTrigger: false + +# --- Simulation / offline (no hardware; uses FakeTransport) ---------------------- +ptz_cam_sim: + description: Canon CR-N300 PTZ camera (simulation, no hardware required) + deviceClass: csaxs_bec.devices.canon.canon_ptz.CanonPTZCamera + deviceConfig: + simulation: true + gradual_motion: true # simulate finite-speed travel (exercise move status) + buffer_maxlen: 32 + stream_start: "connect" # start the synthetic stream at connect + deviceTags: + - camera + - ptz + - simulation + enabled: true + onFailure: buffer + readOnly: false + readoutPriority: on_request + softwareTrigger: false diff --git a/csaxs_bec/devices/canon/COMMISSIONING.md b/csaxs_bec/devices/canon/COMMISSIONING.md new file mode 100644 index 0000000..9d09530 --- /dev/null +++ b/csaxs_bec/devices/canon/COMMISSIONING.md @@ -0,0 +1,454 @@ +# Canon CR-N300 — Commissioning & Verification Guide + +This document is the bring-up companion to the +[`CanonPTZCamera`](./canon_ptz.py) device. It exists so that, when the device is +deployed at the beamline against real Canon hardware, you can systematically confirm +**what works as designed** and pinpoint **where reality deviates from the assumptions +baked into the code** — without reverse-engineering the source. + +It contains: + +1. [How to read this guide](#1-how-to-read-this-guide) +2. [The UI surface — every function and its intended effect](#2-the-ui-surface) +3. [What has already been tested (and how)](#3-what-has-already-been-tested) +4. [What remains to be tested (needs hardware)](#4-what-remains-to-be-tested) +5. [The deviation-risk map — protocol assumptions to confirm](#5-deviation-risk-map) +6. [The commissioning test regime — a gated schedule](#6-commissioning-test-regime) +7. [Recording template](#7-recording-template) + +--- + +## 1. How to read this guide + +Every function below is tagged with a **verification status**: + +| Tag | Meaning | +|-----|---------| +| 🟢 **VERIFIED (sim)** | Exercised by the automated test-suite against the fake transport / mocked HTTP. The *Python logic* is proven; the *protocol wire-compatibility* is not. | +| 🟡 **LOGIC-ONLY** | The code path runs in sim, but its real effect depends on a protocol assumption that only hardware can confirm (flagged with the relevant [risk item](#5-deviation-risk-map)). | +| 🔴 **UNTESTED** | No automated coverage; requires hardware or a running BEC stack. | + +A 🟢 tag means *"if the camera speaks the protocol we assume, this works."* The job of +hardware commissioning is to confirm that assumption, function by function. When a 🟡/🔴 +step fails, the [deviation-risk map](#5-deviation-risk-map) tells you **which constant in +the code to adjust** — almost always a single table entry in +[`transport/xc.py`](./transport/xc.py), not application logic. + +Throughout, commands are shown in the BEC client form `dev..…`. The example device +name used is **`ptz_cam`** (from [`device_configs/canon_crn300.yaml`](../../device_configs/canon_crn300.yaml)). + +--- + +## 2. The UI surface + +### 2.1 Quick reference + +| Call | Kind | Intended effect | Status | +|------|------|-----------------|--------| +| `dev.ptz_cam.pan.move(deg)` / `.set(deg)` | positioner | Absolute pan, ±170° | 🟡 [R5](#r5) | +| `dev.ptz_cam.tilt.move(deg)` | positioner | Absolute tilt, −30…+100° | 🟡 [R5](#r5) | +| `dev.ptz_cam.zoom.move(x)` | positioner | Absolute zoom, 1–20× | 🟡 [R5](#r5) | +| `dev.ptz_cam.focus.move(f)` | positioner | Manual focus, 0–1 (MF only) | 🟡 [R5](#r5) | +| `dev.ptz_cam..stop()` | method | Halt one axis | 🟡 [R9](#r9) | +| `dev.ptz_cam.stop_all()` | USER_ACCESS | Halt all four axes | 🟡 [R9](#r9) | +| `dev.ptz_cam.jog_pan(speed)` / `jog_tilt(speed)` | USER_ACCESS | Continuous move at deg/s; `0` stops | 🟡 [R5](#r5) | +| `dev.ptz_cam.recall_preset(id)` | USER_ACCESS | Recall stored PTZ preset | 🟡 [R3](#r3) | +| `dev.ptz_cam.gain.set(v)` | signal (config) | Set gain, verify read-back | 🟡 [R4](#r4) | +| `dev.ptz_cam.iris.set(v)` | signal (config) | Set iris/aperture, verify read-back | 🟡 [R4](#r4) | +| `dev.ptz_cam.exposure_mode.set(s)` | signal (config) | Set exposure mode | 🟡 [R4](#r4) | +| `dev.ptz_cam.white_balance.set(s)` | signal (config) | Set white balance | 🟡 [R4](#r4) | +| `dev.ptz_cam.shutter_speed.set(s)` | signal (config) | Set shutter speed | 🟡 [R4](#r4) | +| `dev.ptz_cam.focus_mode.set(m)` | signal (config) | manual / continuous_af / face_af / tracking_af | 🟡 [R6](#r6) | +| `dev.ptz_cam.preset_state.get()` | signal (RO) | Last recalled preset id | 🟢 | +| `dev.ptz_cam.read()` / `.describe()` | ophyd | Readback of all axes + normal signals | 🟢 | +| `dev.ptz_cam.read_configuration()` | ophyd | Readback of all config parameters | 🟡 [R4](#r4) | +| `dev.ptz_cam.start_stream()` / `stop_stream()` | USER_ACCESS | Start/stop the acquisition worker | 🟡 [R10](#r10) | +| `dev.ptz_cam.get_latest()` | USER_ACCESS | Most recent buffered frame | 🟡 [R10](#r10) | +| `dev.ptz_cam.get_latest_n(n)` | USER_ACCESS | Last *n* frames, oldest-first | 🟢 | +| `dev.ptz_cam.get_nearest(ts)` | USER_ACCESS | Frame closest to a wall-clock timestamp | 🟢 | +| `dev.ptz_cam.stream_stats()` | USER_ACCESS | Buffer + worker instrumentation | 🟢 | +| `dev.ptz_cam.transport_info()` | USER_ACCESS | Connection / identity diagnostics | 🟡 [R7](#r7) | + +### 2.2 Lifecycle (called by BEC, not normally by the user) + +These run automatically when BEC connects/stages/stops the device. Listed because their +behaviour is what you are implicitly testing in Stage 1–3. + +| Hook | When | Intended effect | Status | +|------|------|-----------------|--------| +| `wait_for_connection()` | on device load | Open XC session (`open.cgi`) **and claim control** (`claim.cgi`), then wait for signals | 🟡 [R1](#r1)[R2](#r2)[R6b](#r6b) | +| `on_connected()` | after connect | Log camera identity; seed pan/tilt/zoom/focus readbacks; start stream if `stream_start: connect` | 🟡 [R7](#r7) | +| `on_stage()` | scan start | Start stream if `stream_start: stage` | 🟢 | +| `on_unstage()` | scan end | No-op (stream persists across scans by design) | 🟢 | +| `on_stop()` | abort | Halt all motion (best-effort) | 🟡 [R9](#r9) | +| `on_destroy()` | device teardown | Stop stream worker; yield control + close session (`yield.cgi`/`close.cgi`) | 🟡 [R2](#r2) | + +### 2.3 Detailed function reference + +#### Motion — `pan`, `tilt`, `zoom`, `focus` + +Each axis is a full ophyd positioner. + +- **`move(value, wait=True)`** — commands an absolute move and returns an ophyd + `Status`. With `wait=True` (default) it blocks until the readback reaches `value` + within tolerance, or the move times out (default 30 s), or `stop()` is called. + - *Intended effect:* the physical axis travels to `value` (pan/tilt in degrees, zoom + in optical ×, focus normalised 0–1) and the readback reflects it. + - *Internally:* issues `move_absolute` on the transport, then polls + `get_position` every `poll_interval` (default 0.1 s) until `|pos − target| ≤ + tolerance`. **The encoding from user units to the camera's native command integer + is assumption [R5](#r5).** + - *Guard rails:* a value outside the configured soft-limits raises `LimitError` + **before** any command is sent. Soft limits can only narrow the hardware range. +- **`set(value)`** — identical but non-blocking (`wait=False`); returns the `Status` + immediately. This is what BEC scans use. +- **`stop()`** — halts the axis immediately (sends the transport stop command) and + fails any in-flight move status. Best-effort: never raises during teardown. +- **`jog(speed)`** (pan/tilt only, exposed as `jog_pan`/`jog_tilt`) — start a + continuous move at `speed` (deg/s, sign = direction); `0` stops. **Continuous-move + encoding is assumption [R5](#r5).** +- **`position`** property and **`read()`** — current readback. +- **`limits`** — the active `(low, high)` soft limits. + +> **Focus is special:** `focus.move(...)` raises `RuntimeError` unless `focus_mode` is +> `manual`. This is enforced in software ([R6](#r6) governs the mode tokens). + +#### Camera parameters — settable config signals + +`gain`, `iris`, `exposure_mode`, `white_balance`, `shutter_speed`, `focus_mode`. + +- **`set(value)`** — writes the parameter through the transport, then (for `gain`/`iris`) + **reads it back and raises `CanonParameterError` if the read-back disagrees**. The + status returned reflects success/failure. Parameters whose camera-side representation + isn't a clean round-trip (`exposure_mode`, `white_balance`, `shutter_speed`, + `focus_mode`) have read-back verification disabled by default (`verify=False`). + - *Depends on:* the parameter **names** the camera uses in its info/control CGIs + ([R4](#r4)) and, for focus, the **mode tokens** ([R6](#r6)). +- **`get()`** — reads the live value from the camera. + +#### Presets + +- **`recall_preset(id)`** — recall a stored PTZ preset; updates `preset_state`. + ([R3](#r3) governs the preset CGI/parameter.) +- **`preset_state.get()`** — the last preset id recalled **through this device** (it is + device-side state, `-1` if none; it does **not** poll the camera). + +#### Streaming & grab (Phase 2) + +- **`start_stream()`** — opens the MJPEG stream once and starts the background worker + filling the ring buffer. Idempotent. ([R10](#r10) governs the stream URL/format.) +- **`get_latest()` / `get_latest_n(n)` / `get_nearest(ts)`** — O(1) retrieval of + buffered frames. Each returns a `Frame` (compressed bytes + monotonic & wall-clock + timestamps + source id), or pass `decode=True` to decode **only that frame** to an + RGB array via `cv2`. +- **`stream_stats()`** — returns `{"buffer": {...}, "worker": {...}}` including + `establishment_time_s` (time to first frame — should be the ~2 s, paid once) and + `last_frame_age_s` (health). +- **`stop_stream()`** — stops the worker cleanly (no leaked socket/thread). + +#### Diagnostics + +- **`transport_info()`** — `{connected, has_control, simulation, model, firmware}`. + The first thing to call when something looks wrong. + +--- + +## 3. What has already been tested + +**47 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 +do **not** touch a camera. + +Coverage by area: + +**Ring buffer (7 tests)** — rejects bad capacity; stamps sequence + monotonic/wall +timestamps; overwrites oldest and counts evictions; `get_latest` / `get_latest_n` / +`get_nearest` (wall and monotonic); thread-safety under 5 concurrent writers. + +**Stream worker (5 tests)** — fills the buffer and measures establishment time; +**fail-soft reconnect** keeps serving the last good frame after a dropped connection; +idempotent `start()`; MJPEG byte-splitting on JPEG markers **without decoding**; clean +thread shutdown with **no thread leak**. + +**Transport — `FakeTransport` (9 tests)** — session lifecycle; **control-privilege is +enforced** (moves rejected without a claim); session required for reads; absolute move +honours hardware limits; parameter round-trip; focus-mode + preset state; gradual motion +with an injected clock; stop freezes an axis; fault injection. + +**Transport — `XCTransport` via mocked HTTP (9 tests)** — parses the session id from the +response body **and** from a `Set-Cookie` header; builds control URLs carrying the +session id; **encodes native axis values** (`12.0° → pan=1200`) and decodes them back; +rejects moves without control; **maps HTTP 401 → session error, 403 → control-privilege +error**; **retries + reconnects** on a transient failure; `close()` is idempotent and +never raises; focus-mode wire-token round-trip. + +**Device logic (17 tests)** — connect claims control; the four axes are positioners with +the **spec limits** (±170 / −30…+100 / 1–20×); `move()` returns a status that reaches +target; `set()` is non-blocking; **soft limits narrow** hardware limits (`LimitError`); +parameter set **with read-back verification**, and a **mismatch fails the status**; +**focus axis is gated** by focus mode; `read()`/`describe()`/`read_configuration()` cover +axes + params; `stop_all()` halts every axis; `recall_preset`; grab API errors cleanly +with no buffer; **simulation streaming** fills the buffer and reports establishment time; +`host` required unless `simulation`; **credentials read from environment variables**; +`destroy()` closes the transport. + +--- + +## 4. What remains to be tested + +Everything in this list requires either the physical camera or a running BEC stack, and +is the actual subject of [the test regime in §6](#6-commissioning-test-regime). + +| Area | Why it can't be tested without hardware | +|------|------------------------------------------| +| **XC wire compatibility** | The camera must actually accept our CGI paths/params and return the format we parse. See every [risk item](#5-deviation-risk-map). | +| **HTTP authentication** | Real Digest/Basic handshake with the camera's credentials. | +| **Axis encoding scale** | That `pan=1200` really means 12.0° (and direction sign) on this model — [R5](#r5). | +| **Readback units & calibration** | That decoded readbacks match physical degrees/zoom/focus. | +| **Motion completion semantics** | Whether the camera exposes a reliable "moving" flag, and our tolerance-based completion is correct — [R8](#r8). | +| **Focus AF modes** | That `continuous_af`/`face_af`/`tracking_af` engage as labelled — [R6](#r6). | +| **Presets** | That `recall_preset(id)` recalls the expected stored position — [R3](#r3). | +| **Stream establishment & latency** | The real ~2 s setup cost, steady frame rate, and sub-ms grab latency — [R10](#r10). | +| **JPEG decode of real frames** | `decode=True` against real camera JPEGs (sim uses non-image bytes). | +| **BEC integration** | Entry-point resolution, Redis, scan staging, GUI preview, file-writer delivery (Phase 2). | +| **Multi-client contention** | claim/yield against another client already holding control. | +| **Long-run robustness** | Real network drops, session timeout/expiry, reconnection over hours. | + +--- + +## 5. Deviation-risk map + +Each item is a protocol assumption the code makes. **All live in one place** — +the tables at the top of [`transport/xc.py`](./transport/xc.py) — and each is marked in +the source with a `# SPEC:` comment referencing Canon doc **BPE-7216-005**. When a +hardware step fails, find the matching item here to know what to change. + +**R1 — CGI path prefix.** `DEFAULT_CGI_PREFIX = "-wvhttp-01-"`. All +requests go to `http://host:port//`. *Confirm:* the real prefix for the +CR-N300. *Symptom if wrong:* every request 404s. *Fix:* `DEFAULT_CGI_PREFIX` (or the +`cgi_prefix` config key). + +**R2 — CGI endpoint names.** `_CGI = {open, claim, yield, close, info, +control, preset}.cgi`. *Confirm:* names exist and serve these roles. *Symptom:* the +corresponding action 404s/errors. *Fix:* the `_CGI` table. + +**R6b — Session id field.** `open.cgi` response is parsed for `s=` / +`session=` / `id=`, else a `Set-Cookie`. *Symptom:* "open.cgi did not return a session +id". *Fix:* `connect()` parsing. + +**R7 — `info.cgi` response format.** Assumed newline-delimited +`key=value`. Used for camera info, parameters **and** positions. *Symptom:* parse +errors, missing keys, or `transport_info()` empty. *Fix:* `_parse_kv` and the field +names below. + +**R4 — Parameter names.** `_PARAM`/info keys: `gain`, `iris`, +`exposure_mode`, `white_balance`, `shutter_speed`. *Symptom:* `CanonParameterError: +parameter not present`. *Fix:* the parameter `cmd=` strings in +[`canon_ptz.py`](./canon_ptz.py) and/or the camera-side names. + +**R5 — Axis encoding & param names.** `_PARAM` `pan/tilt/zoom/focus` +(+`panspeed`/`tiltspeed`) and `_AXIS_ENCODING` scales (pan/tilt ×100, zoom ×100, focus +×1000, offset 0). *Confirm:* native units and **direction sign**. *Symptom:* axis moves +to the wrong magnitude, wrong direction, or is rejected as out-of-range. *Fix:* +`_AXIS_ENCODING` (and `_PARAM`). + +**R6 — Focus-mode tokens.** `_FOCUS_MODE_TO_WIRE = {manual:"manual", +continuous_af:"auto", face_af:"face", tracking_af:"tracking"}`. *Symptom:* mode doesn't +change or read-back doesn't map. *Fix:* `_FOCUS_MODE_TO_WIRE`. + +**R8 — "Moving" flag.** `is_moving()` reads an assumed `moving` +field; absent ⇒ reported as not moving (move completion then relies purely on the +tolerance check). *Confirm:* whether such a flag exists. *Fix:* `is_moving()` / consider +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()`. + +**R10 — Stream URL/format.** Default `…//image.cgi`, parsed as +multipart MJPEG split on JPEG `FFD8…FFD9`. *Confirm:* the real MJPEG/JPEG stream URL and +that frames are standalone JPEGs. *Symptom:* worker never produces a frame / decode +fails. *Fix:* the `stream_url` config key (or `_default_stream_url`). + +--- + +## 6. Commissioning test regime + +A gated schedule: **do not advance to the next stage until the current one passes.** Each +step says what to run, what to observe, and — if it fails — which [risk item](#5-deviation-risk-map) +to suspect. Stages 0 can be done **now**; Stage 1+ need the camera on the network. + +> **Safety first (Stage 4 onward):** before any motion, ensure the camera can pan/tilt +> through its full range without hitting cabling, mounts, or people. Start every axis +> with the **smallest** move and be ready to power-cut. Keep `move_timeout` modest. + +### 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 47 passed.* If not, stop — the environment is wrong, not the +camera. + +0.2 **Load in BEC in simulation.** Add the `ptz_cam_sim` entry, start a BEC session, and +confirm `dev.ptz_cam_sim` appears, `dev.ptz_cam_sim.transport_info()` shows +`simulation: True, connected: True`, and `dev.ptz_cam_sim.pan.move(20)` returns a +completed status. *This proves the BEC plumbing (entry-point resolution, config mapping, +RPC, staging) independently of hardware.* → if this fails it's a BEC/packaging issue, not +the protocol. + +0.3 **Desk-check the SPEC tables.** With BPE-7216-005 open, walk the +[deviation-risk map](#5-deviation-risk-map) item by item and pre-correct any of the +`# SPEC:` tables in `transport/xc.py` you can already confirm from the document. *Every +correction here saves a hardware round-trip.* + +**Gate:** 47 green + sim device usable in BEC + SPEC tables reviewed. → proceed. + +### Stage 1 — Network & session (camera on network; read-only) + +1.1 **Reachability.** From the device-server host: `ping `, then +`curl -v http://:/`. *Confirm* TCP/HTTP responds. → if not: network/port/firewall. + +1.2 **Auth + open session.** Set `CANON_PTZ_CAM_USER` / `CANON_PTZ_CAM_PASSWORD`, then in +Python (outside BEC is fine): +```python +from csaxs_bec.devices.canon.transport.xc import XCTransport +t = XCTransport(host="", username="…", password="…") +t.connect(); print(t.connected) +``` +*Confirm* `connected` is `True`. → 401 ⇒ credentials/auth scheme; "no session id" ⇒ +[R6b](#r6b); 404 ⇒ [R1](#r1)/[R2](#r2). + +1.3 **Identity / info parse.** `print(t.get_camera_info()); print(t.list_parameters())`. +*Confirm* model/firmware look right and parameters are populated. → empty/garbled ⇒ +[R7](#r7). + +**Gate:** session opens, auth works, info parses. → proceed. + +### Stage 2 — Read-only positions & parameters + +2.1 **Positions.** `for a in Axis: print(a, t.get_position(a))`. *Confirm* values are +plausible **and in the right units** (move the camera by hand / OSD if possible and see +the readback follow). → wrong magnitude ⇒ [R5](#r5) scale; missing key ⇒ [R5](#r5)/[R7](#r7). + +2.2 **Parameters.** Read `gain`, `iris`, etc. *Confirm* names resolve. → `parameter not +present` ⇒ [R4](#r4). + +**Gate:** readbacks correct and correctly scaled. → proceed (this is the prerequisite for +trusting motion completion). + +### Stage 3 — Control privilege + +3.1 **Claim/yield.** `t.claim_control(); print(t.has_control); t.yield_control()`. +*Confirm* no error. → 403 ⇒ another client holds control, or [R2](#r2) claim endpoint. + +3.2 **Contention (optional).** Hold control from a second client; confirm our claim +behaves sensibly (errors or waits, not silent). + +**Gate:** control can be claimed and released. → proceed to motion. + +### Stage 4 — Motion (incremental, safety-critical) + +Do this through the transport first (tightest control), then through the ophyd device. + +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. + +4.2 **Absolute pan, then tilt, then zoom, then focus** (focus: set `focus_mode=manual` +first). For each: command a modest absolute target; confirm readback converges and the +move `Status` completes. → never completes ⇒ [R5](#r5) (scale so tolerance never met) or +[R8](#r8) (no moving flag + too-tight tolerance — loosen `tolerance` in config). + +4.3 **Soft + hardware limits.** Confirm an out-of-range target raises `LimitError` +(software, no command sent) and an in-range one near the hardware edge behaves. + +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.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. + +**Gate:** all four axes move accurately, complete, respect limits/speed, and stop. → +proceed. + +### Stage 5 — Focus modes + +5.1 For each of `manual`, `continuous_af`, `face_af`, `tracking_af`: +`dev.ptz_cam.focus_mode.set()` and confirm the camera engages that mode (OSD / +visible behaviour). → no change / wrong mode ⇒ [R6](#r6). + +5.2 Confirm `focus.move(...)` is **blocked** in any AF mode and **allowed** in `manual`. + +**Gate:** modes switch correctly and the manual-focus gate holds. → proceed. + +### Stage 6 — Presets + +6.1 Store a couple of presets on the camera (via its own UI). `dev.ptz_cam.recall_preset(1)` +and confirm it drives to the stored pose; `preset_state.get()` returns `1`. → no +move/error ⇒ [R3](#r3). + +**Gate:** presets recall. → proceed. + +### Stage 7 — Streaming & grab (Phase 2) + +7.1 **Establish.** `dev.ptz_cam.start_stream()`, then `dev.ptz_cam.stream_stats()`. +*Confirm* `worker.establishment_time_s` is populated (≈ the expected ~2 s, **once**) and +`buffer.size` grows. → never any frame ⇒ [R10](#r10) (stream URL/format). + +7.2 **Grab latency.** Time `dev.ptz_cam.get_latest()` — expect sub-millisecond, **no** +re-establishment cost. Grab repeatedly during simulated DAQ activity. + +7.3 **Decode.** `img = dev.ptz_cam.get_latest(decode=True)`; confirm it's a real RGB array +of the expected resolution. → decode returns `None` ⇒ frames aren't standalone JPEGs +([R10](#r10)). + +7.4 **Resilience.** Pull the network briefly; confirm `stream_stats()` shows `reconnects` +incrementing, `healthy` recovering, and **DAQ never blocks** (last good frames still +served). + +**Gate:** stream establishes once, grabs are cheap, decode works, reconnect is fail-soft. + +### Stage 8 — Full BEC integration + +8.1 Run a real scan with the camera staged; confirm staging/unstaging is clean and the +stream persists across scans (no per-scan 2 s cost). + +8.2 Confirm the live `preview` reaches the GUI (when Phase-2 preview delivery is wired). + +8.3 Confirm `on_destroy` releases control and closes the session (next client can claim). + +### Stage 9 — Soak + +9.1 Leave the device connected and streaming for hours; confirm no socket/thread leak, +session stays alive (or reconnects), and memory is bounded (ring buffer caps it). + +--- + +## 7. Recording template + +Copy per stage when commissioning: + +``` +Stage __ : ____________________ date: ______ operator: ______ +firmware: ______ device cfg: ______ + +step result(P/F) observed value / note risk item if F +---- ----------- ------------------------------- -------------- +__.1 [ ] ______________________________ R__ +__.2 [ ] ______________________________ R__ + +SPEC tables changed (file:line → old → new): + ____________________________________________________________ + +Outcome: [ ] PASS — proceed [ ] BLOCKED — see notes +``` + +When you adjust a `# SPEC:` table to match the real camera, note the change here **and** +in the commit message, so the calibrated values become the documented truth for the next +deployment. diff --git a/csaxs_bec/devices/canon/README.md b/csaxs_bec/devices/canon/README.md new file mode 100644 index 0000000..b548ff1 --- /dev/null +++ b/csaxs_bec/devices/canon/README.md @@ -0,0 +1,168 @@ +# Canon CR-N300 PTZ camera — BEC integration + +A custom (non-EPICS) ophyd device for the **Canon CR-N300** PTZ remote camera, +integrated into BEC. It provides standalone PTZ/zoom/focus motion and camera-parameter +read/write (Phase 1), plus the seams for persistent image streaming into a ring buffer +that DAQ can grab from on demand (Phase 2). + +## Status + +| Phase | Scope | State | +|-------|-------|-------| +| **1** | Standalone control: parameters + PTZ/zoom/focus motion, no imaging stack needed | **implemented** | +| **2** | Persistent acquisition worker + ring buffer + grab API + preview/file-writer delivery | **seams in place; network pull + decode to be finished against hardware** | + +## Architecture + +``` +devices/canon/ +├── transport/ # control plane — pure stdlib, no ophyd, no imaging +│ ├── base.py # CameraTransport ABC + Axis/FocusMode/limits +│ ├── xc.py # XCTransport — Canon XC Control Protocol (HTTP/CGI) +│ ├── fake.py # FakeTransport — offline/sim + tests +│ └── errors.py # typed exceptions +├── acquisition/ # imaging plane — pure stdlib, no ophyd, no transport import +│ ├── ring_buffer.py # FrameRingBuffer — bounded, stores *compressed* frames +│ └── stream_worker.py # StreamWorker + FrameSource (persistent, fail-soft) +├── positioner.py # CanonAxis — transport-backed ophyd positioner +└── canon_ptz.py # CanonPTZCamera — the BEC ophyd device +``` + +Key boundaries: + +* **Control never depends on imaging.** `transport/` and `acquisition/` import neither + ophyd nor each other. `canon_ptz.py` imports the acquisition subsystem *lazily and + guarded*, so PTZ/focus/parameter control works with the imaging stack absent. +* **The transport is injectable/mockable.** The ophyd device talks to the camera only + through the `CameraTransport` interface. `XCTransport` uses an injectable HTTP + `opener`; `FakeTransport` is a full in-memory implementation. No hardware is needed + for tests or for `simulation: true`. + +## Control protocol + +The transport implements the **Canon XC Control Protocol** (session-based HTTP/CGI: +`open` → `claim` → command CGIs → `yield`/`close`), chosen for full parameter +read-out/set coverage. The HTTP/session/auth/retry machinery is complete and unit +tested; the **exact CGI paths, parameter names and the native pan/tilt/zoom/focus +encodings are defined by Canon doc BPE-7216-005** and are isolated in the clearly +marked `_CGI` / `_PARAM` / `_AXIS_ENCODING` tables (and `DEFAULT_CGI_PREFIX`) at the +top of `transport/xc.py`. Each carries a `# SPEC:` marker — bringing the device up +against a real camera means confirming those tables, not changing logic. A VISCA-over-IP +transport could be added later behind the same `CameraTransport` interface. + +## Connection & credentials + +Credentials are **never** stored in the YAML config / Redis. The device reads them from +environment variables at connect: + +``` +CANON__USER # e.g. CANON_PTZ_CAM_USER +CANON__PASSWORD # e.g. CANON_PTZ_CAM_PASSWORD +``` + +`` is the device name upper-cased (non-alphanumerics → `_`). Override the names +with `username_env` / `password_env` in `deviceConfig`. + +The session lifecycle maps onto ophyd/BEC: `wait_for_connection()` opens the session and +claims control; `destroy()` yields control and closes the session. + +## Configuration + +See [`device_configs/canon_crn300.yaml`](../../device_configs/canon_crn300.yaml) for +full real-camera and simulation examples. Minimal real-camera entry: + +```yaml +ptz_cam: + deviceClass: csaxs_bec.devices.canon.canon_ptz.CanonPTZCamera + deviceConfig: + host: 192.168.0.100 + port: 80 + enabled: true + readoutPriority: on_request +``` + +Minimal offline/simulation entry (no hardware): + +```yaml +ptz_cam_sim: + deviceClass: csaxs_bec.devices.canon.canon_ptz.CanonPTZCamera + deviceConfig: + simulation: true + enabled: true + readoutPriority: on_request +``` + +`deviceConfig` keys: `host`, `port`, `timeout`, `cgi_prefix`, `simulation`, +`gradual_motion`, `username_env`, `password_env`, `pan_speed`, `tilt_speed`, +`buffer_maxlen`, `stream_url`, `stream_start` (`off` | `connect` | `stage`). + +## Supported commands + +Motion — each axis is a positioner (`move()`/`set()` → ophyd status, `stop()`): + +| Axis | Unit | Range (hardware) | +|------|------|------------------| +| `pan` | deg | ±170 (0.2–300 deg/s) | +| `tilt` | deg | −30 … +100 (0.2–170 deg/s) | +| `zoom` | × (optical) | 1.0 … 20.0 | +| `focus` | normalised | 0.0 (near) … 1.0 (far), manual-focus only | + +```python +cam.pan.move(45.0) # blocking; returns a completed status +st = cam.tilt.set(-10.0) # non-blocking; st.wait() to block +cam.zoom.move(5.0) +cam.jog_pan(30.0) # continuous pan at 30 deg/s; jog_pan(0) stops +cam.stop_all() # halt every axis +cam.recall_preset(3) # recall a stored PTZ preset +``` + +Parameters (read via `read()`/`read_configuration()`, set with read-back verification): +`exposure_mode`, `iris`, `gain`, `white_balance`, `shutter_speed`, `focus_mode` +(`manual` | `continuous_af` | `face_af` | `tracking_af`), and read-only `preset_state`. + +```python +cam.gain.set(6.0).wait() +cam.focus_mode.set("manual").wait() # focus axis only moves in manual mode +``` + +## Buffer-and-grab API (Phase 2) + +A persistent worker opens the stream **once** (paying the ~2 s establishment cost up +front, never inside a scan) and fills a bounded ring buffer of *compressed* JPEG frames, +each tagged with a monotonic + wall-clock timestamp and source id. DAQ grabs with O(1) +latency; if it never grabs, frames simply cycle out. + +```python +cam.start_stream() # open the stream + start the worker (idempotent) +frame = cam.get_latest() # most recent Frame (compressed bytes + timestamps) +frames = cam.get_latest_n(5) # last 5, oldest-first +frame = cam.get_nearest(ts) # closest to a wall-clock timestamp +img = cam.get_latest(decode=True) # decode only the grabbed frame (uses cv2) +cam.stream_stats() # ring-buffer + worker instrumentation (latency, health) +cam.stop_stream() +``` + +Set `stream_start: connect` (or `stage`) in `deviceConfig` to start the stream +automatically. Buffering the compressed bytes (not decoded arrays) keeps memory small +and spends no CPU decoding frames nobody grabs; decoding happens only on grab. Frame +delivery into a scan follows the area-detector convention (file-writer references via +`SUB_FILE_EVENT` for full frames, `PreviewSignal` for live preview) — those hooks exist +and are completed in Phase 2. + +> Multiple cameras = multiple devices, each with its own worker + buffer. An inter-frame +> codec (H.264/H.265 over RTSP) would force continuous decoding in the worker thread and +> is deliberately out of scope; the JPEG/MJPEG path keeps frames independently decodable. + +## Tests + +```bash +cd csaxs_bec +OPHYD_CONTROL_LAYER=dummy pytest tests/tests_devices/test_canon_*.py --random-order +``` + +No hardware required. Coverage: transport (`FakeTransport` behaviour + `XCTransport` +HTTP/session logic via a mocked opener), the ring buffer (incl. thread-safety), the +stream worker (fail-soft reconnect, no thread leak), and the ophyd device (motion +status, limits, parameter read-back verification, focus-mode gating, simulation +streaming, credential handling). +``` diff --git a/csaxs_bec/devices/canon/__init__.py b/csaxs_bec/devices/canon/__init__.py new file mode 100644 index 0000000..09736bd --- /dev/null +++ b/csaxs_bec/devices/canon/__init__.py @@ -0,0 +1,23 @@ +"""Canon CR-N300 PTZ remote camera integration for BEC. + +Public entry point is :class:`CanonPTZCamera`. The control transport +(:mod:`.transport`) and the Phase-2 acquisition subsystem (:mod:`.acquisition`) are +independently importable and free of ophyd, so standalone control works with the +imaging stack absent. +""" + +from __future__ import annotations + +from .canon_ptz import CanonPTZCamera +from .positioner import CanonAxis +from .transport import Axis, CameraTransport, FakeTransport, FocusMode, XCTransport + +__all__ = [ + "CanonPTZCamera", + "CanonAxis", + "Axis", + "FocusMode", + "CameraTransport", + "FakeTransport", + "XCTransport", +] diff --git a/csaxs_bec/devices/canon/acquisition/__init__.py b/csaxs_bec/devices/canon/acquisition/__init__.py new file mode 100644 index 0000000..c9e71b9 --- /dev/null +++ b/csaxs_bec/devices/canon/acquisition/__init__.py @@ -0,0 +1,20 @@ +"""Persistent frame acquisition + ring buffering for the Canon CR-N300 (Phase 2). + +Pure standard library. Imports neither ophyd nor the control transport, so the +control plane never depends on imaging. The ophyd device imports this lazily/guarded +so standalone control keeps working when the imaging stack is absent or disabled. +""" + +from __future__ import annotations + +from .ring_buffer import Frame, FrameRingBuffer +from .stream_worker import FrameSource, IterableFrameSource, MJPEGHttpSource, StreamWorker + +__all__ = [ + "Frame", + "FrameRingBuffer", + "FrameSource", + "IterableFrameSource", + "MJPEGHttpSource", + "StreamWorker", +] diff --git a/csaxs_bec/devices/canon/acquisition/ring_buffer.py b/csaxs_bec/devices/canon/acquisition/ring_buffer.py new file mode 100644 index 0000000..c9a6e4e --- /dev/null +++ b/csaxs_bec/devices/canon/acquisition/ring_buffer.py @@ -0,0 +1,171 @@ +"""Bounded, thread-safe ring buffer of recent (compressed) camera frames. + +This is the heart of the buffer-and-grab model (Phase 2). A persistent acquisition +worker pushes *compressed* JPEG frames in; DAQ pulls recent frames out with O(1) +latency and no setup cost. Storing the compressed bytes (rather than decoded arrays) +keeps memory small -- a 4K JPEG is tens of kilobytes vs ~24 MB decoded -- and means +no CPU is spent decoding frames nobody grabs. Decoding happens only when a frame is +actually retrieved (see the device's grab API). + +Pure standard library: no ophyd, no networking, no transport import. The buffer does +not care whether the bytes are real or synthetic, which makes it trivially testable. +""" + +from __future__ import annotations + +import threading +import time +from collections import deque +from dataclasses import dataclass +from typing import Iterable, Literal + + +@dataclass(frozen=True) +class Frame: + """A single buffered frame. + + Attributes + ---------- + data: + The raw, still-compressed frame payload (e.g. JPEG bytes). Never decoded by + the buffer. + seq: + Monotonic per-source sequence number assigned on insertion (0, 1, 2, ...). + monotonic_ts: + ``time.monotonic()`` at insertion -- for measuring intervals/latency. + wall_ts: + ``time.time()`` (Unix epoch) at insertion -- for correlating with experiment + time. :meth:`FrameRingBuffer.get_nearest` matches against this by default. + source_id: + Identifier of the stream/camera this frame came from. + """ + + data: bytes + seq: int + monotonic_ts: float + wall_ts: float + source_id: str + + @property + def nbytes(self) -> int: + return len(self.data) + + +class FrameRingBuffer: + """A fixed-capacity ring of :class:`Frame` objects for one source. + + Oldest frames are overwritten once ``maxlen`` is reached. All public methods are + thread-safe. If DAQ never grabs, frames simply cycle out -- this is not an error. + + Parameters + ---------- + maxlen: + Maximum number of frames retained. Must be >= 1. + source_id: + Default source id stamped onto frames added via :meth:`put` without an + explicit source. + time_func / wall_time_func: + Injectable clocks (monotonic / wall) for deterministic tests. + """ + + def __init__( + self, + maxlen: int, + *, + source_id: str = "default", + time_func=time.monotonic, + wall_time_func=time.time, + ) -> None: + if maxlen < 1: + raise ValueError("maxlen must be >= 1") + self._maxlen = maxlen + self._source_id = source_id + self._mono = time_func + self._wall = wall_time_func + self._lock = threading.RLock() + self._frames: deque[Frame] = deque(maxlen=maxlen) + self._seq = 0 + self._dropped = 0 # frames evicted unread (informational only) + self._total = 0 + + @property + def maxlen(self) -> int: + return self._maxlen + + @property + def source_id(self) -> str: + return self._source_id + + def __len__(self) -> int: + with self._lock: + return len(self._frames) + + def put(self, data: bytes, *, source_id: str | None = None) -> Frame: + """Insert a compressed frame, stamping it with sequence + timestamps.""" + with self._lock: + if len(self._frames) == self._maxlen: + self._dropped += 1 + frame = Frame( + data=data, + seq=self._seq, + monotonic_ts=self._mono(), + wall_ts=self._wall(), + source_id=source_id if source_id is not None else self._source_id, + ) + self._frames.append(frame) + self._seq += 1 + self._total += 1 + return frame + + def get_latest(self) -> Frame | None: + """Return the most recently inserted frame, or ``None`` if empty.""" + with self._lock: + return self._frames[-1] if self._frames else None + + def get_latest_n(self, n: int) -> list[Frame]: + """Return up to *n* most recent frames, oldest-first.""" + if n <= 0: + return [] + with self._lock: + if n >= len(self._frames): + return list(self._frames) + return list(self._frames)[-n:] + + def get_nearest( + self, timestamp: float, *, clock: Literal["wall", "monotonic"] = "wall" + ) -> Frame | None: + """Return the frame whose timestamp is closest to *timestamp*. + + Matches against the wall-clock stamp by default (for correlating with + experiment time); pass ``clock="monotonic"`` to match monotonic stamps. + Returns ``None`` if the buffer is empty. + """ + attr = "wall_ts" if clock == "wall" else "monotonic_ts" + with self._lock: + if not self._frames: + return None + return min(self._frames, key=lambda f: abs(getattr(f, attr) - timestamp)) + + def snapshot(self) -> list[Frame]: + """Return a copy of all currently buffered frames, oldest-first.""" + with self._lock: + return list(self._frames) + + def clear(self) -> None: + with self._lock: + self._frames.clear() + + def extend(self, items: Iterable[bytes], *, source_id: str | None = None) -> None: + for data in items: + self.put(data, source_id=source_id) + + def stats(self) -> dict[str, int]: + """Return buffer statistics (current size, capacity, totals, evictions).""" + with self._lock: + return { + "size": len(self._frames), + "maxlen": self._maxlen, + "total_received": self._total, + "dropped_unread": self._dropped, + "bytes_buffered": sum(f.nbytes for f in self._frames), + } diff --git a/csaxs_bec/devices/canon/acquisition/stream_worker.py b/csaxs_bec/devices/canon/acquisition/stream_worker.py new file mode 100644 index 0000000..d6728de --- /dev/null +++ b/csaxs_bec/devices/canon/acquisition/stream_worker.py @@ -0,0 +1,314 @@ +"""Persistent frame-acquisition worker feeding a :class:`FrameRingBuffer`. + +Design goals (from the integration brief): + +* A stream is opened **once** -- ideally at device connect/stage -- and kept running + for the whole session. The expensive (~2 s) stream *establishment* is paid up front, + never inside a scan. +* The worker pushes *compressed* frames into a ring buffer; DAQ pulls with O(1) + latency. If DAQ never grabs, frames just cycle out. +* Reconnects on stream drop are **fail-soft**: they never block consumers, the last + good frames keep being served, and the event is logged. +* Establishment time and per-frame intervals are measured so we can confirm the + ~2 s cost is paid once and grabs are cheap. + +This module is pure standard library and imports neither ophyd nor the control +transport: acquisition is never a dependency of control. + +GIL note: the HTTP/socket reads below release the GIL while blocked on I/O, so the +worker thread does not starve the main interpreter. No decoding happens here -- we +only split and buffer compressed bytes -- so there is no Python-level hot loop. If a +future inter-frame codec (H.264/H.265 via RTSP) is ever required it would force +continuous decoding in this thread; that is explicitly out of scope here. +""" + +from __future__ import annotations + +import abc +import threading +import time +from typing import Callable, Iterable, Iterator + +try: # pragma: no cover - logging is best-effort + from bec_lib.logger import bec_logger + + logger = bec_logger.logger +except Exception: # pragma: no cover + import logging + + logger = logging.getLogger(__name__) + +from .ring_buffer import FrameRingBuffer + +# JPEG start-of-image / end-of-image markers, used to split an MJPEG byte stream. +_JPEG_SOI = b"\xff\xd8" +_JPEG_EOI = b"\xff\xd9" + + +class FrameSource(abc.ABC): + """A source of compressed frames. ``open()`` then iterate ``frames()``.""" + + @abc.abstractmethod + def open(self, *, timeout: float | None = None) -> None: + """Establish the underlying stream (the expensive, once-per-session step).""" + + @abc.abstractmethod + def frames(self) -> Iterator[bytes]: + """Yield compressed frame payloads until the stream ends or is closed.""" + + @abc.abstractmethod + def close(self) -> None: + """Tear down the stream. Idempotent; must not raise.""" + + +class IterableFrameSource(FrameSource): + """Wraps an iterable/callable of ``bytes`` -- used for the fake feed and tests. + + Parameters + ---------- + frames_iterable: + Either an iterable of ``bytes`` or a zero-arg callable returning one + ``bytes`` frame per call. A callable yields frames indefinitely. + delay: + Seconds to sleep between yielded frames (simulates frame rate). + open_delay: + Seconds to sleep inside :meth:`open` (simulates the ~2 s establishment). + """ + + def __init__( + self, + frames_iterable: Iterable[bytes] | Callable[[], bytes], + *, + delay: float = 0.0, + open_delay: float = 0.0, + ) -> None: + self._src = frames_iterable + self._delay = delay + self._open_delay = open_delay + self._closed = False + + def open(self, *, timeout: float | None = None) -> None: + self._closed = False + if self._open_delay: + time.sleep(self._open_delay) + + def frames(self) -> Iterator[bytes]: + if callable(self._src): + while not self._closed: + yield self._src() + if self._delay: + time.sleep(self._delay) + else: + for frame in self._src: + if self._closed: + break + yield frame + if self._delay: + time.sleep(self._delay) + + def close(self) -> None: + self._closed = True + + +class MJPEGHttpSource(FrameSource): + """Read an MJPEG (multipart JPEG) HTTP stream and yield individual JPEG frames. + + Standard-library only (``urllib``). Splits the byte stream on JPEG SOI/EOI + markers -- the same approach the repo's ``WebcamViewer`` uses -- but, crucially, + yields the *compressed* JPEG bytes without decoding them. Decoding is deferred to + grab time. + """ + + def __init__(self, url: str, *, chunk_size: int = 4096, auth: tuple[str, str] | None = None): + self.url = url + self.chunk_size = chunk_size + self._auth = auth + self._response = None + self._buffer = b"" + + def open(self, *, timeout: float | None = None) -> None: + import urllib.request + + handlers = [] + if self._auth is not None: + mgr = urllib.request.HTTPPasswordMgrWithDefaultRealm() + mgr.add_password(None, self.url, self._auth[0], self._auth[1]) + handlers.append(urllib.request.HTTPDigestAuthHandler(mgr)) + handlers.append(urllib.request.HTTPBasicAuthHandler(mgr)) + opener = urllib.request.build_opener(*handlers) + self._response = opener.open(self.url, timeout=timeout) + self._buffer = b"" + + def frames(self) -> Iterator[bytes]: + if self._response is None: + raise RuntimeError("MJPEGHttpSource.open() must be called before frames()") + while True: + chunk = self._response.read(self.chunk_size) + if not chunk: + break + self._buffer += chunk + while True: + start = self._buffer.find(_JPEG_SOI) + end = self._buffer.find(_JPEG_EOI, start + 2) if start != -1 else -1 + if start == -1 or end == -1: + break + jpg = self._buffer[start : end + 2] + self._buffer = self._buffer[end + 2 :] + yield jpg + + def close(self) -> None: + if self._response is not None: + try: + self._response.close() + except Exception: # pylint: disable=broad-except + pass + self._response = None + self._buffer = b"" + + +class StreamWorker: + """Owns a background thread that keeps a :class:`FrameSource` feeding a buffer. + + Parameters + ---------- + source: + The frame source to keep alive. + buffer: + The ring buffer to push frames into. + name: + Human-readable worker name (used for the thread name and logs). + open_timeout: + Timeout passed to :meth:`FrameSource.open`. + reconnect_delay / max_reconnect_delay: + Initial and capped backoff between reconnect attempts after a drop. + time_func: + Injectable monotonic clock for deterministic latency measurement in tests. + """ + + def __init__( + self, + source: FrameSource, + buffer: FrameRingBuffer, + *, + name: str = "canon-stream", + open_timeout: float = 10.0, + reconnect_delay: float = 2.0, + max_reconnect_delay: float = 30.0, + time_func: Callable[[], float] = time.monotonic, + ) -> None: + self.source = source + self.buffer = buffer + self.name = name + self.open_timeout = open_timeout + self.reconnect_delay = reconnect_delay + self.max_reconnect_delay = max_reconnect_delay + self._time = time_func + + self._thread: threading.Thread | None = None + self._stop = threading.Event() + self._first_frame = threading.Event() + + # Instrumentation. + self._start_t: float | None = None + self._establishment_time: float | None = None + self._frames_received = 0 + self._last_frame_t: float | None = None + self._reconnects = 0 + self._healthy = False + + # -- lifecycle --------------------------------------------------------- + + def start(self) -> None: + """Start the worker thread. Idempotent.""" + if self._thread is not None and self._thread.is_alive(): + return + self._stop.clear() + self._first_frame.clear() + self._establishment_time = None + self._start_t = self._time() + self._thread = threading.Thread(target=self._run, name=self.name, daemon=True) + self._thread.start() + + def stop(self, *, timeout: float = 5.0) -> None: + """Signal the worker to stop, close the source, and join the thread.""" + self._stop.set() + try: + self.source.close() + except Exception: # pylint: disable=broad-except + pass + if self._thread is not None: + self._thread.join(timeout=timeout) + if self._thread.is_alive(): + logger.warning(f"Stream worker {self.name!r} did not stop within {timeout}s") + self._thread = None + self._healthy = False + + def wait_for_first_frame(self, timeout: float | None = None) -> bool: + """Block until the first frame arrives (or timeout). Returns success.""" + return self._first_frame.wait(timeout=timeout) + + # -- worker loop ------------------------------------------------------- + + def _run(self) -> None: + delay = self.reconnect_delay + while not self._stop.is_set(): + try: + self.source.open(timeout=self.open_timeout) + self._healthy = True + delay = self.reconnect_delay # reset backoff after a good open + for frame in self.source.frames(): + if self._stop.is_set(): + break + now = self._time() + if self._establishment_time is None and self._start_t is not None: + self._establishment_time = now - self._start_t + self.buffer.put(frame) + self._frames_received += 1 + self._last_frame_t = now + self._first_frame.set() + except Exception as exc: # pylint: disable=broad-except + # Fail soft: log, keep the last good frames, back off, retry. + self._healthy = False + logger.warning(f"Stream worker {self.name!r} error: {exc}; reconnecting") + finally: + try: + self.source.close() + except Exception: # pylint: disable=broad-except + pass + if self._stop.is_set(): + break + self._reconnects += 1 + # Interruptible backoff so stop() is responsive. + self._stop.wait(timeout=delay) + delay = min(delay * 2, self.max_reconnect_delay) + + # -- instrumentation --------------------------------------------------- + + @property + def running(self) -> bool: + return self._thread is not None and self._thread.is_alive() + + @property + def healthy(self) -> bool: + """Whether the source is currently open and delivering frames.""" + return self._healthy + + @property + def establishment_time(self) -> float | None: + """Seconds from :meth:`start` to the first buffered frame (``None`` if not yet).""" + return self._establishment_time + + def stats(self) -> dict[str, object]: + """Return worker instrumentation for latency/health monitoring.""" + last_age = None + if self._last_frame_t is not None: + last_age = self._time() - self._last_frame_t + return { + "name": self.name, + "running": self.running, + "healthy": self._healthy, + "establishment_time_s": self._establishment_time, + "frames_received": self._frames_received, + "reconnects": self._reconnects, + "last_frame_age_s": last_age, + } diff --git a/csaxs_bec/devices/canon/canon_ptz.py b/csaxs_bec/devices/canon/canon_ptz.py new file mode 100644 index 0000000..03c365c --- /dev/null +++ b/csaxs_bec/devices/canon/canon_ptz.py @@ -0,0 +1,461 @@ +"""Canon CR-N300 PTZ remote camera -- BEC ophyd device. + +Phase 1 (this module): standalone control -- parameter read/write and PTZ/zoom/focus +motion -- working with the imaging/streaming stack absent. Phase 2 seams (persistent +acquisition worker + ring buffer + grab API + preview) are wired here but the heavy +network/decoder work is deferred; the acquisition subsystem is imported lazily and +guarded so control never depends on it. + +Architecture +------------ +* ``self.transport`` -- a :class:`CameraTransport` (real ``XCTransport`` or, when + ``simulation: true``, ``FakeTransport``). All control goes through it. +* ``pan/tilt/zoom/focus`` -- :class:`CanonAxis` positioners sharing the transport. +* parameter signals -- read-only and settable signals that proxy to the transport. +* ``preview`` + ring buffer + :class:`StreamWorker` -- Phase-2 buffer-and-grab path. + +Credentials are read from environment variables (never from the YAML/Redis config): +``CANON__USER`` / ``CANON__PASSWORD`` by default (```` = the device +name upper-cased), overridable via ``username_env`` / ``password_env``. +""" + +from __future__ import annotations + +import os +import re +import time +from typing import TYPE_CHECKING + +from ophyd import Component as Cpt +from ophyd import DeviceStatus, Kind, Signal +from ophyd.utils import ReadOnlyError +from ophyd_devices import PreviewSignal +from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase + +from .positioner import CanonAxis +from .transport.base import Axis, CameraTransport, FocusMode +from .transport.errors import CanonParameterError +from .transport.fake import FakeTransport +from .transport.xc import DEFAULT_CGI_PREFIX, XCTransport + +if TYPE_CHECKING: # pragma: no cover + from bec_lib.devicemanager import ScanInfo + +try: # pragma: no cover + from bec_lib.logger import bec_logger + + logger = bec_logger.logger +except Exception: # pragma: no cover + import logging + + logger = logging.getLogger(__name__) + + +# Parameter names handled specially by the device (not generic info/control params). +_FOCUS_MODE_PARAM = "focus_mode" +_PRESET_STATE_PARAM = "preset_state" + + +class CanonParameterSignal(Signal): + """Read-only signal proxying a single camera parameter through the transport.""" + + def __init__(self, *args, cmd: str, **kwargs): + self._cmd = cmd + super().__init__(*args, **kwargs) + self._metadata.update(connected=True, write_access=False) + + # pylint: disable=arguments-differ + def get(self, **kwargs): + value = self.parent._get_parameter(self._cmd) # noqa: SLF001 + self._readback = value + return value + + def put(self, value, **kwargs): + raise ReadOnlyError(f"{self.name} is read-only") + + @property + def timestamp(self): + return time.time() + + +class CanonConfigSignal(CanonParameterSignal): + """Settable camera parameter with read-back verification on write.""" + + def __init__(self, *args, cmd: str, verify: bool = True, rtol: float = 1e-3, **kwargs): + self._verify = verify + self._rtol = rtol + super().__init__(*args, cmd=cmd, **kwargs) + self._metadata.update(write_access=True) + + # pylint: disable=arguments-differ + def put(self, value, **kwargs): + self.parent._set_parameter(self._cmd, value) # noqa: SLF001 + if self._verify: + readback = self.parent._get_parameter(self._cmd) # noqa: SLF001 + if not self._matches(readback, value): + raise CanonParameterError( + f"{self.name}: read-back {readback!r} does not match set value {value!r}" + ) + self._readback = value + self._run_subs(sub_type=self.SUB_VALUE, old_value=None, value=value) + + def set(self, value, **kwargs) -> DeviceStatus: + status = DeviceStatus(self) + try: + self.put(value) + except Exception as exc: # pylint: disable=broad-except + status.set_exception(exc) + else: + status.set_finished() + return status + + def _matches(self, readback, value) -> bool: + try: + return abs(float(readback) - float(value)) <= self._rtol * max(1.0, abs(float(value))) + except (TypeError, ValueError): + return str(readback) == str(value) + + +class CanonPTZCamera(PSIDeviceBase): + """BEC ophyd device for the Canon CR-N300 PTZ camera. + + See module docstring for the architecture overview and credential handling. + """ + + USER_ACCESS = [ + "stop_all", + "recall_preset", + "jog_pan", + "jog_tilt", + "start_stream", + "stop_stream", + "get_latest", + "get_latest_n", + "get_nearest", + "stream_stats", + "transport_info", + ] + + # -- motion axes (look like motors to BEC) ----------------------------- + pan = Cpt(CanonAxis, axis=Axis.PAN, egu="deg", tolerance=0.2, kind=Kind.normal) + tilt = Cpt(CanonAxis, axis=Axis.TILT, egu="deg", tolerance=0.2, kind=Kind.normal) + zoom = Cpt(CanonAxis, axis=Axis.ZOOM, egu="x", tolerance=0.05, kind=Kind.normal) + focus = Cpt(CanonAxis, axis=Axis.FOCUS, egu="", tolerance=0.01, kind=Kind.normal) + + # -- camera parameters ------------------------------------------------- + exposure_mode = Cpt(CanonConfigSignal, cmd="exposure_mode", verify=False, kind=Kind.config) + iris = Cpt(CanonConfigSignal, cmd="iris", kind=Kind.config) + gain = Cpt(CanonConfigSignal, cmd="gain", kind=Kind.config) + white_balance = Cpt(CanonConfigSignal, cmd="white_balance", verify=False, kind=Kind.config) + shutter_speed = Cpt(CanonConfigSignal, cmd="shutter_speed", verify=False, kind=Kind.config) + focus_mode = Cpt(CanonConfigSignal, cmd=_FOCUS_MODE_PARAM, verify=False, kind=Kind.config) + preset_state = Cpt(CanonParameterSignal, cmd=_PRESET_STATE_PARAM, kind=Kind.normal) + + # -- Phase-2 live preview ---------------------------------------------- + preview = Cpt(PreviewSignal, ndim=2, num_rotation_90=0, transpose=False, kind=Kind.omitted) + + def __init__( + self, + *, + name: str, + host: str | None = None, + port: int = 80, + simulation: bool = False, + cgi_prefix: str = DEFAULT_CGI_PREFIX, + timeout: float = 5.0, + username_env: str | None = None, + password_env: str | None = None, + gradual_motion: bool = False, + buffer_maxlen: int = 64, + stream_url: str | None = None, + stream_start: str = "off", + pan_speed: float | None = None, + tilt_speed: float | None = None, + scan_info: "ScanInfo | None" = None, + device_manager=None, + **kwargs, + ) -> None: + """See class/module docstring for parameter meanings. + + Args: + host: camera IP/hostname (ignored when ``simulation`` is true). + port: HTTP port (default 80). + simulation: when true, use :class:`FakeTransport` -- no hardware needed. + cgi_prefix: XC CGI path prefix (see ``xc.py``; verify vs BPE-7216-005). + timeout: per-request transport timeout in seconds. + username_env/password_env: env-var names holding XC credentials. + gradual_motion: simulate finite-speed travel in the fake transport. + buffer_maxlen: ring-buffer depth (Phase 2). + stream_url: MJPEG stream URL (Phase 2). Defaults derived if omitted. + stream_start: one of ``off`` | ``connect`` | ``stage`` -- when to start + the persistent acquisition worker. + pan_speed/tilt_speed: optional move speeds (deg/s) for those axes. + """ + if stream_start not in ("off", "connect", "stage"): + raise ValueError(f"stream_start must be off|connect|stage, got {stream_start!r}") + if not simulation and not host: + raise ValueError("host is required unless simulation=True") + + self._simulation = simulation + self._stream_url = stream_url + self._stream_start = stream_start + self._buffer_maxlen = buffer_maxlen + + # Shared lock serialising all transport access across axes/signals. + import threading + + self._transport_lock = threading.RLock() + + # Build the transport BEFORE super().__init__ so axis components can reach it. + if simulation: + self.transport: CameraTransport = FakeTransport(gradual_motion=gradual_motion) + else: + user, password = self._read_credentials(name, username_env, password_env) + self.transport = XCTransport( + host=host, + port=port, + username=user, + password=password, + cgi_prefix=cgi_prefix, + timeout=timeout, + ) + + # Phase-2 acquisition objects (created lazily in start_stream()). + self._buffer = None + self._worker = None + + super().__init__(name=name, scan_info=scan_info, device_manager=device_manager, **kwargs) + + # Apply optional per-axis move speeds. + if pan_speed is not None: + self.pan._speed = pan_speed # noqa: SLF001 + if tilt_speed is not None: + self.tilt._speed = tilt_speed # noqa: SLF001 + # Focus is only movable in manual focus mode. + self.focus.pre_move_hook = self._focus_move_guard + + # -- credentials ------------------------------------------------------- + + @staticmethod + def _read_credentials( + name: str, username_env: str | None, password_env: str | None + ) -> tuple[str | None, str | None]: + slug = re.sub(r"[^A-Z0-9]+", "_", name.upper()) + user_var = username_env or f"CANON_{slug}_USER" + pass_var = password_env or f"CANON_{slug}_PASSWORD" + user = os.environ.get(user_var) + password = os.environ.get(pass_var) + if user is None and password is None: + logger.warning( + f"No XC credentials found in env ({user_var}/{pass_var}); " + "connecting without authentication." + ) + return user, password + + # -- parameter routing ------------------------------------------------- + + def _get_parameter(self, name: str): + with self._transport_lock: + if name == _FOCUS_MODE_PARAM: + return self.transport.get_focus_mode().value + if name == _PRESET_STATE_PARAM: + return self.transport.get_preset_state() + return self.transport.get_parameter(name) + + def _set_parameter(self, name: str, value) -> None: + with self._transport_lock: + if name == _FOCUS_MODE_PARAM: + self.transport.set_focus_mode(FocusMode(value)) + return + self.transport.set_parameter(name, value) + + def _focus_move_guard(self, _position: float) -> None: + mode = self.transport.get_focus_mode() + if mode != FocusMode.MANUAL: + raise RuntimeError( + f"{self.name}: focus is in {mode.value} mode; switch focus_mode to " + "'manual' before moving the focus axis" + ) + + # -- connection lifecycle ---------------------------------------------- + + # pylint: disable=arguments-differ + def wait_for_connection(self, all_signals: bool = False, timeout: float = 10.0) -> None: + """Open the transport session, claim control, then wait for signals.""" + with self._transport_lock: + self.transport.connect(timeout=timeout) + self.transport.claim_control(timeout=timeout) + super().wait_for_connection(all_signals=all_signals, timeout=timeout) + + def on_connected(self) -> None: + """Log identity, seed positioner readbacks, optionally start streaming.""" + try: + info = self.transport.get_camera_info() + logger.info(f"{self.name} connected to {info.model} (fw {info.firmware})") + except Exception as exc: # pylint: disable=broad-except + logger.warning(f"{self.name}: could not read camera info: {exc}") + for ax in (self.pan, self.tilt, self.zoom, self.focus): + try: + ax._update_readback(self.transport.get_position(ax._axis)) # noqa: SLF001 + except Exception as exc: # pylint: disable=broad-except + logger.warning(f"{self.name}: initial readback for {ax.name} failed: {exc}") + if self._stream_start == "connect": + self.start_stream() + + def on_stage(self) -> None: + """Ensure the persistent stream is running if configured to start at stage.""" + if self._stream_start == "stage": + self.start_stream() + + def on_unstage(self) -> None: + """Leave the stream running across scans (it is opened once per session).""" + + def on_stop(self) -> None: + """Halt all motion (axes are also stopped by the base class).""" + try: + with self._transport_lock: + if self.transport.has_control: + self.transport.stop(None) + except Exception as exc: # pylint: disable=broad-except + logger.warning(f"{self.name}: stop_all during on_stop failed: {exc}") + + def on_destroy(self) -> None: + """Tear down the stream worker and close the transport session.""" + self.stop_stream() + try: + self.transport.close() + except Exception as exc: # pylint: disable=broad-except + logger.warning(f"{self.name}: transport close failed: {exc}") + + # -- control convenience (USER_ACCESS) --------------------------------- + + def stop_all(self) -> None: + """Halt every axis immediately.""" + for ax in (self.pan, self.tilt, self.zoom, self.focus): + ax.stop() + + def recall_preset(self, preset_id: int) -> None: + """Recall a stored PTZ preset by id.""" + with self._transport_lock: + self.transport.recall_preset(int(preset_id)) + + def jog_pan(self, speed: float) -> None: + """Continuously pan at *speed* deg/s (sign = direction); 0 stops.""" + self.pan.jog(speed) + + def jog_tilt(self, speed: float) -> None: + """Continuously tilt at *speed* deg/s (sign = direction); 0 stops.""" + self.tilt.jog(speed) + + def transport_info(self) -> dict: + """Return connection/identity info for diagnostics.""" + info = self.transport.get_camera_info() if self.transport.connected else None + return { + "connected": self.transport.connected, + "has_control": self.transport.has_control, + "simulation": self._simulation, + "model": getattr(info, "model", None), + "firmware": getattr(info, "firmware", None), + } + + # -- Phase-2 streaming / grab API -------------------------------------- + + def _default_stream_url(self) -> str: + # SPEC: the MJPEG stream URL form is defined by BPE-7216-005; this is a + # reasonable default to verify against the real camera. + host = getattr(self.transport, "host", "localhost") + port = getattr(self.transport, "port", 80) + prefix = getattr(self.transport, "cgi_prefix", DEFAULT_CGI_PREFIX) + return f"http://{host}:{port}/{prefix}/image.cgi" + + def start_stream(self) -> None: + """Start the persistent acquisition worker + ring buffer (idempotent). + + Imports the acquisition subsystem lazily and guarded: if it is unavailable, + control still works and a warning is logged. + """ + if self._worker is not None and self._worker.running: + return + try: + from .acquisition import ( # pylint: disable=import-outside-toplevel + FrameRingBuffer, + IterableFrameSource, + MJPEGHttpSource, + StreamWorker, + ) + except Exception as exc: # pylint: disable=broad-except + logger.warning(f"{self.name}: acquisition subsystem unavailable: {exc}") + return + + if self._buffer is None: + self._buffer = FrameRingBuffer(self._buffer_maxlen, source_id=self.name) + + if self._simulation: + source = IterableFrameSource(self._synthetic_jpeg, delay=0.05) + else: + url = self._stream_url or self._default_stream_url() + auth = getattr(self.transport, "_auth", None) + source = MJPEGHttpSource(url, auth=auth) + + self._worker = StreamWorker(source, self._buffer, name=f"{self.name}-stream") + self._worker.start() + logger.info(f"{self.name}: stream worker started") + + def stop_stream(self) -> None: + """Stop the acquisition worker cleanly (idempotent).""" + if self._worker is not None: + self._worker.stop() + self._worker = None + logger.info(f"{self.name}: stream worker stopped") + + @staticmethod + def _synthetic_jpeg() -> bytes: + """Minimal JPEG-framed payload for simulation (not a valid image).""" + return b"\xff\xd8" + b"SIMFRAME" + b"\xff\xd9" + + def _require_buffer(self): + if self._buffer is None: + raise RuntimeError( + f"{self.name}: no frame buffer; call start_stream() (or set " + "stream_start) before grabbing frames" + ) + return self._buffer + + def get_latest(self, decode: bool = False): + """Return the most recent buffered frame (``Frame`` or decoded array).""" + frame = self._require_buffer().get_latest() + return self._decode(frame) if (decode and frame is not None) else frame + + def get_latest_n(self, n: int, decode: bool = False): + """Return up to *n* most recent frames, oldest-first.""" + frames = self._require_buffer().get_latest_n(n) + return [self._decode(f) for f in frames] if decode else frames + + def get_nearest(self, timestamp: float, decode: bool = False, clock: str = "wall"): + """Return the buffered frame closest to *timestamp*.""" + frame = self._require_buffer().get_nearest(timestamp, clock=clock) + return self._decode(frame) if (decode and frame is not None) else frame + + def stream_stats(self) -> dict: + """Return ring-buffer + worker instrumentation (latency, health, counts).""" + stats: dict = {"buffer": self._buffer.stats() if self._buffer else None} + if self._worker is not None: + stats["worker"] = self._worker.stats() + return stats + + @staticmethod + def _decode(frame): + """Decode a single buffered JPEG frame to an RGB array. + + Phase-2 helper. Uses ``cv2`` (already a declared csaxs_bec dependency); only + the frame actually grabbed is decoded. + """ + if frame is None: + return None + import cv2 # pylint: disable=import-outside-toplevel + import numpy as np # pylint: disable=import-outside-toplevel + + image = cv2.imdecode(np.frombuffer(frame.data, np.uint8), cv2.IMREAD_COLOR) + if image is None: + return None + return cv2.cvtColor(image, cv2.COLOR_BGR2RGB) diff --git a/csaxs_bec/devices/canon/positioner.py b/csaxs_bec/devices/canon/positioner.py new file mode 100644 index 0000000..0a51dde --- /dev/null +++ b/csaxs_bec/devices/canon/positioner.py @@ -0,0 +1,228 @@ +"""Transport-backed positioner component for a single Canon PTZ/focus axis. + +``CanonAxis`` is a real ophyd positioner (``Device`` + ``PositionerBase``) so that +BEC treats pan/tilt/zoom/focus as motors: each exposes ``move()``/``set()`` returning +a ``DeviceStatus`` that completes when the axis reaches target (or on timeout/stop), +plus ``stop()``. It follows the threaded-move pattern of ``ophyd_devices.SimPositioner`` +but reads/writes through the parent device's injected +:class:`~csaxs_bec.devices.canon.transport.base.CameraTransport` rather than a +simulator backend, so the same class drives both the real camera and the fake. + +The axis never opens its own connection; it borrows the parent's transport and a +shared lock so concurrent axis commands are serialised onto the single HTTP session. +""" + +from __future__ import annotations + +import threading +import time +from typing import Callable + +from ophyd import Component as Cpt +from ophyd import Device, DeviceStatus, Kind, PositionerBase, Signal +from ophyd.utils import LimitError + +from .transport.base import HARDWARE_LIMITS, Axis, CameraTransport + + +class CanonAxis(Device, PositionerBase): + """One motion axis (pan/tilt/zoom/focus) of the Canon camera. + + Parameters + ---------- + axis: + Which :class:`Axis` this positioner drives. + egu: + Engineering units for the readback (e.g. ``"deg"``). + limits: + Soft ``(low, high)`` limits. Defaults to the axis hardware limits. Soft + limits may narrow, never widen, the hardware range. + tolerance: + Absolute readback tolerance for declaring a move complete. + speed: + Optional move speed (axis-units/s) passed to the transport on absolute moves. + poll_interval: + Seconds between readback polls while moving. + move_timeout: + Seconds before a move fails with ``TimeoutError``. + """ + + readback = Cpt(Signal, value=0.0, kind=Kind.hinted) + setpoint = Cpt(Signal, value=0.0, kind=Kind.normal) + motor_is_moving = Cpt(Signal, value=0, kind=Kind.normal) + + SUB_READBACK = "readback" + _default_sub = SUB_READBACK + + def __init__( + self, + prefix: str = "", + *, + name: str, + axis: Axis, + egu: str = "", + limits: tuple[float, float] | None = None, + tolerance: float = 0.1, + speed: float | None = None, + poll_interval: float = 0.1, + move_timeout: float = 30.0, + parent=None, + kind=None, + **kwargs, + ) -> None: + self._axis = Axis(axis) + self._egu = egu + self._tolerance = tolerance + self._speed = speed + self._poll_interval = poll_interval + self._move_timeout = move_timeout + self._stopped = False + self._move_thread: threading.Thread | None = None + self._fallback_lock = threading.RLock() + # Optional gate set by the parent (e.g. focus axis only movable in MF). + self.pre_move_hook: Callable[[float], None] | None = None + + hw = HARDWARE_LIMITS[self._axis] + if limits is None: + self._limits: tuple[float, float] = (hw.low, hw.high) + else: + low, high = limits + # Clamp soft limits into the hardware range. + self._limits = (max(low, hw.low), min(high, hw.high)) + + super().__init__(prefix, name=name, parent=parent, kind=kind, **kwargs) + + # -- transport access -------------------------------------------------- + + @property + def _transport(self) -> CameraTransport: + transport = getattr(self.parent, "transport", None) + if transport is None: + raise RuntimeError(f"{self.name}: parent has no transport") + return transport + + @property + def _lock(self) -> threading.RLock: + return getattr(self.parent, "_transport_lock", None) or self._fallback_lock + + def _call(self, fn): + """Run a transport call under the shared lock.""" + with self._lock: + return fn() + + # -- positioner interface ---------------------------------------------- + + @property + def limits(self) -> tuple[float, float]: + return self._limits + + @property + def low_limit(self) -> float: + return self._limits[0] + + @property + def high_limit(self) -> float: + return self._limits[1] + + @property + def egu(self) -> str: + return self._egu + + @property + def position(self) -> float: + return self.readback.get() + + def check_value(self, value: float) -> None: + low, high = self._limits + if low != high and not low <= value <= high: + raise LimitError(f"{self.name}: position {value} outside limits {self._limits}") + + def _update_readback(self, value: float) -> None: + old = self.readback.get() + self.readback.put(value) + self._set_position(value) + self._run_subs( + sub_type=self.SUB_READBACK, old_value=old, value=value, timestamp=time.time() + ) + + def _poll_until_target(self, target: float, status: DeviceStatus) -> None: + deadline = time.monotonic() + self._move_timeout + try: + while not self._stopped: + pos = self._call(lambda: self._transport.get_position(self._axis)) + self._update_readback(pos) + if abs(pos - target) <= self._tolerance: + break + if time.monotonic() > deadline: + raise TimeoutError( + f"{self.name}: move to {target} timed out after {self._move_timeout}s" + ) + time.sleep(self._poll_interval) + # Final readback refresh. + pos = self._call(lambda: self._transport.get_position(self._axis)) + self._update_readback(pos) + except Exception as exc: # pylint: disable=broad-except + self.motor_is_moving.put(0) + if not status.done: + status.set_exception(exc) + return + self.motor_is_moving.put(0) + if self._stopped: + if not status.done: + status.set_exception(RuntimeError(f"{self.name}: move interrupted by stop()")) + return + if not status.done: + status.set_finished() + + def move(self, position: float, wait: bool = True, timeout=None, moved_cb=None) -> DeviceStatus: + """Move the axis to *position*; return a status that completes at target.""" + self.check_value(position) + if self.pre_move_hook is not None: + self.pre_move_hook(position) + self._stopped = False + self.setpoint.put(position) + + status = DeviceStatus(self, timeout=timeout or self._move_timeout) + if moved_cb is not None: + status.add_callback(moved_cb) + self.motor_is_moving.put(1) + + try: + self._call( + lambda: self._transport.move_absolute(self._axis, position, speed=self._speed) + ) + except Exception as exc: # pylint: disable=broad-except + self.motor_is_moving.put(0) + status.set_exception(exc) + return status + + self._move_thread = threading.Thread( + target=self._poll_until_target, args=(position, status), daemon=True + ) + self._move_thread.start() + + if wait: + status.wait() + return status + + def set(self, position: float, **kwargs) -> DeviceStatus: + """Alias for :meth:`move` with ``wait=False`` (BEC positioner interface).""" + return self.move(position, wait=False, **kwargs) + + def jog(self, speed: float) -> None: + """Start a continuous move at *speed* (sign = direction); 0 stops.""" + self._call(lambda: self._transport.jog(self._axis, speed)) + + def stop(self, *, success: bool = False) -> None: + """Halt this axis immediately (best-effort: never raises during teardown).""" + self._stopped = True + try: + self._call(lambda: self._transport.stop(self._axis)) + except Exception as exc: # pylint: disable=broad-except + self.log.warning("%s: transport stop failed: %s", self.name, exc) + finally: + if self._move_thread is not None: + self._move_thread.join(timeout=2.0) + self._move_thread = None + self.motor_is_moving.put(0) + super().stop(success=success) diff --git a/csaxs_bec/devices/canon/transport/__init__.py b/csaxs_bec/devices/canon/transport/__init__.py new file mode 100644 index 0000000..1b126cb --- /dev/null +++ b/csaxs_bec/devices/canon/transport/__init__.py @@ -0,0 +1,39 @@ +"""Control-plane transport for the Canon CR-N300 PTZ camera. + +Pure standard-library; imports neither ophyd nor the acquisition subsystem, so that +standalone control works with the imaging stack absent and the transport is unit +testable without hardware. +""" + +from __future__ import annotations + +from .base import HARDWARE_LIMITS, Axis, AxisLimits, CameraInfo, CameraTransport, FocusMode +from .errors import ( + CanonCommandError, + CanonConnectionError, + CanonControlPrivilegeError, + CanonParameterError, + CanonSessionError, + CanonTimeoutError, + CanonTransportError, +) +from .fake import FakeTransport +from .xc import XCTransport + +__all__ = [ + "Axis", + "AxisLimits", + "CameraInfo", + "CameraTransport", + "FocusMode", + "HARDWARE_LIMITS", + "FakeTransport", + "XCTransport", + "CanonTransportError", + "CanonConnectionError", + "CanonTimeoutError", + "CanonSessionError", + "CanonControlPrivilegeError", + "CanonCommandError", + "CanonParameterError", +] diff --git a/csaxs_bec/devices/canon/transport/base.py b/csaxs_bec/devices/canon/transport/base.py new file mode 100644 index 0000000..4a382e6 --- /dev/null +++ b/csaxs_bec/devices/canon/transport/base.py @@ -0,0 +1,208 @@ +"""Transport abstraction for the Canon CR-N300 PTZ camera. + +This module is intentionally free of any ophyd / BEC imports so that it can be unit +tested without hardware and so that *control* never depends on the *acquisition* +(imaging) subsystem. The ophyd device talks to the camera exclusively through the +:class:`CameraTransport` interface; concrete implementations are: + +* :class:`~csaxs_bec.devices.canon.transport.xc.XCTransport` -- the real + Canon XC Control Protocol client (session-based HTTP/CGI), and +* :class:`~csaxs_bec.devices.canon.transport.fake.FakeTransport` -- an in-memory + simulator used for offline mode and the test-suite. + +Units (user-facing, as exposed by the positioners): + +* ``pan`` -- degrees, hardware range +/-170 deg +* ``tilt`` -- degrees, hardware range -30 .. +100 deg +* ``zoom`` -- optical magnification, 1.0 .. 20.0 x +* ``focus`` -- normalised manual-focus position, 0.0 (near) .. 1.0 (far) + +The mapping between these user units and the camera's native command encoding lives +inside the concrete transport (see ``xc.py``), so the ophyd layer stays unit-correct +and protocol-agnostic. +""" + +from __future__ import annotations + +import abc +import enum +from dataclasses import dataclass + + +class Axis(str, enum.Enum): + """The four motion axes of the camera.""" + + PAN = "pan" + TILT = "tilt" + ZOOM = "zoom" + FOCUS = "focus" + + +class FocusMode(str, enum.Enum): + """Focus operating modes supported by the CR-N300. + + ``MANUAL`` exposes the manual-focus position as a positioner; the autofocus + modes drive focus internally and the manual-focus positioner is inactive. + """ + + MANUAL = "manual" + CONTINUOUS_AF = "continuous_af" + FACE_AF = "face_af" + TRACKING_AF = "tracking_af" + + +@dataclass(frozen=True) +class AxisLimits: + """Inclusive low/high travel limits and speed range for an axis. + + Speeds are in axis-units per second (deg/s for pan/tilt). ``None`` speed bounds + mean the axis does not support a settable move speed. + """ + + low: float + high: float + speed_min: float | None = None + speed_max: float | None = None + + +# Hardware limits taken from the CR-N300 specification. These are the *physical* +# limits; per-deployment soft limits can be narrower and are applied in the ophyd +# layer (the positioner's ``limits``). +HARDWARE_LIMITS: dict[Axis, AxisLimits] = { + Axis.PAN: AxisLimits(low=-170.0, high=170.0, speed_min=0.2, speed_max=300.0), + Axis.TILT: AxisLimits(low=-30.0, high=100.0, speed_min=0.2, speed_max=170.0), + Axis.ZOOM: AxisLimits(low=1.0, high=20.0), + Axis.FOCUS: AxisLimits(low=0.0, high=1.0), +} + + +@dataclass(frozen=True) +class CameraInfo: + """Static identity information read once after connecting.""" + + model: str = "" + firmware: str = "" + serial: str = "" + + +class CameraTransport(abc.ABC): + """Control-plane interface to a Canon PTZ camera. + + A transport owns the session lifecycle (connect/claim/yield/close), exposes + camera parameters, drives the four axes (absolute moves plus optional + continuous jog), and recalls presets. It does **not** handle the image stream: + persistent frame acquisition lives in + :mod:`csaxs_bec.devices.canon.acquisition`, which never imports this module. + + All methods raise subclasses of + :class:`~csaxs_bec.devices.canon.transport.errors.CanonTransportError` on + failure. Implementations must be safe to call from a single controlling thread; + the ophyd device serialises access. + """ + + # -- session lifecycle ------------------------------------------------- + + @abc.abstractmethod + def connect(self, *, timeout: float | None = None) -> None: + """Open a session with the camera (Canon ``open.cgi``). + + Idempotent: calling :meth:`connect` on an already-connected transport is a + no-op. Does **not** claim control privilege; call :meth:`claim_control` + for that. + """ + + @abc.abstractmethod + def close(self) -> None: + """Yield any held control privilege and close the session. + + Idempotent and must never raise; cleanup errors are swallowed/logged so + that ophyd ``unstage``/``destroy`` can always complete. + """ + + @property + @abc.abstractmethod + def connected(self) -> bool: + """Whether a session is currently open.""" + + @abc.abstractmethod + def claim_control(self, *, timeout: float | None = None) -> None: + """Acquire the exclusive control privilege (Canon ``claim.cgi``).""" + + @abc.abstractmethod + def yield_control(self) -> None: + """Release the control privilege (Canon ``yield.cgi``). Idempotent.""" + + @property + @abc.abstractmethod + def has_control(self) -> bool: + """Whether this transport currently holds the control privilege.""" + + @abc.abstractmethod + def get_camera_info(self) -> CameraInfo: + """Return static identity information (model/firmware/serial).""" + + # -- parameters -------------------------------------------------------- + + @abc.abstractmethod + def get_parameter(self, name: str) -> object: + """Read a single camera parameter (e.g. ``"gain"``, ``"white_balance"``).""" + + @abc.abstractmethod + def set_parameter(self, name: str, value: object) -> None: + """Write a single camera parameter. Requires control privilege.""" + + @abc.abstractmethod + def list_parameters(self) -> dict[str, object]: + """Return a snapshot of all known parameters as ``{name: value}``.""" + + # -- motion ------------------------------------------------------------ + + @abc.abstractmethod + def get_position(self, axis: Axis) -> float: + """Return the current position of *axis* in its user unit.""" + + @abc.abstractmethod + def move_absolute(self, axis: Axis, value: float, *, speed: float | None = None) -> None: + """Command an absolute move of *axis* to *value* (non-blocking). + + Returns as soon as the command is accepted; completion is observed by the + caller polling :meth:`get_position` (and optionally :meth:`is_moving`). + Requires control privilege. + """ + + @abc.abstractmethod + def jog(self, axis: Axis, speed: float) -> None: + """Start a continuous move of *axis* at *speed* (sign sets direction). + + ``speed == 0`` stops the axis. Requires control privilege. Implementations + that do not support continuous jog raise + :class:`~csaxs_bec.devices.canon.transport.errors.CanonCommandError`. + """ + + @abc.abstractmethod + def is_moving(self, axis: Axis) -> bool: + """Whether *axis* is currently in motion (best-effort).""" + + @abc.abstractmethod + def stop(self, axis: Axis | None = None) -> None: + """Halt *axis*, or all axes when *axis* is ``None``. Requires control.""" + + # -- focus mode -------------------------------------------------------- + + @abc.abstractmethod + def get_focus_mode(self) -> FocusMode: + """Return the current focus mode.""" + + @abc.abstractmethod + def set_focus_mode(self, mode: FocusMode) -> None: + """Set the focus mode. Requires control privilege.""" + + # -- presets ----------------------------------------------------------- + + @abc.abstractmethod + def recall_preset(self, preset_id: int) -> None: + """Recall a stored PTZ preset by id. Requires control privilege.""" + + @abc.abstractmethod + def get_preset_state(self) -> int: + """Return the id of the last recalled preset (``-1`` if none/unknown).""" diff --git a/csaxs_bec/devices/canon/transport/errors.py b/csaxs_bec/devices/canon/transport/errors.py new file mode 100644 index 0000000..cf19812 --- /dev/null +++ b/csaxs_bec/devices/canon/transport/errors.py @@ -0,0 +1,40 @@ +"""Typed exceptions for the Canon CR-N300 transport layer. + +These are deliberately defined in a tiny, dependency-free module so that both the +transport implementations and the ophyd device can import them without pulling in +ophyd or any networking stack. +""" + +from __future__ import annotations + + +class CanonTransportError(Exception): + """Base class for all Canon transport errors.""" + + +class CanonConnectionError(CanonTransportError): + """Raised when the transport cannot reach the camera (socket/HTTP failure).""" + + +class CanonTimeoutError(CanonTransportError): + """Raised when a request to the camera exceeds its timeout.""" + + +class CanonSessionError(CanonTransportError): + """Raised on session-lifecycle problems (open/claim/yield/close).""" + + +class CanonControlPrivilegeError(CanonSessionError): + """Raised when a control command is attempted without an active control claim. + + Canon's XC protocol grants exactly one client the "control privilege" at a time + via ``claim.cgi``. Camera-control CGIs fail until the privilege is held. + """ + + +class CanonCommandError(CanonTransportError): + """Raised when the camera rejects a command or returns an error response.""" + + +class CanonParameterError(CanonCommandError): + """Raised when a parameter name/value is invalid or read-back verification fails.""" diff --git a/csaxs_bec/devices/canon/transport/fake.py b/csaxs_bec/devices/canon/transport/fake.py new file mode 100644 index 0000000..867a97a --- /dev/null +++ b/csaxs_bec/devices/canon/transport/fake.py @@ -0,0 +1,248 @@ +"""In-memory fake transport for the Canon CR-N300. + +:class:`FakeTransport` implements the full :class:`CameraTransport` interface with +no network access. It backs two things: + +* **offline / simulation mode** -- set ``simulation: true`` in the device config and + the ophyd device loads without a physical camera, and +* **the unit-test suite** -- tests inject it (or assert against it) instead of + touching hardware. + +Motion is *instant* by default (a move sets the position immediately), which keeps +positioner tests fast and deterministic. Pass ``gradual_motion=True`` to simulate +finite-speed travel for exercising the polling / status path; positions then advance +toward the target according to the commanded (or default) speed and the injected +clock ``time_func``. +""" + +from __future__ import annotations + +from typing import Callable + +from .base import HARDWARE_LIMITS, Axis, CameraInfo, CameraTransport, FocusMode +from .errors import CanonControlPrivilegeError, CanonParameterError, CanonSessionError + +# Default parameter set surfaced by the fake. Mirrors the read/settable parameters +# the ophyd device exposes, with plausible defaults. +_DEFAULT_PARAMETERS: dict[str, object] = { + "exposure_mode": "auto", + "iris": 5.6, + "gain": 0.0, + "white_balance": "auto", + "shutter_speed": "1/50", +} + +_DEFAULT_POSITIONS: dict[Axis, float] = { + Axis.PAN: 0.0, + Axis.TILT: 0.0, + Axis.ZOOM: 1.0, + Axis.FOCUS: 0.5, +} + +# Default jog/move speeds (axis-units per second) used when a move does not specify +# one and ``gradual_motion`` is enabled. +_DEFAULT_SPEED: dict[Axis, float] = { + Axis.PAN: 60.0, + Axis.TILT: 60.0, + Axis.ZOOM: 4.0, + Axis.FOCUS: 1.0, +} + + +class FakeTransport(CameraTransport): + """A deterministic, hardware-free :class:`CameraTransport` implementation.""" + + def __init__( + self, + *, + gradual_motion: bool = False, + time_func: Callable[[], float] | None = None, + info: CameraInfo | None = None, + ) -> None: + import time as _time + + self._time = time_func or _time.monotonic + self._gradual = gradual_motion + self._info = info or CameraInfo(model="CR-N300 (fake)", firmware="0.0", serial="FAKE0001") + + self._connected = False + self._has_control = False + self._parameters = dict(_DEFAULT_PARAMETERS) + self._focus_mode = FocusMode.MANUAL + self._preset_state = -1 + + # Motion state. For gradual motion we record where/when a move started. + self._positions = dict(_DEFAULT_POSITIONS) + self._targets = dict(_DEFAULT_POSITIONS) + self._move_start_pos: dict[Axis, float] = {} + self._move_start_t: dict[Axis, float] = {} + self._move_speed: dict[Axis, float] = {} + self._jog_speed: dict[Axis, float] = {a: 0.0 for a in Axis} + + # Test hooks: count calls and optionally inject failures. + self.call_log: list[str] = [] + self.fail_next: dict[str, Exception] = {} + + # -- internal helpers -------------------------------------------------- + + def _record(self, name: str) -> None: + self.call_log.append(name) + exc = self.fail_next.pop(name, None) + if exc is not None: + raise exc + + def _require_session(self) -> None: + if not self._connected: + raise CanonSessionError("transport is not connected") + + def _require_control(self) -> None: + self._require_session() + if not self._has_control: + raise CanonControlPrivilegeError("control privilege not held") + + def _settle(self, axis: Axis) -> None: + """Advance gradual-motion state for *axis* to 'now'.""" + if axis not in self._move_start_t: + return + elapsed = max(0.0, self._time() - self._move_start_t[axis]) + start = self._move_start_pos[axis] + target = self._targets[axis] + speed = self._move_speed[axis] + if speed <= 0: + self._positions[axis] = target + else: + travelled = speed * elapsed + remaining = abs(target - start) + if travelled >= remaining: + self._positions[axis] = target + else: + direction = 1.0 if target >= start else -1.0 + self._positions[axis] = start + direction * travelled + if self._positions[axis] == target: + # Move finished; drop the in-flight bookkeeping. + self._move_start_t.pop(axis, None) + self._move_start_pos.pop(axis, None) + self._move_speed.pop(axis, None) + + # -- session lifecycle ------------------------------------------------- + + def connect(self, *, timeout: float | None = None) -> None: + self._record("connect") + self._connected = True + + def close(self) -> None: + self._record("close") + self._has_control = False + self._connected = False + + @property + def connected(self) -> bool: + return self._connected + + def claim_control(self, *, timeout: float | None = None) -> None: + self._record("claim_control") + self._require_session() + self._has_control = True + + def yield_control(self) -> None: + self._record("yield_control") + self._has_control = False + + @property + def has_control(self) -> bool: + return self._has_control + + def get_camera_info(self) -> CameraInfo: + self._require_session() + return self._info + + # -- parameters -------------------------------------------------------- + + def get_parameter(self, name: str) -> object: + self._record(f"get_parameter:{name}") + self._require_session() + if name not in self._parameters: + raise CanonParameterError(f"unknown parameter {name!r}") + return self._parameters[name] + + def set_parameter(self, name: str, value: object) -> None: + self._record(f"set_parameter:{name}") + self._require_control() + if name not in self._parameters: + raise CanonParameterError(f"unknown parameter {name!r}") + self._parameters[name] = value + + def list_parameters(self) -> dict[str, object]: + self._require_session() + return dict(self._parameters) + + # -- motion ------------------------------------------------------------ + + def get_position(self, axis: Axis) -> float: + self._require_session() + if self._gradual: + self._settle(axis) + return self._positions[axis] + + def move_absolute(self, axis: Axis, value: float, *, speed: float | None = None) -> None: + self._record(f"move_absolute:{axis.value}") + self._require_control() + lim = HARDWARE_LIMITS[axis] + if not lim.low <= value <= lim.high: + raise CanonParameterError( + f"{axis.value}={value} outside hardware range [{lim.low}, {lim.high}]" + ) + self._targets[axis] = value + if not self._gradual: + self._positions[axis] = value + return + self._move_start_pos[axis] = self._positions[axis] + self._move_start_t[axis] = self._time() + self._move_speed[axis] = speed if speed is not None else _DEFAULT_SPEED[axis] + + def jog(self, axis: Axis, speed: float) -> None: + self._record(f"jog:{axis.value}") + self._require_control() + self._jog_speed[axis] = speed + + def is_moving(self, axis: Axis) -> bool: + self._require_session() + if self._jog_speed[axis] != 0.0: + return True + if self._gradual: + self._settle(axis) + return self._positions[axis] != self._targets[axis] + return False + + def stop(self, axis: Axis | None = None) -> None: + self._record("stop" if axis is None else f"stop:{axis.value}") + self._require_control() + axes = list(Axis) if axis is None else [axis] + for ax in axes: + if self._gradual: + self._settle(ax) + self._targets[ax] = self._positions[ax] + self._jog_speed[ax] = 0.0 + self._move_start_t.pop(ax, None) + + # -- focus mode -------------------------------------------------------- + + def get_focus_mode(self) -> FocusMode: + self._require_session() + return self._focus_mode + + def set_focus_mode(self, mode: FocusMode) -> None: + self._record(f"set_focus_mode:{mode.value}") + self._require_control() + self._focus_mode = FocusMode(mode) + + # -- presets ----------------------------------------------------------- + + def recall_preset(self, preset_id: int) -> None: + self._record(f"recall_preset:{preset_id}") + self._require_control() + self._preset_state = int(preset_id) + + def get_preset_state(self) -> int: + self._require_session() + return self._preset_state diff --git a/csaxs_bec/devices/canon/transport/xc.py b/csaxs_bec/devices/canon/transport/xc.py new file mode 100644 index 0000000..b9414e6 --- /dev/null +++ b/csaxs_bec/devices/canon/transport/xc.py @@ -0,0 +1,415 @@ +"""Canon XC Control Protocol transport (session-based HTTP/CGI), standard library only. + +This is the real client for the CR-N300. It implements the XC session model: + +1. ``open.cgi`` -- open a session, receive a session id; +2. ``claim.cgi`` -- acquire the exclusive *control privilege*; +3. command CGIs -- camera info, camera/preset control, menu, focus metadata; +4. ``yield.cgi`` -- release the control privilege; +5. session close. + +Design notes / honesty about the spec +-------------------------------------- +The HTTP, session, authentication and retry machinery below is implemented fully +and is unit-tested through an injectable ``opener`` (no hardware needed). However, +the **exact CGI path prefix, parameter names and the native encoding of pan/tilt/ +zoom/focus values are defined by Canon document BPE-7216-005**, which is not bundled +here. Every such spec-defined detail is isolated in the clearly-marked constants and +the ``_CGI`` / ``_PARAM`` / ``_AXIS_ENCODING`` tables at the top of this module, so +that bringing this up against a real camera is a matter of adjusting those tables -- +not rewriting logic. Each carries a ``# SPEC:`` marker. + +The defaults follow Canon's long-standing WebView/Livescope CGI conventions +(``-wvhttp-01-`` prefix, newline-delimited ``key=value`` response bodies) which the +XC protocol is derived from; treat them as a starting point to verify. +""" + +from __future__ import annotations + +import socket +import urllib.error +import urllib.parse +import urllib.request +from dataclasses import dataclass + +from .base import HARDWARE_LIMITS, Axis, CameraInfo, CameraTransport, FocusMode +from .errors import ( + CanonCommandError, + CanonConnectionError, + CanonControlPrivilegeError, + CanonParameterError, + CanonSessionError, + CanonTimeoutError, + CanonTransportError, +) + +try: # pragma: no cover - logging is best-effort and never required for control + from bec_lib.logger import bec_logger + + logger = bec_logger.logger +except Exception: # pragma: no cover + import logging + + logger = logging.getLogger(__name__) + + +# SPEC: path prefix for all CGIs. Canon WebView/Livescope uses "-wvhttp-01-". +# Verify against BPE-7216-005 for the CR-N300 XC protocol. +DEFAULT_CGI_PREFIX = "-wvhttp-01-" + +# SPEC: CGI endpoint names. Adjust to match BPE-7216-005. +_CGI = { + "open": "open.cgi", + "claim": "claim.cgi", + "yield": "yield.cgi", + "close": "close.cgi", + "info": "info.cgi", + "control": "control.cgi", + "preset": "preset.cgi", +} + +# SPEC: query-parameter names used by the control/info CGIs. Adjust per spec. +_PARAM = { + "session": "s", # session-id parameter appended to authenticated requests + "pan": "pan", + "tilt": "tilt", + "zoom": "zoom", + "focus": "focus", + "pan_speed": "panspeed", + "tilt_speed": "tiltspeed", + "focus_mode": "focusmode", + "preset": "preset", + "stop": "stop", +} + +# SPEC: focus-mode wire tokens. +_FOCUS_MODE_TO_WIRE = { + FocusMode.MANUAL: "manual", + FocusMode.CONTINUOUS_AF: "auto", + FocusMode.FACE_AF: "face", + FocusMode.TRACKING_AF: "tracking", +} +_WIRE_TO_FOCUS_MODE = {v: k for k, v in _FOCUS_MODE_TO_WIRE.items()} + + +@dataclass(frozen=True) +class _AxisEncoding: + """Linear map between a user unit and the camera's native command integer. + + native = round(user * scale + offset). SPEC: scale/offset are device-native and + must be calibrated/confirmed against BPE-7216-005. + """ + + scale: float + offset: float = 0.0 + + def to_native(self, user: float) -> int: + return int(round(user * self.scale + self.offset)) + + def to_user(self, native: float) -> float: + return (native - self.offset) / self.scale + + +# SPEC: per-axis encodings. Pan/tilt commonly in 1/100 degree; zoom/focus device +# specific. Confirm against BPE-7216-005. +_AXIS_ENCODING = { + Axis.PAN: _AxisEncoding(scale=100.0), + Axis.TILT: _AxisEncoding(scale=100.0), + Axis.ZOOM: _AxisEncoding(scale=100.0), + Axis.FOCUS: _AxisEncoding(scale=1000.0), +} + + +@dataclass +class HttpResponse: + """Minimal HTTP response container returned by an :class:`XCTransport` opener.""" + + status: int + body: bytes + headers: dict[str, str] + + @property + def text(self) -> str: + return self.body.decode("ascii", errors="replace") + + +def _default_opener( + url: str, data: bytes | None, timeout: float, auth: tuple[str, str] | None +) -> HttpResponse: + """Standard-library HTTP opener (urllib) with optional HTTP Basic/Digest auth.""" + handlers: list[urllib.request.BaseHandler] = [] + if auth is not None: + pwd_mgr = urllib.request.HTTPPasswordMgrWithDefaultRealm() + pwd_mgr.add_password(None, url, auth[0], auth[1]) + handlers.append(urllib.request.HTTPDigestAuthHandler(pwd_mgr)) + handlers.append(urllib.request.HTTPBasicAuthHandler(pwd_mgr)) + opener = urllib.request.build_opener(*handlers) + req = urllib.request.Request(url, data=data, method="POST" if data is not None else "GET") + try: + with opener.open(req, timeout=timeout) as resp: + return HttpResponse( + status=resp.status, + body=resp.read(), + headers={k.lower(): v for k, v in resp.headers.items()}, + ) + except urllib.error.HTTPError as exc: # 4xx/5xx still carry a body + return HttpResponse( + status=exc.code, + body=exc.read() if hasattr(exc, "read") else b"", + headers={k.lower(): v for k, v in (exc.headers or {}).items()}, + ) + except socket.timeout as exc: + raise CanonTimeoutError(f"request to {url} timed out after {timeout}s") from exc + except urllib.error.URLError as exc: + raise CanonConnectionError(f"cannot reach {url}: {exc.reason}") from exc + + +class XCTransport(CameraTransport): + """HTTP/CGI client implementing the Canon XC Control Protocol. + + Parameters + ---------- + host, port: + Camera network location. ``port`` defaults to 80 (HTTP). + username, password: + Credentials for the control session. Supplied by the ophyd device from + environment variables -- never stored in the YAML config / Redis. + cgi_prefix: + Path prefix for the CGI endpoints (SPEC, see module docstring). + timeout: + Per-request timeout in seconds. + max_retries: + Number of transient-failure retries (with one reconnect) per request. + opener: + Injectable HTTP backend ``opener(url, data, timeout, auth) -> HttpResponse``. + Defaults to a urllib-based opener. Tests pass a fake to avoid the network. + """ + + def __init__( + self, + *, + host: str, + port: int = 80, + username: str | None = None, + password: str | None = None, + cgi_prefix: str = DEFAULT_CGI_PREFIX, + timeout: float = 5.0, + max_retries: int = 2, + opener=None, + ) -> None: + self.host = host + self.port = port + self._auth = (username, password) if username is not None else None + self.cgi_prefix = cgi_prefix.strip("/") + self.timeout = timeout + self.max_retries = max_retries + self._opener = opener or _default_opener + + self._session_id: str | None = None + self._has_control = False + self._preset_state = -1 + + # -- URL / request plumbing ------------------------------------------- + + def _url(self, cgi_key: str, params: dict[str, object] | None = None) -> str: + base = f"http://{self.host}:{self.port}/{self.cgi_prefix}/{_CGI[cgi_key]}" + query = dict(params or {}) + if self._session_id is not None and cgi_key not in ("open",): + query.setdefault(_PARAM["session"], self._session_id) + if query: + base = f"{base}?{urllib.parse.urlencode(query)}" + return base + + def _request( + self, cgi_key: str, params: dict[str, object] | None = None, *, timeout: float | None = None + ) -> HttpResponse: + """Issue a CGI request with retry + single reconnect on transient failures.""" + url = self._url(cgi_key, params) + to = self.timeout if timeout is None else timeout + last_exc: Exception | None = None + for attempt in range(self.max_retries + 1): + try: + resp = self._opener(url, None, to, self._auth) + except (CanonConnectionError, CanonTimeoutError) as exc: + last_exc = exc + logger.warning(f"Canon XC request {cgi_key} attempt {attempt + 1} failed: {exc}") + # On a transient failure, drop the (possibly dead) session so the + # next attempt re-opens it. Never reconnect inside a control claim + # silently -- callers re-claim as needed. + if cgi_key != "open" and self._session_id is not None: + self._session_id = None + try: + self.connect(timeout=to) + except CanonTransportError: + pass + url = self._url(cgi_key, params) + continue + if resp.status == 401: + raise CanonSessionError("authentication failed (HTTP 401)") + if resp.status == 403: + raise CanonControlPrivilegeError("control privilege required (HTTP 403)") + if resp.status >= 400: + raise CanonCommandError(f"{cgi_key} failed: HTTP {resp.status}: {resp.text[:200]}") + return resp + assert last_exc is not None + raise last_exc + + @staticmethod + def _parse_kv(text: str) -> dict[str, str]: + """Parse a newline-delimited ``key=value`` CGI response body.""" + result: dict[str, str] = {} + for line in text.replace("\r", "\n").split("\n"): + line = line.strip() + if not line or "=" not in line: + continue + key, _, value = line.partition("=") + result[key.strip()] = value.strip() + return result + + # -- session lifecycle ------------------------------------------------- + + def connect(self, *, timeout: float | None = None) -> None: + if self._session_id is not None: + return + resp = self._request("open", timeout=timeout) + kv = self._parse_kv(resp.text) + # SPEC: the session-id field name. "s" / "session" / "id" are plausible. + session_id = kv.get("s") or kv.get("session") or kv.get("id") + if not session_id: + # Some firmware returns the session id only in a Set-Cookie header. + cookie = resp.headers.get("set-cookie", "") + if "=" in cookie: + session_id = cookie.split(";", 1)[0].split("=", 1)[1] + if not session_id: + raise CanonSessionError(f"open.cgi did not return a session id: {resp.text[:200]!r}") + self._session_id = session_id + logger.info(f"Canon XC session opened to {self.host}:{self.port}") + + def close(self) -> None: + # Idempotent and must never raise. + try: + if self._has_control: + self.yield_control() + except Exception as exc: # pylint: disable=broad-except + logger.warning(f"Canon XC yield during close failed: {exc}") + try: + if self._session_id is not None: + self._request("close") + except Exception as exc: # pylint: disable=broad-except + logger.warning(f"Canon XC close.cgi failed: {exc}") + finally: + self._session_id = None + self._has_control = False + + @property + def connected(self) -> bool: + return self._session_id is not None + + def claim_control(self, *, timeout: float | None = None) -> None: + if self._session_id is None: + raise CanonSessionError("cannot claim control before connecting") + self._request("claim", timeout=timeout) + self._has_control = True + logger.info("Canon XC control privilege claimed") + + def yield_control(self) -> None: + if self._session_id is None or not self._has_control: + self._has_control = False + return + self._request("yield") + self._has_control = False + logger.info("Canon XC control privilege yielded") + + @property + def has_control(self) -> bool: + return self._has_control + + def get_camera_info(self) -> CameraInfo: + resp = self._request("info") + kv = self._parse_kv(resp.text) + return CameraInfo( + model=kv.get("model", ""), + firmware=kv.get("firmware", kv.get("version", "")), + serial=kv.get("serial", ""), + ) + + # -- parameters -------------------------------------------------------- + + def get_parameter(self, name: str) -> object: + kv = self._parse_kv(self._request("info").text) + if name not in kv: + raise CanonParameterError(f"parameter {name!r} not present in info.cgi response") + return kv[name] + + def set_parameter(self, name: str, value: object) -> None: + if not self._has_control: + raise CanonControlPrivilegeError("control privilege required to set parameters") + self._request("control", {name: value}) + + def list_parameters(self) -> dict[str, object]: + return dict(self._parse_kv(self._request("info").text)) + + # -- motion ------------------------------------------------------------ + + def get_position(self, axis: Axis) -> float: + kv = self._parse_kv(self._request("info").text) + key = _PARAM[axis.value] + if key not in kv: + raise CanonParameterError(f"axis {axis.value!r} ({key}) not in info.cgi response") + return _AXIS_ENCODING[axis].to_user(float(kv[key])) + + def move_absolute(self, axis: Axis, value: float, *, speed: float | None = None) -> None: + if not self._has_control: + raise CanonControlPrivilegeError("control privilege required to move") + lim = HARDWARE_LIMITS[axis] + if not lim.low <= value <= lim.high: + raise CanonParameterError( + f"{axis.value}={value} outside hardware range [{lim.low}, {lim.high}]" + ) + params: dict[str, object] = {_PARAM[axis.value]: _AXIS_ENCODING[axis].to_native(value)} + if speed is not None and axis in (Axis.PAN, Axis.TILT): + params[_PARAM[f"{axis.value}_speed"]] = int(round(speed)) + self._request("control", params) + + def jog(self, axis: Axis, speed: float) -> None: + if not self._has_control: + raise CanonControlPrivilegeError("control privilege required to jog") + if axis not in (Axis.PAN, Axis.TILT): + raise CanonCommandError(f"continuous jog not supported for axis {axis.value}") + # SPEC: continuous-move encoding. Modelled as a signed speed on the axis. + self._request("control", {_PARAM[f"{axis.value}_speed"]: int(round(speed))}) + + def is_moving(self, axis: Axis) -> bool: + kv = self._parse_kv(self._request("info").text) + # SPEC: a "moving" flag is assumed; absence means we report False. + flag = kv.get(f"{axis.value}moving") + return flag not in (None, "0", "false", "") + + def stop(self, axis: Axis | None = None) -> None: + 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}) + + # -- focus mode -------------------------------------------------------- + + def get_focus_mode(self) -> FocusMode: + kv = self._parse_kv(self._request("info").text) + wire = kv.get(_PARAM["focus_mode"], "manual") + return _WIRE_TO_FOCUS_MODE.get(wire, FocusMode.MANUAL) + + def set_focus_mode(self, mode: FocusMode) -> None: + if not self._has_control: + raise CanonControlPrivilegeError("control privilege required to set focus mode") + self._request("control", {_PARAM["focus_mode"]: _FOCUS_MODE_TO_WIRE[FocusMode(mode)]}) + + # -- presets ----------------------------------------------------------- + + def recall_preset(self, preset_id: int) -> None: + if not self._has_control: + raise CanonControlPrivilegeError("control privilege required to recall presets") + self._request("preset", {_PARAM["preset"]: int(preset_id)}) + self._preset_state = int(preset_id) + + def get_preset_state(self) -> int: + return self._preset_state diff --git a/tests/tests_devices/test_canon_device.py b/tests/tests_devices/test_canon_device.py new file mode 100644 index 0000000..bc11964 --- /dev/null +++ b/tests/tests_devices/test_canon_device.py @@ -0,0 +1,153 @@ +"""Tests for the CanonPTZCamera ophyd device (simulation mode, no hardware).""" + +import os +from unittest import mock + +import pytest +from ophyd.utils import LimitError + +from csaxs_bec.devices.canon import CanonPTZCamera +from csaxs_bec.devices.canon.transport.base import FocusMode +from csaxs_bec.devices.canon.transport.errors import CanonParameterError +from csaxs_bec.devices.canon.transport.fake import FakeTransport +from csaxs_bec.devices.canon.transport.xc import XCTransport + + +@pytest.fixture +def cam(): + device = CanonPTZCamera(name="canon_test", simulation=True) + device.wait_for_connection() + yield device + device.destroy() + + +def test_connect_claims_control(cam): + assert cam.transport.connected + assert cam.transport.has_control + assert isinstance(cam.transport, FakeTransport) + + +def test_axes_are_positioners_with_spec_limits(cam): + assert cam.pan.limits == (-170.0, 170.0) + assert cam.tilt.limits == (-30.0, 100.0) + assert cam.zoom.limits == (1.0, 20.0) + for ax in (cam.pan, cam.tilt, cam.zoom, cam.focus): + assert hasattr(ax, "move") and hasattr(ax, "stop") + + +def test_move_returns_status_and_reaches_target(cam): + status = cam.pan.move(30.0, wait=True) + assert status.done and status.success + assert cam.pan.position == pytest.approx(30.0) + + +def test_set_alias_is_nonblocking_and_completes(cam): + status = cam.tilt.set(15.0) + status.wait(timeout=5) + assert status.success + assert cam.tilt.position == pytest.approx(15.0) + + +def test_soft_limits_narrow_hardware_limits(): + device = CanonPTZCamera(name="lim", simulation=True) + device.wait_for_connection() + try: + device.pan._limits = (-50.0, 50.0) + with pytest.raises(LimitError): + device.pan.move(120.0) + finally: + device.destroy() + + +def test_parameter_set_with_readback_verification(cam): + cam.gain.set(6.0).wait() + assert cam.gain.get() == 6.0 + assert cam.iris.get() == cam.transport.get_parameter("iris") + + +def test_parameter_readback_mismatch_raises(): + device = CanonPTZCamera(name="mism", simulation=True) + device.wait_for_connection() + try: + # Force the transport to swallow the write so read-back disagrees. + with mock.patch.object(device.transport, "set_parameter"): + status = device.gain.set(9.0) + with pytest.raises(CanonParameterError): + status.wait(timeout=2) + assert status.done and not status.success + finally: + device.destroy() + + +def test_focus_axis_gated_by_focus_mode(cam): + cam.focus_mode.set(FocusMode.CONTINUOUS_AF.value).wait() + with pytest.raises(RuntimeError): + cam.focus.move(0.3) + cam.focus_mode.set(FocusMode.MANUAL.value).wait() + cam.focus.move(0.3, wait=True) + assert cam.focus.position == pytest.approx(0.3) + + +def test_read_and_describe_cover_axes_and_params(cam): + reading = cam.read() + assert "canon_test_pan_readback" in reading + desc = cam.describe() + assert desc # non-empty + cfg = cam.read_configuration() + assert "canon_test_gain" in cfg + + +def test_stop_all_halts_axes(cam): + with mock.patch.object(cam.pan, "stop") as p, mock.patch.object(cam.tilt, "stop") as t: + cam.stop_all() + p.assert_called_once() + t.assert_called_once() + + +def test_recall_preset(cam): + cam.recall_preset(5) + assert cam.preset_state.get() == 5 + + +def test_grab_api_requires_buffer(cam): + with pytest.raises(RuntimeError): + cam.get_latest() + + +def test_streaming_in_simulation_fills_buffer(): + device = CanonPTZCamera(name="stream_sim", simulation=True, buffer_maxlen=8) + device.wait_for_connection() + try: + device.start_stream() + assert device._worker.wait_for_first_frame(timeout=2.0) + frame = device.get_latest() + assert frame is not None and frame.data.startswith(b"\xff\xd8") + stats = device.stream_stats() + assert stats["buffer"]["size"] >= 1 + assert stats["worker"]["establishment_time_s"] is not None + finally: + device.stop_stream() + device.destroy() + + +def test_non_simulation_requires_host(): + with pytest.raises(ValueError): + CanonPTZCamera(name="nohost", simulation=False) + + +def test_invalid_stream_start_rejected(): + with pytest.raises(ValueError): + CanonPTZCamera(name="bad", simulation=True, stream_start="whenever") + + +def test_real_transport_reads_credentials_from_env(): + env = {"CANON_REALCAM_USER": "operator", "CANON_REALCAM_PASSWORD": "secret"} + with mock.patch.dict(os.environ, env, clear=False): + device = CanonPTZCamera(name="realcam", simulation=False, host="192.168.0.50") + assert isinstance(device.transport, XCTransport) + assert device.transport._auth == ("operator", "secret") + + +def test_destroy_closes_transport(cam): + cam.destroy() + assert not cam.transport.connected diff --git a/tests/tests_devices/test_canon_ring_buffer.py b/tests/tests_devices/test_canon_ring_buffer.py new file mode 100644 index 0000000..6173027 --- /dev/null +++ b/tests/tests_devices/test_canon_ring_buffer.py @@ -0,0 +1,83 @@ +"""Tests for the Canon acquisition ring buffer (pure stdlib, no hardware).""" + +import threading + +import pytest + +from csaxs_bec.devices.canon.acquisition.ring_buffer import Frame, FrameRingBuffer + + +def test_rejects_bad_maxlen(): + with pytest.raises(ValueError): + FrameRingBuffer(0) + + +def test_put_stamps_sequence_and_timestamps(): + clock = iter([1.0, 2.0, 3.0]) + wall = iter([100.0, 200.0, 300.0]) + buf = FrameRingBuffer( + 8, source_id="cam", time_func=lambda: next(clock), wall_time_func=lambda: next(wall) + ) + f0 = buf.put(b"a") + f1 = buf.put(b"bb") + assert isinstance(f0, Frame) + assert (f0.seq, f0.monotonic_ts, f0.wall_ts, f0.source_id) == (0, 1.0, 100.0, "cam") + assert (f1.seq, f1.nbytes) == (1, 2) + + +def test_get_latest_empty_and_filled(): + buf = FrameRingBuffer(4) + assert buf.get_latest() is None + buf.put(b"x") + buf.put(b"y") + assert buf.get_latest().data == b"y" + + +def test_overwrites_oldest_and_counts_drops(): + buf = FrameRingBuffer(3) + for i in range(5): + buf.put(bytes([i])) + snap = buf.snapshot() + assert [f.data for f in snap] == [b"\x02", b"\x03", b"\x04"] # oldest two evicted + stats = buf.stats() + assert stats["size"] == 3 + assert stats["total_received"] == 5 + assert stats["dropped_unread"] == 2 + + +def test_get_latest_n(): + buf = FrameRingBuffer(5) + for i in range(4): + buf.put(bytes([i])) + assert [f.data for f in buf.get_latest_n(2)] == [b"\x02", b"\x03"] + assert len(buf.get_latest_n(10)) == 4 # capped at available + assert buf.get_latest_n(0) == [] + + +def test_get_nearest_wall_and_monotonic(): + mono = iter([1.0, 2.0, 3.0]) + wall = iter([10.0, 20.0, 30.0]) + buf = FrameRingBuffer(8, time_func=lambda: next(mono), wall_time_func=lambda: next(wall)) + buf.put(b"a") + buf.put(b"b") + buf.put(b"c") + assert buf.get_nearest(19.0).data == b"b" # wall clock default + assert buf.get_nearest(2.9, clock="monotonic").data == b"c" + empty = FrameRingBuffer(2) + assert empty.get_nearest(0.0) is None + + +def test_thread_safety_under_concurrent_writers(): + buf = FrameRingBuffer(1000) + + def worker(tag): + for i in range(200): + buf.put(f"{tag}-{i}".encode()) + + threads = [threading.Thread(target=worker, args=(t,)) for t in range(5)] + for t in threads: + t.start() + for t in threads: + t.join() + assert buf.stats()["total_received"] == 1000 + assert len(buf) == 1000 diff --git a/tests/tests_devices/test_canon_stream_worker.py b/tests/tests_devices/test_canon_stream_worker.py new file mode 100644 index 0000000..c7e943d --- /dev/null +++ b/tests/tests_devices/test_canon_stream_worker.py @@ -0,0 +1,113 @@ +"""Tests for the persistent stream worker (fail-soft, instrumented).""" + +import threading +import time + +from csaxs_bec.devices.canon.acquisition.ring_buffer import FrameRingBuffer +from csaxs_bec.devices.canon.acquisition.stream_worker import ( + FrameSource, + IterableFrameSource, + MJPEGHttpSource, + StreamWorker, +) + + +def _wait(predicate, timeout=2.0): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return True + time.sleep(0.01) + return False + + +def test_worker_fills_buffer_and_measures_establishment(): + buf = FrameRingBuffer(16) + frames = [b"\xff\xd8a\xff\xd9", b"\xff\xd8b\xff\xd9", b"\xff\xd8c\xff\xd9"] + worker = StreamWorker(IterableFrameSource(frames, delay=0.01), buf, name="t") + worker.start() + assert worker.wait_for_first_frame(timeout=2.0) + assert _wait(lambda: len(buf) == 3) + worker.stop() + assert worker.establishment_time is not None + assert worker.stats()["frames_received"] == 3 + assert not worker.running + + +def test_worker_reconnects_fail_soft_and_keeps_serving(): + buf = FrameRingBuffer(16) + + class FlakySource(FrameSource): + def __init__(self): + self.opens = 0 + self._closed = False + + def open(self, *, timeout=None): + self.opens += 1 + self._closed = False + + def frames(self): + if self.opens == 1: + yield b"good1" + raise ConnectionError("drop") # first connection dies + # second connection delivers steadily + while not self._closed: + yield b"good2" + time.sleep(0.01) + + def close(self): + self._closed = True + + src = FlakySource() + worker = StreamWorker(src, buf, name="flaky", reconnect_delay=0.05) + worker.start() + # The first good frame is served even though the connection later drops. + assert _wait(lambda: buf.get_latest() is not None) + # It reconnects (fail-soft) and keeps delivering. + assert _wait(lambda: src.opens >= 2 and buf.get_latest().data == b"good2") + worker.stop() + assert worker.stats()["reconnects"] >= 1 + + +def test_worker_start_is_idempotent(): + buf = FrameRingBuffer(8) + worker = StreamWorker(IterableFrameSource(lambda: b"x", delay=0.01), buf, name="idem") + worker.start() + t1 = worker._thread + worker.start() # no-op while running + assert worker._thread is t1 + worker.stop() + + +def test_mjpeg_source_splits_frames_without_decoding(): + # Build a fake multipart MJPEG byte stream with two JPEG frames split across reads. + stream = b"--bound\r\n\xff\xd8AAA\xff\xd9garbage\xff\xd8BBBB\xff\xd9tail" + + class FakeResponse: + def __init__(self, data, chunk): + self._data = data + self._chunk = chunk + self._pos = 0 + + def read(self, n): + chunk = self._data[self._pos : self._pos + self._chunk] + self._pos += self._chunk + return chunk + + def close(self): + pass + + src = MJPEGHttpSource("http://x/image.cgi", chunk_size=5) + src._response = FakeResponse(stream, 5) + got = list(src.frames()) + assert got == [b"\xff\xd8AAA\xff\xd9", b"\xff\xd8BBBB\xff\xd9"] + + +def test_worker_thread_stops_cleanly_no_leak(): + buf = FrameRingBuffer(8) + before = threading.active_count() + worker = StreamWorker(IterableFrameSource(lambda: b"y", delay=0.01), buf, name="leak") + worker.start() + assert _wait(lambda: len(buf) > 0) + worker.stop() + assert _wait(lambda: threading.active_count() == before) diff --git a/tests/tests_devices/test_canon_transport.py b/tests/tests_devices/test_canon_transport.py new file mode 100644 index 0000000..8877e4f --- /dev/null +++ b/tests/tests_devices/test_canon_transport.py @@ -0,0 +1,238 @@ +"""Tests for the Canon transport layer: FakeTransport behaviour and the XCTransport +HTTP/session logic driven through a mocked opener (no hardware, no network).""" + +import pytest + +from csaxs_bec.devices.canon.transport.base import Axis, FocusMode +from csaxs_bec.devices.canon.transport.errors import ( + CanonCommandError, + CanonConnectionError, + CanonControlPrivilegeError, + CanonParameterError, + CanonSessionError, +) +from csaxs_bec.devices.canon.transport.fake import FakeTransport +from csaxs_bec.devices.canon.transport.xc import HttpResponse, XCTransport + +# --------------------------------------------------------------------------- # +# FakeTransport # +# --------------------------------------------------------------------------- # + + +@pytest.fixture +def fake(): + t = FakeTransport() + t.connect() + t.claim_control() + return t + + +def test_fake_session_lifecycle(): + t = FakeTransport() + assert not t.connected + t.connect() + assert t.connected and not t.has_control + t.claim_control() + assert t.has_control + t.yield_control() + assert not t.has_control + t.close() + assert not t.connected + + +def test_fake_requires_control_to_move(): + t = FakeTransport() + t.connect() + with pytest.raises(CanonControlPrivilegeError): + t.move_absolute(Axis.PAN, 10.0) + + +def test_fake_requires_session(): + t = FakeTransport() + with pytest.raises(CanonSessionError): + t.get_position(Axis.PAN) + + +def test_fake_absolute_move_and_limits(fake): + fake.move_absolute(Axis.PAN, 90.0) + assert fake.get_position(Axis.PAN) == 90.0 + with pytest.raises(CanonParameterError): + fake.move_absolute(Axis.PAN, 200.0) # beyond +/-170 + with pytest.raises(CanonParameterError): + fake.move_absolute(Axis.TILT, -45.0) # below -30 + + +def test_fake_parameters_roundtrip(fake): + assert fake.get_parameter("gain") == 0.0 + fake.set_parameter("gain", 6.0) + assert fake.get_parameter("gain") == 6.0 + with pytest.raises(CanonParameterError): + fake.get_parameter("does_not_exist") + + +def test_fake_focus_mode_and_presets(fake): + assert fake.get_focus_mode() == FocusMode.MANUAL + fake.set_focus_mode(FocusMode.FACE_AF) + assert fake.get_focus_mode() == FocusMode.FACE_AF + assert fake.get_preset_state() == -1 + fake.recall_preset(3) + assert fake.get_preset_state() == 3 + + +def test_fake_gradual_motion_with_injected_clock(): + now = [0.0] + t = FakeTransport(gradual_motion=True, time_func=lambda: now[0]) + t.connect() + t.claim_control() + t.move_absolute(Axis.PAN, 60.0, speed=60.0) # 60 deg/s -> 1 s total + now[0] = 0.5 + assert t.is_moving(Axis.PAN) + assert t.get_position(Axis.PAN) == pytest.approx(30.0) + now[0] = 1.0 + assert t.get_position(Axis.PAN) == pytest.approx(60.0) + assert not t.is_moving(Axis.PAN) + + +def test_fake_stop_halts_axis(): + now = [0.0] + t = FakeTransport(gradual_motion=True, time_func=lambda: now[0]) + t.connect() + t.claim_control() + t.move_absolute(Axis.PAN, 100.0, speed=100.0) + now[0] = 0.3 + t.stop(Axis.PAN) + pos = t.get_position(Axis.PAN) + now[0] = 5.0 + assert t.get_position(Axis.PAN) == pytest.approx(pos) # frozen after stop + + +def test_fake_fault_injection(fake): + fake.fail_next["move_absolute:pan"] = CanonCommandError("boom") + with pytest.raises(CanonCommandError): + fake.move_absolute(Axis.PAN, 10.0) + + +# --------------------------------------------------------------------------- # +# XCTransport with a recording mock opener # +# --------------------------------------------------------------------------- # + + +class RecordingOpener: + """Mock HTTP opener; records URLs and returns canned bodies keyed by CGI name.""" + + def __init__(self, bodies=None, status=200): + self.requests = [] + self.bodies = bodies or {} + self.status = status + self.fail_times = 0 # raise CanonConnectionError this many times first + + def __call__(self, url, data, timeout, auth): + self.requests.append(url) + if self.fail_times > 0: + self.fail_times -= 1 + raise CanonConnectionError("transient") + # pick body by CGI filename present in URL + for key, body in self.bodies.items(): + if f"/{key}" in url: + return HttpResponse(status=self.status, body=body.encode(), headers={}) + return HttpResponse(status=self.status, body=b"", headers={}) + + +def make_xc(opener): + return XCTransport(host="cam.test", port=80, username="u", password="p", opener=opener) + + +def test_xc_open_parses_session_and_builds_urls(): + opener = RecordingOpener(bodies={"open.cgi": "s=ABC123\n"}) + t = make_xc(opener) + t.connect() + assert t.connected + assert "open.cgi" in opener.requests[0] + # subsequent requests carry the session id + t.claim_control() + assert any("claim.cgi" in u and "s=ABC123" in u for u in opener.requests) + assert t.has_control + + +def test_xc_open_via_set_cookie_header(): + class CookieOpener(RecordingOpener): + def __call__(self, url, data, timeout, auth): + self.requests.append(url) + return HttpResponse(200, b"", {"set-cookie": "session=COOKIEVAL; path=/"}) + + t = make_xc(CookieOpener()) + t.connect() + assert t.connected + + +def test_xc_move_absolute_encodes_native_value_and_session(): + opener = RecordingOpener(bodies={"open.cgi": "s=S1\n"}) + t = make_xc(opener) + t.connect() + t.claim_control() + t.move_absolute(Axis.PAN, 12.0, speed=30.0) + move_url = [u for u in opener.requests if "control.cgi" in u][-1] + assert "pan=1200" in move_url # 12.0 deg * scale(100) = 1200 native + assert "panspeed=30" in move_url + assert "s=S1" in move_url + + +def test_xc_get_position_decodes_native(): + opener = RecordingOpener(bodies={"open.cgi": "s=S1\n", "info.cgi": "pan=4500\ntilt=-300\n"}) + t = make_xc(opener) + t.connect() + assert t.get_position(Axis.PAN) == pytest.approx(45.0) + assert t.get_position(Axis.TILT) == pytest.approx(-3.0) + + +def test_xc_move_requires_control(): + opener = RecordingOpener(bodies={"open.cgi": "s=S1\n"}) + t = make_xc(opener) + t.connect() + with pytest.raises(CanonControlPrivilegeError): + t.move_absolute(Axis.PAN, 10.0) + + +def test_xc_http_401_and_403_map_to_typed_errors(): + opener = RecordingOpener(bodies={"open.cgi": "s=S1\n"}, status=401) + t = make_xc(opener) + with pytest.raises(CanonSessionError): + t.connect() + + opener2 = RecordingOpener(bodies={"open.cgi": "s=S1\n"}) + t2 = make_xc(opener2) + t2.connect() + opener2.status = 403 + with pytest.raises(CanonControlPrivilegeError): + t2.claim_control() + + +def test_xc_retries_and_reconnects_on_transient_failure(): + opener = RecordingOpener(bodies={"open.cgi": "s=S1\n", "info.cgi": "pan=0\n"}) + t = make_xc(opener) + t.connect() + # next info request fails once, then the transport reconnects + retries. + opener.fail_times = 1 + assert t.get_position(Axis.PAN) == pytest.approx(0.0) + assert opener.requests.count("http://cam.test:80/-wvhttp-01-/open.cgi") >= 2 + + +def test_xc_close_is_idempotent_and_safe(): + opener = RecordingOpener(bodies={"open.cgi": "s=S1\n"}) + t = make_xc(opener) + t.connect() + t.claim_control() + t.close() + assert not t.connected and not t.has_control + t.close() # second call must not raise + + +def test_xc_focus_mode_roundtrip_wire_tokens(): + opener = RecordingOpener(bodies={"open.cgi": "s=S1\n", "info.cgi": "focusmode=face\n"}) + t = make_xc(opener) + t.connect() + t.claim_control() + assert t.get_focus_mode() == FocusMode.FACE_AF + t.set_focus_mode(FocusMode.CONTINUOUS_AF) + set_url = [u for u in opener.requests if "control.cgi" in u][-1] + assert "focusmode=auto" in set_url