diff --git a/csaxs_bec/device_configs/canon_crn300.yaml b/csaxs_bec/device_configs/canon_crn300.yaml index b3bdd7d3..f4002047 100644 --- a/csaxs_bec/device_configs/canon_crn300.yaml +++ b/csaxs_bec/device_configs/canon_crn300.yaml @@ -18,10 +18,22 @@ ptz_cam: # 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) + # --- Phase-2 acquisition (safe to leave at defaults for control-only use) --- + # acquisition_mode picks the network footprint; the grab API is the same for + # mjpeg and jpeg_poll: + # jpeg_poll one still JPEG every 1/poll_rate s. Low bandwidth, idle CPU + # ~0, full grab API, frame may be <= 1 poll interval old. + # mjpeg continuous MJPEG connection held open all session. Lowest + # latency and full frame rate, but continuous 4K bandwidth -- + # pick this only when high rate is genuinely needed. + # jpeg_on_demand no worker at all; each get_latest() fetches one still. Zero + # idle cost, but no history: get_latest_n/get_nearest raise. + acquisition_mode: "jpeg_poll" + poll_rate: 2.0 # [Hz] jpeg_poll only + stream_start: "off" # off | connect | stage (when to start acquisition) buffer_maxlen: 64 # ring-buffer depth (compressed frames) - # stream_url: http://192.168.0.100/-wvhttp-01-/image.cgi # MJPEG URL override + # still_url: http://192.168.0.100/-wvhttp-01-/image.cgi # jpeg_poll URL override + # stream_url: http://192.168.0.100/-wvhttp-01-/image.cgi # mjpeg URL override deviceTags: - camera - ptz @@ -39,7 +51,9 @@ ptz_cam_sim: simulation: true gradual_motion: true # simulate finite-speed travel (exercise move status) buffer_maxlen: 32 - stream_start: "connect" # start the synthetic stream at connect + acquisition_mode: "jpeg_poll" + poll_rate: 5.0 # faster than the real default; keeps the sim lively + stream_start: "connect" # start the synthetic acquisition at connect deviceTags: - camera - ptz diff --git a/csaxs_bec/devices/canon/COMMISSIONING.md b/csaxs_bec/devices/canon/COMMISSIONING.md index 9d09530a..924e7b0d 100644 --- a/csaxs_bec/devices/canon/COMMISSIONING.md +++ b/csaxs_bec/devices/canon/COMMISSIONING.md @@ -8,6 +8,10 @@ baked into the code** β€” without reverse-engineering the source. It contains: +> **Hardware has arrived?** Go straight to +> [Stage 0.5 β€” First contact](#stage-05--first-contact-hardware-just-arrived--start-here), +> and run `python -m csaxs_bec.devices.canon.probe --host ` as the first command. + 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) @@ -62,11 +66,11 @@ name used is **`ptz_cam`** (from [`device_configs/canon_crn300.yaml`](../../devi | `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.start_stream()` / `stop_stream()` | USER_ACCESS | Start/stop acquisition (no worker in `jpeg_on_demand`) | 🟑 [R10](#r10) | +| `dev.ptz_cam.get_latest()` | USER_ACCESS | Newest buffered frame, or one live fetch in `jpeg_on_demand` | 🟑 [R10](#r10) | +| `dev.ptz_cam.get_latest_n(n)` | USER_ACCESS | Last *n* frames, oldest-first (raises in `jpeg_on_demand`) | 🟒 | +| `dev.ptz_cam.get_nearest(ts)` | USER_ACCESS | Frame closest to a wall-clock timestamp (raises in `jpeg_on_demand`) | 🟒 | +| `dev.ptz_cam.stream_stats()` | USER_ACCESS | Mode + 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) @@ -135,16 +139,29 @@ Each axis is a full ophyd positioner. #### 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). +Behaviour depends on `acquisition_mode` (`deviceConfig`), which selects the network +footprint behind one grab API. Default is **`jpeg_poll`**. + +| Mode | What `start_stream()` does | Idle bandwidth | `get_latest()` | History API | +|---|---|---|---|---| +| `jpeg_poll` *(default)* | starts a worker polling the still endpoint at `poll_rate` Hz | low | O(1), ≀ 1 poll old | βœ… | +| `mjpeg` | opens the continuous MJPEG stream once, starts the worker | high (continuous 4K) | O(1), sub-ms | βœ… | +| `jpeg_on_demand` | prepares the buffer only β€” **no worker, no connection held** | none | one synchronous HTTP fetch | ❌ raises | + +- **`start_stream()`** β€” idempotent. ([R10](#r10) governs the endpoints.) +- **`get_latest()`** β€” returns a `Frame` (compressed bytes + monotonic & wall-clock + timestamps + source id); `decode=True` decodes **only that frame** to an RGB array + via `cv2`. In `jpeg_on_demand` this performs one round-trip through the control + session (`transport.get_still_jpeg()`) instead of reading the buffer. +- **`get_latest_n(n)` / `get_nearest(ts)`** β€” O(1) retrieval from buffered history. + In `jpeg_on_demand` there is no history, so both **raise `RuntimeError`** by design: + silently returning a stale frame for a timestamp query would mis-correlate images + with scan points. +- **`stream_stats()`** β€” returns `{"mode": …, "buffer": {...}, "worker": {...}}` + (plus `poll_rate_hz` in `jpeg_poll`), including `establishment_time_s` (time to first + frame) and `last_frame_age_s` (health). No `worker` key in `jpeg_on_demand`. +- **`stop_stream()`** β€” stops the worker cleanly (no leaked socket/thread); a no-op in + `jpeg_on_demand`. #### Diagnostics @@ -155,7 +172,7 @@ Each axis is a full ophyd positioner. ## 3. What has already been tested -**47 automated tests, no hardware required** (`pytest tests/tests_devices/test_canon_*.py +**76 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 @@ -167,24 +184,40 @@ Coverage by area: 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**. +**Stream worker & frame sources (10 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**. For `SingleJpegPollSource`: +one frame per interval at the configured rate; the **same worker + buffer + `get_nearest` +history** work unchanged through the polled source (the seam does its job); `close()` +**interrupts a long poll interval** so `stop()` stays responsive; bad rate/config +rejected; the HTTP path issues a fresh request per poll and holds no connection. -**Transport β€” `FakeTransport` (9 tests)** β€” session lifecycle; **control-privilege is +**Transport β€” `FakeTransport` (11 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. +with an injected clock; stop freezes an axis; fault injection; synthetic stills are +JPEG-framed and distinct per call; a transport without an image path reports single-frame +capture as unsupported rather than failing obscurely. -**Transport β€” `XCTransport` via mocked HTTP (9 tests)** β€” parses the session id from the +**Transport β€” `XCTransport` via mocked HTTP (12 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. +never raises; focus-mode wire-token round-trip; single-frame capture returns the raw body +and carries the session, needs a session but **not** the control privilege, and **rejects +a non-JPEG body** (an HTML error page served with HTTP 200 fails at the grab, not at +decode). -**Device logic (17 tests)** β€” connect claims control; the four axes are positioners with +**Probe, end-to-end against a mock camera (10 tests)** β€” the success path reports a +session and parses model/firmware; the sweep identifies the **right prefix** and +eliminates wrong ones; the still endpoint is flagged; **path discovery works without +credentials** (401s confirm endpoints, and nothing is mistaken for an image); `main()` +runs end-to-end and writes its JSON report; and a camera that speaks **none** of our +assumptions degrades cleanly instead of crashing. + +**Device logic (26 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**; @@ -192,7 +225,12 @@ parameter set **with read-back verification**, and a **mismatch fails the status 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. +`destroy()` closes the transport. **Acquisition modes:** the default is `jpeg_poll`; +invalid mode/`poll_rate` rejected at construction; `jpeg_poll` fills the buffer and keeps +the **full** grab API incl. `get_nearest`; `jpeg_poll` targets the **still** endpoint (not +the stream URL); `jpeg_on_demand` starts **no worker**, fetches fresh bytes **per grab**, +works **without** `start_stream()`, and **raises** on `get_latest_n`/`get_nearest`; +`mjpeg` still streams and honours a `stream_url` override. --- @@ -210,7 +248,9 @@ is the actual subject of [the test regime in Β§6](#6-commissioning-test-regime). | **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). | +| **Still endpoint** | That the camera serves a complete standalone JPEG there β€” both default modes rest on it β€” [R10](#r10). | +| **Acquisition establishment & latency** | The real setup cost (~2 s for `mjpeg`, one round-trip for `jpeg_poll`), steady frame rate, and grab latency β€” [R10](#r10). | +| **Actual bandwidth per mode** | That `jpeg_poll` really is the light option on this network, and what `mjpeg` costs β€” measurable only against the camera. | | **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. | @@ -269,10 +309,19 @@ 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`). +**R10 β€” Image endpoints (stream *and* still).** Two separate SPEC +constants in `xc.py`, both defaulting to `image.cgi`; on real firmware they may differ +(e.g. a `nphMotionJpeg`-style push stream vs a single-JPEG CGI): + +| Constant | Used by | Default | Symptom if wrong | +|---|---|---|---| +| `DEFAULT_STREAM_CGI` | `acquisition_mode: mjpeg` β€” parsed as multipart MJPEG split on `FFD8…FFD9` | `image.cgi` | worker never produces a frame | +| `DEFAULT_STILL_CGI` (also `_CGI["still"]`) | `jpeg_poll` (polled over HTTP) and `jpeg_on_demand` (via `transport.get_still_jpeg()`) | `image.cgi` | `CanonCommandError: … did not return a JPEG` | + +*Confirm:* both real URLs, and that MJPEG frames are standalone JPEGs. *Fix:* the two +constants, or per-device overrides `stream_url` / `still_url`. Note `get_still_jpeg()` +already rejects a non-JPEG body up front, so a wrong still endpoint fails at the grab +with a clear message rather than at decode time. --- @@ -280,7 +329,22 @@ fails. *Fix:* the `stream_url` config key (or `_default_stream_url`). 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. +to suspect. Stage 0 needs no camera; **Stage 0.5 is where you start once hardware is on +the bench**; Stage 1+ need it on the network. + +| Stage | What it establishes | Needs | +|-------|--------------------|-------| +| **0** | The code and BEC plumbing are sound, independently of hardware | nothing | +| **0.5** | **First contact: address, credentials, and what protocol it really speaks** | camera on a bench | +| 1 | XC session opens from the device-server host | camera on network | +| 2 | Positions and parameters read back, correctly scaled | β€” | +| 3 | Control privilege can be claimed and released | β€” | +| 4 | All four axes move accurately, respect limits, and stop | **safe mounting** | +| 5 | Focus modes switch; the manual-focus gate holds | β€” | +| 6 | Presets recall | presets stored | +| 7 | Acquisition establishes, grabs are cheap, decode works, bandwidth is acceptable | β€” | +| 8 | Full BEC integration across a real scan | BEC session | +| 9 | Soak: no leaks, session survives hours | time | > **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 @@ -289,26 +353,170 @@ to suspect. Stages 0 can be done **now**; Stage 1+ need the camera on the networ ### 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 +--random-order`. *Expect 76 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 +0.2 **Point the editable install at *this* worktree β€” before starting BEC.** `csaxs_bec` +is checked out as several git worktrees but only one is the active editable install, and +the DeviceServer can only import `deviceClass` from that one. Check it **from a directory +that is not a worktree root** β€” inside a worktree, the local `csaxs_bec/` folder shadows +the install and the check silently lies: + +```sh +cd /tmp && "$BEC_STACK"/.venv/bin/python -c "import csaxs_bec; print(csaxs_bec.__file__)" +``` + +If that is not `…/csaxs_bec-canon/…`, repoint it and restart the services: + +```sh +cd "$BEC_STACK" && .venv/bin/pip install -e ./csaxs_bec-canon --no-deps +becstop && becstart +``` + +β†’ skipping this gives `ModuleNotFoundError: No module named 'csaxs_bec.devices.canon'`, +the config load is flushed, and the client reports a `DeviceConfigError`. + +0.3 **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 +0.4 **Rehearse the probe against a mock camera.** Learn what a PASS looks like *before* +hardware day, on a fake camera that speaks the protocol we assume: + +```sh +python tests/tests_devices/test_canon_probe.py # serves a fake camera on :8400 +# in another shell: +python -m csaxs_bec.devices.canon.probe --host 127.0.0.1 --port 8400 \ + --user operator --password secret +``` + +*Expect* the sweep to find prefix `/-wvhttp-01-/`, `image.cgi` flagged `<-- JPEG!`, step 5 +to report a session id, and the verdict **"full XC session established"**. That is the +output you are hoping to see from the real camera in Stage 0.5.2 β€” anything less tells you +which assumption broke. + +Rehearse the **single-file** mode too, since that is how you will run it at the beamline: +```sh +cp csaxs_bec/devices/canon/probe.py /tmp/ && (cd /tmp && python3 probe.py --host 127.0.0.1 \ + --port 8400 --user operator --password secret) +``` +*Expect* steps 1–5 identical, and step 6 to report `SKIP … standalone file`. + +0.5 **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. +**Gate:** 76 green + the canon worktree is the live install + sim device usable in BEC + +probe rehearsal shows a full session + SPEC tables reviewed. β†’ proceed. + +### Stage 0.5 β€” First contact (hardware just arrived) β€” **start here** + +**Goal: establish *any* connection whatsoever, and find out what protocol this camera +actually speaks.** Everything downstream assumes the `# SPEC:` tables are right, and they +are educated guesses. Resolve them empirically now β€” one afternoon here saves days of +one-round-trip-per-guess later. **Do not skip to Stage 1**: Stage 1.2 presumes the CGI +prefix and endpoint names are already correct. + +> **Bench it first.** Do all of Stage 0.5 with the camera on a bench or an isolated +> switch, not mounted in the hutch. You will power-cycle it, factory-reset it, and change +> its network settings. Mount it only once Stage 3 passes. + +0.5.1 **Power, link, address.** Power up, confirm the network LED links. Get its IP: +- the camera's own menu/OSD or its display, if it has one β€” fastest and unambiguous; +- otherwise `arp -a` on a machine on the same segment before/after powering it up, or ask + the network group for the DHCP lease on its MAC (the MAC is usually on the chassis label). + +Record: **IP, MAC, firmware version (from the label), factory-default credentials** (from +the manual β€” CR-series cameras typically ship with an admin account that must be given a +password on first login). + +0.5.2 **Run the probe.** One read-only command that answers "is it there, what is it, and +which of our assumptions hold". **Run it on a beamline machine** β€” that avoids tunnelling +entirely and tests from where BEC will actually run. + +The probe is stdlib-only and Python 3.7+, so it needs no BEC, no venv and no checkout β€” +copy the single file across: + +```bash +scp csaxs_bec/devices/canon/probe.py @:/tmp/ +ssh @ +export CANON_PROBE_USER= CANON_PROBE_PASSWORD='' # not on argv: `ps` leaks it +python3 /tmp/probe.py --host --json /tmp/first_contact.json +``` + +It checks TCP, identifies the HTTP server, sweeps candidate CGI prefixes Γ— endpoint names +([R1](#r1)/[R2](#r2)), flags anything returning real JPEG bytes ([R10](#r10)), opens a +session with a **built-in stdlib attempt against the prefix it actually discovered**, and +β€” when run from a checkout rather than as a copied file β€” also tries `XCTransport.connect()` +as shipped. It sends **GETs only** β€” no control or motion commands β€” and talks to the one +host you name. + +> Step 6 is skipped in single-file mode and says so; step 5 already answers "can a session +> be opened, and under which prefix". Re-run from a checkout later to validate the shipped +> transport itself. + +Read the status codes, not just pass/fail: **`401` means the endpoint *exists* but needs +auth** β€” that confirms a path even before you have a working login. `404` is what +eliminates a candidate. Keep `first_contact.json`; it is the baseline record. + +*If the probe reaches step 5 and reports a session:* the guesses fit β€” skip ahead to +Stage 1.3 (identity / info parse). + +0.5.3 **The camera's own web UI is the spec.** If the sweep found nothing, stop guessing +and harvest the protocol: +1. Browse to `http:///`, log in, and confirm you get live video. **This alone proves + the camera, network, and credentials are all good** β€” a big result on day one, even if + our client can't talk to it yet. +2. Open **DevTools β†’ Network**, then drive the UI: move the pan/tilt control, change zoom, + switch focus mode, recall a preset. +3. Every request the camera's *own* client makes is ground truth. Record the exact URLs. + +This single trace typically resolves **[R1](#r1) (prefix), [R2](#r2) (endpoint names), +[R4](#r4) (parameter names), [R5](#r5) (axis encoding β€” watch what integer a known +on-screen angle produces), [R7](#r7) (response format) and [R10](#r10) (image URLs)** +at once. Save a HAR export next to `first_contact.json`. + +Also check the camera's settings pages for an **"XC protocol" / "external control" / +"IP control" toggle β€” it may be off by default**, and no amount of correct URLs will work +until it is on. Note the control port it names; it need not be 80. + +0.5.4 **Reproduce one call outside the browser.** Take the simplest URL from the trace +(ideally the still image) and fetch it with `curl`: +```bash +curl -v --digest -u : "http:////" -o still.jpg +curl -v --basic -u : "http:////" -o still.jpg +``` +*Confirm* `still.jpg` opens as an image. This proves the **auth scheme** (note which of +`--digest`/`--basic` worked β€” the `WWW-Authenticate` header in the probe output says so +too) and that the camera is drivable headless, not just from its own JavaScript. + +0.5.5 **Correct the SPEC tables, then re-probe.** Edit the marked tables at the top of +[`transport/xc.py`](./transport/xc.py) β€” `DEFAULT_CGI_PREFIX`, `_CGI`, `DEFAULT_STILL_CGI` +/ `DEFAULT_STREAM_CGI`, `_PARAM`, `_AXIS_ENCODING` β€” to match what you observed. Re-run +the probe until step 5 reports a session. **Commit the corrected tables with the evidence +in the message** (the observed URL for each change); they are the documented truth for the +next deployment. + +**Gate:** you know the camera's address and credentials, its web UI shows live video, one +call works from `curl`, and the probe opens an XC session. β†’ proceed to Stage 1. + +> **If the camera turns out not to speak XC at all** (no session CGIs anywhere in the +> trace, only a plain still/MJPEG endpoint): that is a finding, not a dead end. The +> architecture already covers it β€” `jpeg_poll` needs *only* a still URL, so imaging can +> proceed independently while control is re-targeted at whatever the trace showed +> (a VISCA-over-IP transport behind the same `CameraTransport` interface). Say so early; +> it changes the schedule, not the design. ### Stage 1 β€” Network & session (camera on network; read-only) -1.1 **Reachability.** From the device-server host: `ping `, then +*Prerequisite: Stage 0.5 passed, so the SPEC tables now reflect the real camera.* + +1.1 **Reachability.** From the **device-server host** (not your laptop β€” this re-checks +routing/firewall from where BEC will actually run): `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 @@ -319,7 +527,7 @@ 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). +[R6b](#r6b); 404 β‡’ [R1](#r1)/[R2](#r2) (and Stage 0.5.5 was incomplete). 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 β‡’ @@ -395,24 +603,54 @@ move/error β‡’ [R3](#r3). **Gate:** presets recall. β†’ proceed. -### Stage 7 β€” Streaming & grab (Phase 2) +### Stage 7 β€” Acquisition & grab (Phase 2) + +Do **7.0 first**: it is the cheapest way to validate the still endpoint, and both default +modes depend on it. Then run 7.1–7.5 for the mode(s) you intend to deploy. + +7.0 **Still endpoint (all modes).** With a session open: +```python +data = t.get_still_jpeg(); print(len(data), data[:4]) +``` +*Confirm* a JPEG comes back (it starts `ff d8`; the call raises otherwise). Write it to +disk and open it. β†’ `CanonCommandError: … did not return a JPEG` or a 404 β‡’ [R10](#r10) +`DEFAULT_STILL_CGI`. **This one call de-risks `jpeg_poll` and `jpeg_on_demand` together.** 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). +*Confirm* `mode` is what you configured, `worker.establishment_time_s` is populated, and +`buffer.size` grows. In `jpeg_poll` establishment should be ~one HTTP round-trip (not +~2 s) and `buffer.size` should grow at about `poll_rate` Hz; in `mjpeg` expect the ~2 s +cost **once**. β†’ never any frame β‡’ [R10](#r10). *(In `jpeg_on_demand` there is no worker: +`stream_stats()` has no `worker` key β€” that is correct, skip to 7.3.)* -7.2 **Grab latency.** Time `dev.ptz_cam.get_latest()` β€” expect sub-millisecond, **no** -re-establishment cost. Grab repeatedly during simulated DAQ activity. +7.2 **Grab latency.** Time `dev.ptz_cam.get_latest()` β€” in the buffered modes expect +sub-millisecond and **no** re-establishment cost; in `jpeg_on_demand` expect one HTTP +round-trip per call (tens of ms for a modest JPEG, more for 4K). 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). +7.4 **Mode semantics.** In a buffered mode confirm `get_latest_n(5)` and +`get_nearest(ts)` return sensible frames and that `get_nearest` timestamps track wall +clock. In `jpeg_on_demand` confirm both **raise** `RuntimeError` β€” that is the designed +behaviour, not a fault. -**Gate:** stream establishes once, grabs are cheap, decode works, reconnect is fail-soft. +7.5 **Resilience.** Pull the network briefly; confirm `stream_stats()` shows `reconnects` +incrementing, `healthy` recovering, and **DAQ never blocks** (last good frames still +served). In `jpeg_poll` a failed poll is logged and retried with backoff. + +7.6 **Bandwidth check (the reason `jpeg_poll` is the default).** With the stream running, +measure actual traffic to the camera (e.g. `iftop`/`nload` on the device-server host, or +switch port counters). *Confirm* `jpeg_poll` sits at roughly `poll_rate Γ— JPEG size` and +that `mjpeg`, if you try it, is the order-of-magnitude heavier continuous load the +default exists to avoid. Record both numbers β€” they are the evidence for the deployed +mode choice. + +**Gate:** the still endpoint serves real JPEGs, the configured mode establishes and grabs +cheaply, decode works, reconnect is fail-soft, and measured bandwidth is acceptable to the +network owner. ### Stage 8 β€” Full BEC integration diff --git a/csaxs_bec/devices/canon/README.md b/csaxs_bec/devices/canon/README.md index b548ff19..4ea44c31 100644 --- a/csaxs_bec/devices/canon/README.md +++ b/csaxs_bec/devices/canon/README.md @@ -24,8 +24,10 @@ devices/canon/ β”œβ”€β”€ 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) +β”‚ # sources: MJPEGHttpSource, SingleJpegPollSource, Iterable… β”œβ”€β”€ positioner.py # CanonAxis β€” transport-backed ophyd positioner -└── canon_ptz.py # CanonPTZCamera β€” the BEC ophyd device +β”œβ”€β”€ canon_ptz.py # CanonPTZCamera β€” the BEC ophyd device +└── probe.py # operator tool: read-only hardware first-contact probe ``` Key boundaries: @@ -50,6 +52,35 @@ top of `transport/xc.py`. Each carries a `# SPEC:` marker β€” bringing the devic 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. +## Bringing it up against hardware + +Start with the read-only first-contact probe, then follow the gated regime in +[`COMMISSIONING.md`](./COMMISSIONING.md) (Stage 0.5 onward): + +```bash +python -m csaxs_bec.devices.canon.probe --host --user --password +``` + +It reports TCP reachability, what HTTP server answers, which candidate CGI prefixes and +endpoints exist (a `401` confirms a path even without a login), whether anything serves +real JPEG bytes, whether a session opens under the **discovered** prefix, and whether +`XCTransport.connect()` works as shipped. GETs only β€” it never sends a control or motion +command. + +**It is stdlib-only and Python 3.7+**, so on a beamline machine you can copy the single +file over instead of tunnelling β€” no BEC, no venv, no checkout needed: + +```bash +scp csaxs_bec/devices/canon/probe.py @:/tmp/ +ssh @ 'CANON_PROBE_USER=u CANON_PROBE_PASSWORD=p python3 /tmp/probe.py --host ' +``` + +Pass credentials via those environment variables rather than `--user/--password` on shared +machines β€” command-line arguments are visible to other users through `ps`. + +To learn what a PASS looks like before hardware day, rehearse against the mock camera in +`tests/tests_devices/test_canon_probe.py` (run that file directly; see COMMISSIONING 0.4). + ## Connection & credentials Credentials are **never** stored in the YAML config / Redis. The device reads them from @@ -94,7 +125,8 @@ ptz_cam_sim: `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`). +`buffer_maxlen`, `stream_start` (`off` | `connect` | `stage`), `acquisition_mode` +(`mjpeg` | `jpeg_poll` | `jpeg_on_demand`), `poll_rate`, `still_url`, `stream_url`. ## Supported commands @@ -127,27 +159,59 @@ 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. +A persistent worker 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. + +### Acquisition modes β€” choosing the network footprint + +`StreamWorker` only calls `source.open()` and iterates `source.frames()`, so *how* frames +arrive is a `FrameSource` choice, not a redesign. Set it with `acquisition_mode` in +`deviceConfig`: + +| Mode | Idle bandwidth | Idle CPU | Grab latency | Grab API | Establish cost | +|------|----------------|----------|--------------|----------|----------------| +| `jpeg_poll` **(default)** | low (one JPEG Γ— `poll_rate`) | ~0 | sub-ms (≀ 1 poll old) | full | none/cheap | +| `mjpeg` | high (continuous 4K) | ~0 | sub-ms | full | ~2 s, once | +| `jpeg_on_demand` | none | none | one HTTP round-trip per grab | `get_latest` only | none | + +* **`jpeg_poll`** polls the camera's single-JPEG endpoint at `poll_rate` Hz (default 2). + It is the default because it keeps the entire buffer-and-grab benefit β€” O(1) grabs, + `get_nearest` timestamp correlation, fail-soft reconnect β€” at a small fraction of MJPEG + bandwidth. The only cost is temporal resolution: the newest frame may be up to one poll + interval old. +* **`mjpeg`** holds one connection open for the whole session, paying the ~2 s + establishment up front (never inside a scan). Choose it deliberately, when genuine + high-rate/low-latency capture is needed β€” a 4K/30 fps camera then streams continuously + whether or not anyone grabs. +* **`jpeg_on_demand`** starts no worker and buffers nothing ahead: each `get_latest()` + performs one synchronous still fetch through the already-authenticated control session + (`transport.get_still_jpeg()`). Zero idle footprint, but there is no history β€” so + `get_latest_n()` and `get_nearest()` **raise** rather than silently answer a timestamp + query with a stale frame, which would mis-correlate images with scan points. + +`start_stream()` is still the entry point in every mode; in `jpeg_on_demand` it only +prepares the buffer, and a bare `get_latest()` works without it. ```python -cam.start_stream() # open the stream + start the worker (idempotent) +cam.start_stream() # start acquisition for the configured mode (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 +frames = cam.get_latest_n(5) # last 5, oldest-first (buffered modes only) +frame = cam.get_nearest(ts) # closest to a wall-clock timestamp (buffered only) img = cam.get_latest(decode=True) # decode only the grabbed frame (uses cv2) -cam.stream_stats() # ring-buffer + worker instrumentation (latency, health) +cam.stream_stats() # mode + ring-buffer + worker instrumentation cam.stop_stream() ``` -Set `stream_start: connect` (or `stage`) in `deviceConfig` to start the stream +Set `stream_start: connect` (or `stage`) in `deviceConfig` to start acquisition 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. +and spends no CPU decoding frames nobody grabs; decoding happens only on grab. + +Frame *delivery into a scan* is the remaining Phase-2 work. The intent is the +area-detector convention β€” file-writer references via `SUB_FILE_EVENT` for full frames, +`PreviewSignal` for live preview. Only the `preview` component is declared today; nothing +writes to it yet and no file event is emitted, so frames currently reach users through the +grab API alone. > 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 @@ -161,8 +225,9 @@ OPHYD_CONTROL_LAYER=dummy pytest tests/tests_devices/test_canon_*.py --random-or ``` 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). +HTTP/session logic via a mocked opener, incl. single-frame capture), the ring buffer +(incl. thread-safety), the stream worker and all three frame sources (fail-soft +reconnect, poll pacing/interruptibility, no thread leak), and the ophyd device (motion +status, limits, parameter read-back verification, focus-mode gating, all three +acquisition modes, credential handling). ``` diff --git a/csaxs_bec/devices/canon/acquisition/__init__.py b/csaxs_bec/devices/canon/acquisition/__init__.py index c9e71b91..d1f00a1d 100644 --- a/csaxs_bec/devices/canon/acquisition/__init__.py +++ b/csaxs_bec/devices/canon/acquisition/__init__.py @@ -8,7 +8,13 @@ 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 +from .stream_worker import ( + FrameSource, + IterableFrameSource, + MJPEGHttpSource, + SingleJpegPollSource, + StreamWorker, +) __all__ = [ "Frame", @@ -16,5 +22,6 @@ __all__ = [ "FrameSource", "IterableFrameSource", "MJPEGHttpSource", + "SingleJpegPollSource", "StreamWorker", ] diff --git a/csaxs_bec/devices/canon/acquisition/stream_worker.py b/csaxs_bec/devices/canon/acquisition/stream_worker.py index d6728de9..287d20f5 100644 --- a/csaxs_bec/devices/canon/acquisition/stream_worker.py +++ b/csaxs_bec/devices/canon/acquisition/stream_worker.py @@ -12,6 +12,23 @@ Design goals (from the integration brief): * Establishment time and per-frame intervals are measured so we can confirm the ~2 s cost is paid once and grabs are cheap. +Frame sources -- the network-footprint seam +------------------------------------------- +:class:`StreamWorker` only calls ``source.open()`` and iterates ``source.frames()``; +it does not care how the bytes arrive. That makes the *bandwidth profile* a choice of +:class:`FrameSource`, not a redesign: + +* :class:`MJPEGHttpSource` -- holds one HTTP connection open and reads continuously. + Lowest latency, but continuous bandwidth (a 4K/30 fps camera held open all session). +* :class:`SingleJpegPollSource` -- fetches one still JPEG per poll interval. Bandwidth + is one JPEG times a few Hz instead of 30 fps, while the ring buffer and the whole + O(1) grab API (``get_latest`` / ``get_latest_n`` / ``get_nearest``) keep working; + the only cost is that the newest frame may be up to one poll interval old. +* :class:`IterableFrameSource` -- synthetic/offline feed for simulation and tests. + +A third, *worker-less* mode ("fetch now") lives in the ophyd device rather than here: +it has no source and no buffer-ahead, so there is nothing to keep alive. + This module is pure standard library and imports neither ophyd nor the control transport: acquisition is never a dependency of control. @@ -27,6 +44,7 @@ from __future__ import annotations import abc import threading import time +import urllib.request from typing import Callable, Iterable, Iterator try: # pragma: no cover - logging is best-effort @@ -45,6 +63,17 @@ _JPEG_SOI = b"\xff\xd8" _JPEG_EOI = b"\xff\xd9" +def _build_opener(url: str, auth: tuple[str, str] | None) -> urllib.request.OpenerDirector: + """Build a urllib opener, adding HTTP Digest/Basic auth handlers when given.""" + handlers: list[urllib.request.BaseHandler] = [] + if auth is not None: + mgr = urllib.request.HTTPPasswordMgrWithDefaultRealm() + mgr.add_password(None, url, auth[0], auth[1]) + handlers.append(urllib.request.HTTPDigestAuthHandler(mgr)) + handlers.append(urllib.request.HTTPBasicAuthHandler(mgr)) + return urllib.request.build_opener(*handlers) + + class FrameSource(abc.ABC): """A source of compressed frames. ``open()`` then iterate ``frames()``.""" @@ -127,16 +156,7 @@ class MJPEGHttpSource(FrameSource): 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._response = _build_opener(self.url, self._auth).open(self.url, timeout=timeout) self._buffer = b"" def frames(self) -> Iterator[bytes]: @@ -166,6 +186,102 @@ class MJPEGHttpSource(FrameSource): self._buffer = b"" +class SingleJpegPollSource(FrameSource): + """Poll a *single-JPEG* (still-image) endpoint at a fixed rate. + + The network-friendly alternative to :class:`MJPEGHttpSource`: instead of holding a + continuous video stream open, this fetches one complete JPEG per poll interval. + Bandwidth becomes ``one JPEG x rate`` rather than the camera's full frame rate, + idle CPU stays ~0, and -- because the frames still land in the ring buffer -- the + whole grab API (``get_latest`` / ``get_latest_n`` / ``get_nearest``) keeps working + at O(1). The trade-off is temporal resolution: the newest buffered frame may be up + to one poll interval old. + + Frames are fetched either over HTTP from *url* (stdlib ``urllib``, optional + Digest/Basic auth) or through an injected *fetch* callable -- the latter lets the + ophyd device route stills through an already-authenticated control session, and + lets tests run without a network. + + Parameters + ---------- + url: + Still-image endpoint. Required unless *fetch* is given. + rate: + Poll rate in Hz (> 0). ``2.0`` means one JPEG every 500 ms. + auth: + ``(username, password)`` for HTTP auth on *url*. + timeout: + Per-request timeout in seconds. + fetch: + Zero-arg callable returning one JPEG as ``bytes``; overrides *url*. + + Notes + ----- + Each HTTP poll is a fresh request (``urllib`` does not pool connections), so at + high rates the per-request TCP/auth setup starts to dominate -- this source is + aimed at the few-Hz regime it was designed for, not at replacing MJPEG at 30 fps. + A failing fetch propagates to :class:`StreamWorker`, which logs it and retries with + backoff; the last good frames keep being served meanwhile. + """ + + def __init__( + self, + url: str | None = None, + *, + rate: float = 2.0, + auth: tuple[str, str] | None = None, + timeout: float = 5.0, + fetch: Callable[[], bytes] | None = None, + ) -> None: + if rate <= 0: + raise ValueError(f"poll rate must be > 0 Hz, got {rate}") + if url is None and fetch is None: + raise ValueError("SingleJpegPollSource requires either a url or a fetch callable") + self.url = url + self.rate = rate + self.timeout = timeout + self._auth = auth + self._fetch = fetch + self._opener: urllib.request.OpenerDirector | None = None + # Doubles as the pacing timer, so close() interrupts a pending sleep at once. + self._closed = threading.Event() + + @property + def interval(self) -> float: + """Seconds between polls.""" + return 1.0 / self.rate + + def open(self, *, timeout: float | None = None) -> None: + """Prepare the poller. Cheap: no connection is held open. + + The *timeout* argument (the worker's stream-establishment timeout) is + deliberately ignored -- there is no establishment step to bound, and it must + not be confused with the per-request ``timeout``. + """ + self._closed.clear() + if self._fetch is None: + self._opener = _build_opener(self.url, self._auth) + + def _http_get(self) -> bytes: + if self._opener is None: + raise RuntimeError("SingleJpegPollSource.open() must be called before frames()") + with self._opener.open(self.url, timeout=self.timeout) as resp: + return resp.read() + + def frames(self) -> Iterator[bytes]: + fetch = self._fetch if self._fetch is not None else self._http_get + while not self._closed.is_set(): + data = fetch() + if data: + yield data + # Interruptible pacing: stop() must not wait out a slow poll interval. + self._closed.wait(timeout=self.interval) + + def close(self) -> None: + self._closed.set() + self._opener = None + + class StreamWorker: """Owns a background thread that keeps a :class:`FrameSource` feeding a buffer. diff --git a/csaxs_bec/devices/canon/canon_ptz.py b/csaxs_bec/devices/canon/canon_ptz.py index 03c365ce..7cce0e69 100644 --- a/csaxs_bec/devices/canon/canon_ptz.py +++ b/csaxs_bec/devices/canon/canon_ptz.py @@ -1,10 +1,13 @@ """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. +motion -- working with the imaging/streaming stack absent. Phase 2 adds the +buffer-and-grab path: an acquisition worker fills a ring buffer that DAQ grabs from on +demand, in one of three network-footprint modes (see :data:`ACQUISITION_MODES`). The +acquisition subsystem is imported lazily and guarded, so control never depends on it. +Still outstanding in Phase 2: frame *delivery into a scan* (file-writer references via +``SUB_FILE_EVENT``, and driving the ``preview`` signal) -- the component is declared but +nothing writes to it yet. Architecture ------------ @@ -12,7 +15,7 @@ Architecture ``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. +* ring buffer + :class:`StreamWorker` (+ ``preview``) -- the buffer-and-grab path. Credentials are read from environment variables (never from the YAML/Redis config): ``CANON__USER`` / ``CANON__PASSWORD`` by default (```` = the device @@ -36,7 +39,7 @@ 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 +from .transport.xc import DEFAULT_CGI_PREFIX, DEFAULT_STILL_CGI, DEFAULT_STREAM_CGI, XCTransport if TYPE_CHECKING: # pragma: no cover from bec_lib.devicemanager import ScanInfo @@ -55,6 +58,25 @@ except Exception: # pragma: no cover _FOCUS_MODE_PARAM = "focus_mode" _PRESET_STATE_PARAM = "preset_state" +# Acquisition modes -- how frames reach the ring buffer, and at what network cost. +# +# ``mjpeg`` hold one continuous MJPEG connection open for the whole session. +# Lowest latency and full frame rate, but continuous bandwidth -- +# a 4K/30 fps camera streams all session whether or not anyone +# grabs. Choose deliberately, for genuine high-rate needs. +# ``jpeg_poll`` fetch one still JPEG every ``1/poll_rate`` seconds. Bandwidth is +# a small fraction of MJPEG, idle CPU ~0, and the full grab API +# still works at O(1); the newest frame may be up to one poll +# interval old. This is the default: it is the network-friendly +# option that keeps buffer-and-grab intact. +# ``jpeg_on_demand`` no worker and no buffer-ahead -- ``get_latest()`` performs one +# synchronous fetch through the control session. Zero idle +# bandwidth and CPU, but each grab pays an HTTP round-trip and +# there is no history, so ``get_latest_n`` / ``get_nearest`` are +# unavailable (they would silently return stale or absent data). +ACQUISITION_MODES = ("mjpeg", "jpeg_poll", "jpeg_on_demand") +_ON_DEMAND = "jpeg_on_demand" + class CanonParameterSignal(Signal): """Read-only signal proxying a single camera parameter through the transport.""" @@ -169,6 +191,9 @@ class CanonPTZCamera(PSIDeviceBase): buffer_maxlen: int = 64, stream_url: str | None = None, stream_start: str = "off", + acquisition_mode: str = "jpeg_poll", + poll_rate: float = 2.0, + still_url: str | None = None, pan_speed: float | None = None, tilt_speed: float | None = None, scan_info: "ScanInfo | None" = None, @@ -186,13 +211,27 @@ class CanonPTZCamera(PSIDeviceBase): 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_url: MJPEG stream URL (``mjpeg`` mode). Derived if omitted. stream_start: one of ``off`` | ``connect`` | ``stage`` -- when to start the persistent acquisition worker. + acquisition_mode: ``mjpeg`` | ``jpeg_poll`` | ``jpeg_on_demand`` -- how + frames are acquired, and at what network cost. See + :data:`ACQUISITION_MODES` above for the trade-offs. Defaults to + ``jpeg_poll``, the network-friendly mode that keeps the full grab API. + poll_rate: poll rate in Hz for ``jpeg_poll`` (default 2 Hz). + still_url: single-JPEG endpoint for ``jpeg_poll``. Derived if omitted; + unused by ``jpeg_on_demand``, which goes through the control session. 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 acquisition_mode not in ACQUISITION_MODES: + raise ValueError( + f"acquisition_mode must be one of {'|'.join(ACQUISITION_MODES)}, " + f"got {acquisition_mode!r}" + ) + if poll_rate <= 0: + raise ValueError(f"poll_rate must be > 0 Hz, got {poll_rate}") if not simulation and not host: raise ValueError("host is required unless simulation=True") @@ -200,6 +239,9 @@ class CanonPTZCamera(PSIDeviceBase): self._stream_url = stream_url self._stream_start = stream_start self._buffer_maxlen = buffer_maxlen + self._acquisition_mode = acquisition_mode + self._poll_rate = poll_rate + self._still_url = still_url # Shared lock serialising all transport access across axes/signals. import threading @@ -360,46 +402,100 @@ class CanonPTZCamera(PSIDeviceBase): # -- 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. + def _image_url(self, cgi: str) -> str: + # SPEC: the image endpoint URL form is defined by BPE-7216-005; this is a + # reasonable default to verify against the real camera (risk item R10). 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" + return f"http://{host}:{port}/{prefix}/{cgi}" + + def _default_stream_url(self) -> str: + """MJPEG (continuous) stream URL used by ``acquisition_mode: mjpeg``.""" + return self._image_url(DEFAULT_STREAM_CGI) + + def _default_still_url(self) -> str: + """Single-JPEG URL polled by ``acquisition_mode: jpeg_poll``.""" + return self._image_url(DEFAULT_STILL_CGI) + + def _ensure_buffer(self): + """Create the ring buffer on first use; ``None`` if acquisition is absent. + + The acquisition subsystem is imported lazily and guarded, so control keeps + working when the imaging stack is unavailable. + """ + if self._buffer is not None: + return self._buffer + try: + from .acquisition import FrameRingBuffer # pylint: disable=import-outside-toplevel + except Exception as exc: # pylint: disable=broad-except + logger.warning(f"{self.name}: acquisition subsystem unavailable: {exc}") + return None + self._buffer = FrameRingBuffer(self._buffer_maxlen, source_id=self.name) + return self._buffer + + def _make_source(self): + """Build the :class:`FrameSource` for the configured acquisition mode.""" + from .acquisition import ( # pylint: disable=import-outside-toplevel + IterableFrameSource, + MJPEGHttpSource, + SingleJpegPollSource, + ) + + if self._acquisition_mode == "jpeg_poll": + if self._simulation: + # Route the synthetic stills through the fake transport so the poll + # path itself -- not just the buffer -- is exercised offline. + return SingleJpegPollSource( + fetch=self.transport.get_still_jpeg, rate=self._poll_rate + ) + return SingleJpegPollSource( + self._still_url or self._default_still_url(), + rate=self._poll_rate, + auth=getattr(self.transport, "_auth", None), + ) + # mjpeg + if self._simulation: + return IterableFrameSource(self._synthetic_jpeg, delay=0.05) + return MJPEGHttpSource( + self._stream_url or self._default_stream_url(), + auth=getattr(self.transport, "_auth", None), + ) def start_stream(self) -> None: - """Start the persistent acquisition worker + ring buffer (idempotent). + """Start the acquisition worker + ring buffer (idempotent). - Imports the acquisition subsystem lazily and guarded: if it is unavailable, - control still works and a warning is logged. + In ``jpeg_on_demand`` mode there is nothing to keep alive: this only prepares + the buffer, and frames are fetched at grab time instead. """ + if self._acquisition_mode == _ON_DEMAND: + if self._ensure_buffer() is not None: + logger.info( + f"{self.name}: acquisition_mode={_ON_DEMAND}; no worker started " + "(each grab fetches one still)" + ) + return + if self._worker is not None and self._worker.running: return + buffer = self._ensure_buffer() + if buffer is None: + return try: - from .acquisition import ( # pylint: disable=import-outside-toplevel - FrameRingBuffer, - IterableFrameSource, - MJPEGHttpSource, - StreamWorker, - ) + from .acquisition import StreamWorker # pylint: disable=import-outside-toplevel + + source = self._make_source() 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 = StreamWorker(source, buffer, name=f"{self.name}-stream") self._worker.start() - logger.info(f"{self.name}: stream worker started") + logger.info( + f"{self.name}: stream worker started (mode={self._acquisition_mode}" + + (f", {self._poll_rate} Hz" if self._acquisition_mode == "jpeg_poll" else "") + + ")" + ) def stop_stream(self) -> None: """Stop the acquisition worker cleanly (idempotent).""" @@ -421,24 +517,65 @@ class CanonPTZCamera(PSIDeviceBase): ) return self._buffer + def _require_history(self, call: str) -> None: + """Reject history-based grabs in ``jpeg_on_demand`` mode. + + That mode keeps no buffer-ahead, so there is no frame history to search. It + fails loudly rather than returning whatever happened to be grabbed earlier: + silently answering a timestamp query with a stale frame would mis-correlate + images with scan points, which is worse than an error. + """ + if self._acquisition_mode == _ON_DEMAND: + raise RuntimeError( + f"{self.name}: {call} needs buffered frame history, which " + f"acquisition_mode={_ON_DEMAND!r} does not keep (it fetches one frame " + "per grab). Use get_latest(), or configure acquisition_mode " + "'jpeg_poll' or 'mjpeg'." + ) + + def _fetch_still_frame(self): + """Fetch one still through the control session and buffer it (on-demand mode).""" + buffer = self._ensure_buffer() + if buffer is None: + raise RuntimeError( + f"{self.name}: acquisition subsystem unavailable; cannot grab frames" + ) + with self._transport_lock: + data = self.transport.get_still_jpeg() + return buffer.put(data) + def get_latest(self, decode: bool = False): - """Return the most recent buffered frame (``Frame`` or decoded array).""" - frame = self._require_buffer().get_latest() + """Return the most recent frame (``Frame`` or decoded array). + + In ``jpeg_on_demand`` mode this performs one synchronous fetch (an HTTP + round-trip); in the buffered modes it is an O(1) read of the newest frame. + """ + if self._acquisition_mode == _ON_DEMAND: + frame = self._fetch_still_frame() + else: + 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.""" + self._require_history("get_latest_n()") 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*.""" + self._require_history("get_nearest()") 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} + """Return acquisition mode + ring-buffer + worker instrumentation.""" + stats: dict = { + "mode": self._acquisition_mode, + "buffer": self._buffer.stats() if self._buffer else None, + } + if self._acquisition_mode == "jpeg_poll": + stats["poll_rate_hz"] = self._poll_rate if self._worker is not None: stats["worker"] = self._worker.stats() return stats diff --git a/csaxs_bec/devices/canon/probe.py b/csaxs_bec/devices/canon/probe.py new file mode 100644 index 00000000..b6afc0cd --- /dev/null +++ b/csaxs_bec/devices/canon/probe.py @@ -0,0 +1,424 @@ +"""First-contact probe for a Canon CR-N300 -- run this before anything else. + +The device's control protocol is implemented against assumptions (the ``# SPEC:`` +tables in :mod:`~csaxs_bec.devices.canon.transport.xc`, risk items R1-R10 in +COMMISSIONING.md) taken from Canon doc BPE-7216-005, which is not bundled here. This +script replaces "guess a table entry, retry against hardware" with a single sweep that +reports **what the camera actually answers**. + +Usage:: + + python -m csaxs_bec.devices.canon.probe --host 192.168.0.100 + python -m csaxs_bec.devices.canon.probe --host cam.psi.ch --user admin --password s3cret + python -m csaxs_bec.devices.canon.probe --host 192.168.0.100 --json report.json + +**Runs standalone.** Stdlib only, Python 3.7+, so on a beamline machine you can simply +copy this one file over and run it -- no BEC, no ophyd, no venv, no checkout:: + + scp probe.py user@beamline-host:/tmp/ + ssh user@beamline-host 'python3 /tmp/probe.py --host ' + +Step 6 (which exercises the shipped ``XCTransport``) is skipped in that mode and says +so; step 5 answers the same question using only stdlib, and against the prefix the sweep +actually discovered rather than the hardcoded default. + +**On a shared machine, pass credentials via the environment, not argv** -- anything on +the command line is visible to other users via ``ps``:: + + export CANON_PROBE_USER=admin CANON_PROBE_PASSWORD='s3cret' + python3 /tmp/probe.py --host + +It is deliberately **narrow and read-only**: it talks to exactly the one host you name, +issues GETs only, and never sends a control or motion command. It does not scan the +subnet -- if you do not know the camera's address yet, see Stage 0.5.1 of +COMMISSIONING.md (the camera's own menu/OSD and ``arp -a`` are the fast paths). + +Standard library only, so it runs anywhere the device server does -- no BEC session, no +ophyd, no venv extras required. + +Reading the output +------------------ +The status code matters more than success/failure: + +* **200** -- endpoint exists and answered. Note what it returned. +* **401** -- endpoint **exists** but needs credentials. This is a *positive* result for + path discovery: it confirms the prefix/endpoint name even before you have a login. +* **403** -- exists; refused (often "another client holds control"). +* **404** -- wrong path. This is what eliminates a candidate prefix/endpoint. +* **connection refused / timeout** -- wrong host, port, or the service is off. +""" + +from __future__ import annotations + +import argparse +import base64 +import json +import socket +import sys +import urllib.error +import urllib.parse +import urllib.request + +# Candidate CGI path prefixes, best guess first. "-wvhttp-01-" is the Canon +# WebView/Livescope convention the XC protocol derives from (risk item R1). +CANDIDATE_PREFIXES = ["-wvhttp-01-", "", "cgi-bin", "-wvdoc-01-", "webview"] + +# Candidate endpoint names (risk item R2), plus the image endpoints (R10). Grouped so +# the report shows which *role* each hit could fill. +CANDIDATE_ENDPOINTS = { + "session": ["open.cgi", "claim.cgi", "yield.cgi", "close.cgi"], + "query": ["info.cgi", "getparam.cgi", "camera.cgi"], + "control": ["control.cgi", "nphControlCamera", "preset.cgi"], + "image": ["image.cgi", "nphMotionJpeg", "GetLiveImage", "getstill.cgi", "snapshot.cgi"], +} + +# Pages worth fetching to identify the device and find its own web client. +IDENTITY_PATHS = ["/", "/index.html", "/viewer/", "/admin/"] + +_JPEG_SOI = b"\xff\xd8" +_TIMEOUT = 5.0 + + +class Probe: + """Issues the read-only requests and accumulates a structured report.""" + + def __init__(self, host: str, port: int, scheme: str, auth: tuple[str, str] | None): + self.host = host + self.port = port + self.scheme = scheme + self.auth = auth + self.report: dict = { + "host": host, + "port": port, + "scheme": scheme, + "authenticated": auth is not None, + "steps": {}, + } + + # -- plumbing ---------------------------------------------------------- + + def _opener(self, url: str) -> urllib.request.OpenerDirector: + handlers: list[urllib.request.BaseHandler] = [] + if self.auth is not None: + mgr = urllib.request.HTTPPasswordMgrWithDefaultRealm() + mgr.add_password(None, url, self.auth[0], self.auth[1]) + handlers.append(urllib.request.HTTPDigestAuthHandler(mgr)) + handlers.append(urllib.request.HTTPBasicAuthHandler(mgr)) + return urllib.request.build_opener(*handlers) + + def get(self, path: str, *, preemptive_basic: bool = False) -> dict: + """GET *path*; return a dict describing the outcome (never raises).""" + url = f"{self.scheme}://{self.host}:{self.port}{path}" + req = urllib.request.Request(url) + if preemptive_basic and self.auth is not None: + token = base64.b64encode(f"{self.auth[0]}:{self.auth[1]}".encode()).decode() + req.add_header("Authorization", f"Basic {token}") + try: + with self._opener(url).open(req, timeout=_TIMEOUT) as resp: + body = resp.read(4096) + is_jpeg = body.startswith(_JPEG_SOI) + return { + "url": url, + "status": resp.status, + "content_type": resp.headers.get("content-type", ""), + "server": resp.headers.get("server", ""), + "set_cookie": resp.headers.get("set-cookie", ""), + "bytes": len(body), + "head": _describe_body(body), + "is_jpeg": is_jpeg, + # Kept for session-id parsing; never for image payloads. + "text": "" if is_jpeg else body.decode("utf-8", errors="replace")[:2048], + } + except urllib.error.HTTPError as exc: + body = b"" + try: + body = exc.read(2048) + except Exception: # pylint: disable=broad-except + pass + return { + "url": url, + "status": exc.code, + "content_type": (exc.headers or {}).get("content-type", ""), + "server": (exc.headers or {}).get("server", ""), + "auth_header": (exc.headers or {}).get("www-authenticate", ""), + "bytes": len(body), + "head": _describe_body(body), + "is_jpeg": False, + } + except Exception as exc: # pylint: disable=broad-except + return {"url": url, "status": None, "error": f"{type(exc).__name__}: {exc}"} + + # -- steps ------------------------------------------------------------- + + def step_tcp(self) -> bool: + """Is anything listening at all?""" + print(f"\n[1] TCP reachability -- {self.host}:{self.port}") + try: + with socket.create_connection((self.host, self.port), timeout=_TIMEOUT): + pass + except Exception as exc: # pylint: disable=broad-except + print(f" FAIL {type(exc).__name__}: {exc}") + print(" -> Wrong address, wrong port, camera off, or firewall/VLAN.") + print(" Check the camera's own menu/OSD for its IP; try `arp -a`.") + self.report["steps"]["tcp"] = {"ok": False, "error": str(exc)} + return False + print(" OK port is open") + self.report["steps"]["tcp"] = {"ok": True} + return True + + def step_identity(self) -> None: + """What is it, and does it serve its own web client?""" + print(f"\n[2] HTTP identity -- what answers on {self.scheme}://{self.host}:{self.port}/") + results = {} + for path in IDENTITY_PATHS: + res = self.get(path) + results[path] = res + status = res.get("status") + if status is None: + continue + server = res.get("server") or "?" + note = "" + if status == 401: + note = f" [auth: {res.get('auth_header', '')[:40]}]" + print(f" {status} {path:<14} server={server!r} {res.get('head', '')}{note}") + self.report["steps"]["identity"] = results + auth_schemes = { + r.get("auth_header", "").split()[0].lower() + for r in results.values() + if r.get("auth_header") + } + if auth_schemes: + print(f" -> Camera requests HTTP auth: {', '.join(sorted(auth_schemes))}") + if self.auth is None: + print(" Re-run with --user/--password to get past it.") + + def step_endpoints(self) -> None: + """Sweep candidate prefixes x endpoints; 404 eliminates, 401/200 confirms.""" + print("\n[3] CGI surface sweep -- resolves risk items R1 (prefix) and R2 (endpoints)") + print(" (401 = endpoint EXISTS but needs auth -- that still confirms the path)") + found: dict[str, list[dict]] = {} + for prefix in CANDIDATE_PREFIXES: + hits = [] + for role, names in CANDIDATE_ENDPOINTS.items(): + for name in names: + path = f"/{prefix}/{name}" if prefix else f"/{name}" + res = self.get(path) + status = res.get("status") + if status in (200, 401, 403): + hits.append({"role": role, "name": name, "path": path, **res}) + if hits: + label = f"/{prefix}/" if prefix else "/ (root)" + print(f"\n prefix {label}") + for hit in hits: + flag = " <-- JPEG!" if hit.get("is_jpeg") else "" + print( + f" {hit['status']} {hit['role']:<8} {hit['name']:<16}" + f" {hit.get('content_type', '')[:24]}{flag}" + ) + found[prefix] = hits + if not found: + print(" (no candidate path answered -- see step 4, the web UI is the spec)") + self.report["steps"]["endpoints"] = found + + def step_still(self) -> None: + """Did anything hand us a real JPEG? That de-risks both default modes.""" + print("\n[4] Still image -- the single call that de-risks jpeg_poll + jpeg_on_demand") + jpeg_hits = [ + hit + for hits in self.report["steps"].get("endpoints", {}).values() + for hit in hits + if hit.get("is_jpeg") + ] + if jpeg_hits: + for hit in jpeg_hits: + print(f" OK {hit['path']} returned a JPEG ({hit['bytes']}+ bytes)") + print(" -> Set this as DEFAULT_STILL_CGI (or the still_url config key).") + else: + print(" none no probed path returned JPEG bytes (ff d8 ...).") + print(" -> Open the camera's own web UI in a browser, log in, and watch") + print(" DevTools > Network while the live view runs. The URL the") + print(" camera's OWN client fetches images from is the ground truth") + print(" for R10 -- and the same trace resolves R1/R2/R4/R5 for free.") + self.report["steps"]["still"] = jpeg_hits + + def _discovered_prefixes(self) -> list[str]: + """Prefixes the sweep found a session endpoint under, best first.""" + found = self.report["steps"].get("endpoints", {}) + with_session = [ + prefix + for prefix, hits in found.items() + if any(hit["name"] == "open.cgi" for hit in hits) + ] + return with_session or list(found) or [CANDIDATE_PREFIXES[0]] + + def step_session_builtin(self) -> None: + """Open a session using only stdlib, against the prefixes actually discovered. + + Deliberately independent of the rest of the package so a single copied + ``probe.py`` stays fully useful on a beamline machine, and so this reports on + the *observed* prefix rather than the hardcoded default. + """ + print("\n[5] Session open -- built-in attempt against the discovered prefix(es)") + results = {} + for prefix in self._discovered_prefixes(): + path = f"/{prefix}/open.cgi" if prefix else "/open.cgi" + res = self.get(path) + session_id = _parse_session_id(res.get("text", ""), res.get("set_cookie", "")) + label = f"/{prefix}/" if prefix else "/ (root)" + if session_id: + print(f" OK {label}open.cgi -> session id {session_id!r}") + print(f" -> Set DEFAULT_CGI_PREFIX = {prefix!r} in transport/xc.py (R1).") + elif res.get("status") == 200: + print(f" PART {label}open.cgi answered 200 but no session id found (R6b)") + print(f" body: {res.get('head', '')}") + else: + print(f" no {label}open.cgi -> {res.get('status') or res.get('error')}") + results[prefix] = {"status": res.get("status"), "session_id": session_id} + self.report["steps"]["session_builtin"] = results + + def step_xc_session(self) -> None: + """Try the implemented transport as-is, so we know if the guesses happened to fit.""" + print("\n[6] XCTransport.connect() -- does the shipped implementation work as-is?") + try: + from .transport.xc import XCTransport # pylint: disable=import-outside-toplevel + except ImportError: + try: + from csaxs_bec.devices.canon.transport.xc import ( # noqa: E501 pylint: disable=import-outside-toplevel + XCTransport, + ) + except ImportError: + print(" SKIP running as a standalone file -- the shipped transport is not") + print(" importable here. Step 5 above already answers the question;") + print(" re-run from a checkout to also validate XCTransport itself.") + self.report["steps"]["xc_connect"] = {"ok": None, "reason": "standalone"} + return + user, password = self.auth if self.auth else (None, None) + transport = XCTransport( + host=self.host, port=self.port, username=user, password=password, timeout=_TIMEOUT + ) + try: + transport.connect() + except Exception as exc: # pylint: disable=broad-except + # Error bodies can be a whole HTML error page -- keep the report readable. + detail = " ".join(str(exc).split())[:160] + print(f" FAIL {type(exc).__name__}: {detail}") + print(" -> Expected until the SPEC tables match reality. Use steps 3-4 to") + print(" correct DEFAULT_CGI_PREFIX / _CGI in transport/xc.py, then re-run.") + self.report["steps"]["xc_connect"] = {"ok": False, "error": detail} + return + print(" OK session opened, session id acquired") + print(" -> The protocol guesses fit. Proceed to COMMISSIONING.md Stage 1.3.") + self.report["steps"]["xc_connect"] = {"ok": True} + try: + info = transport.get_camera_info() + print(f" info model={info.model!r} firmware={info.firmware!r}") + self.report["steps"]["camera_info"] = { + "model": info.model, + "firmware": info.firmware, + "serial": info.serial, + } + except Exception as exc: # pylint: disable=broad-except + print(f" WARN get_camera_info() failed: {exc} (risk item R7)") + finally: + transport.close() + + def summarise(self) -> None: + print("\n" + "=" * 72) + steps = self.report["steps"] + builtin = steps.get("session_builtin", {}) + opened = {p: r for p, r in builtin.items() if r.get("session_id")} + if not steps.get("tcp", {}).get("ok"): + print("VERDICT: no TCP connection. Nothing else can proceed -- fix the network.") + elif steps.get("xc_connect", {}).get("ok"): + print("VERDICT: full XC session established. The shipped SPEC tables fit.") + print(" Next: COMMISSIONING.md Stage 1.3 (identity/info parse).") + elif opened: + prefix = next(iter(opened)) + print("VERDICT: a session opened with the built-in attempt, but not through") + print(f" XCTransport. Set DEFAULT_CGI_PREFIX = {prefix!r} (and any other") + print(" names from step 3), then re-run. Next: Stage 0.5.5.") + elif steps.get("endpoints"): + print("VERDICT: the camera is reachable and some CGI paths answered, but no") + print(" session opened. Correct the SPEC tables from step 3, then") + print(" re-run. Next: COMMISSIONING.md Stage 0.5.4.") + else: + print("VERDICT: reachable over HTTP, but no candidate CGI path answered.") + print(" The protocol differs from our assumptions -- harvest it from") + print(" the camera's own web UI. Next: COMMISSIONING.md Stage 0.5.3.") + if steps.get("xc_connect", {}).get("ok") is None: + print(" (step 6 was skipped -- standalone file, not a checkout.)") + print("=" * 72) + + +def _parse_session_id(text: str, set_cookie: str) -> str | None: + """Extract a session id from an ``open.cgi`` response (body, else Set-Cookie). + + Mirrors ``XCTransport.connect()`` -- risk item R6b. Kept self-contained so this + file works when copied on its own. + """ + for line in text.replace("\r", "\n").split("\n"): + key, _, value = line.strip().partition("=") + if key.strip() in ("s", "session", "id") and value.strip(): + return value.strip() + if "=" in set_cookie: + candidate = set_cookie.split(";", 1)[0].split("=", 1)[1].strip() + return candidate or None + return None + + +def _describe_body(body: bytes) -> str: + """One-line, safe description of a response body.""" + if not body: + return "(empty)" + if body.startswith(_JPEG_SOI): + return "JPEG image" + text = body.decode("utf-8", errors="replace") + if "" in text.lower(): + start = text.lower().index("<title>") + 7 + end = text.lower().find("", start) + return f"title={text[start:end].strip()[:40]!r}" + return repr(" ".join(text.split())[:60]) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description="Read-only first-contact probe for a Canon CR-N300.", + epilog="Talks only to the host you name. GETs only -- never sends a control command.", + ) + parser.add_argument("--host", required=True, help="camera IP or hostname") + parser.add_argument("--port", type=int, default=80, help="HTTP port (default 80)") + parser.add_argument("--scheme", default="http", choices=("http", "https")) + parser.add_argument("--user", default=None, help="username (or set CANON_PROBE_USER)") + parser.add_argument("--password", default=None, help="password (or CANON_PROBE_PASSWORD)") + parser.add_argument("--json", default=None, help="write the full report to this file") + args = parser.parse_args(argv) + + import os + + user = args.user or os.environ.get("CANON_PROBE_USER") + password = args.password or os.environ.get("CANON_PROBE_PASSWORD") + auth = (user, password) if user is not None else None + + print("=" * 72) + print(f"Canon CR-N300 first-contact probe -- {args.scheme}://{args.host}:{args.port}") + print(f"credentials: {'supplied' if auth else 'NONE (expect 401s)'}") + print("=" * 72) + + probe = Probe(args.host, args.port, args.scheme, auth) + if probe.step_tcp(): + probe.step_identity() + probe.step_endpoints() + probe.step_still() + probe.step_session_builtin() + probe.step_xc_session() + probe.summarise() + + if args.json: + with open(args.json, "w", encoding="utf-8") as handle: + json.dump(probe.report, handle, indent=2, default=str) + print(f"\nfull report written to {args.json}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/csaxs_bec/devices/canon/transport/base.py b/csaxs_bec/devices/canon/transport/base.py index 4a382e6d..989d61bb 100644 --- a/csaxs_bec/devices/canon/transport/base.py +++ b/csaxs_bec/devices/canon/transport/base.py @@ -206,3 +206,28 @@ class CameraTransport(abc.ABC): @abc.abstractmethod def get_preset_state(self) -> int: """Return the id of the last recalled preset (``-1`` if none/unknown).""" + + # -- single-frame capture ---------------------------------------------- + + def get_still_jpeg(self) -> bytes: + """Fetch one complete JPEG still through the control session. + + This is the *only* imaging-adjacent method on the transport, and it is here + deliberately: it lets a caller grab a single frame over the session it has + already authenticated, with no persistent stream and no dependency on the + acquisition subsystem. It backs the device's ``jpeg_on_demand`` acquisition + mode (zero idle bandwidth -- one HTTP round-trip per grab). + + Optional: not every transport can serve stills (a VISCA-over-IP control + transport, for instance, has no image path), so the default implementation + raises rather than forcing every subclass to implement it. + + Raises: + CanonCommandError: if this transport cannot capture stills. + """ + from .errors import CanonCommandError # local import: keeps the ABC import-light + + raise CanonCommandError( + f"{type(self).__name__} does not support single-frame capture " + "(get_still_jpeg); use a streaming acquisition mode instead" + ) diff --git a/csaxs_bec/devices/canon/transport/fake.py b/csaxs_bec/devices/canon/transport/fake.py index 867a97ac..e55aef10 100644 --- a/csaxs_bec/devices/canon/transport/fake.py +++ b/csaxs_bec/devices/canon/transport/fake.py @@ -70,6 +70,7 @@ class FakeTransport(CameraTransport): self._parameters = dict(_DEFAULT_PARAMETERS) self._focus_mode = FocusMode.MANUAL self._preset_state = -1 + self._still_count = 0 # Motion state. For gradual motion we record where/when a move started. self._positions = dict(_DEFAULT_POSITIONS) @@ -246,3 +247,17 @@ class FakeTransport(CameraTransport): def get_preset_state(self) -> int: self._require_session() return self._preset_state + + # -- single-frame capture ---------------------------------------------- + + def get_still_jpeg(self) -> bytes: + """Return a synthetic JPEG-framed payload (SOI ... EOI). + + Not a decodable image -- it exists so the acquisition path (poll source, + ring buffer, grab API) can be exercised end-to-end without hardware. Each + call returns distinct bytes so tests can tell successive frames apart. + """ + self._record("get_still_jpeg") + self._require_session() + self._still_count += 1 + return b"\xff\xd8" + f"SIMSTILL{self._still_count:06d}".encode("ascii") + b"\xff\xd9" diff --git a/csaxs_bec/devices/canon/transport/xc.py b/csaxs_bec/devices/canon/transport/xc.py index b9414e60..fde8dd5d 100644 --- a/csaxs_bec/devices/canon/transport/xc.py +++ b/csaxs_bec/devices/canon/transport/xc.py @@ -57,6 +57,13 @@ except Exception: # pragma: no cover # Verify against BPE-7216-005 for the CR-N300 XC protocol. DEFAULT_CGI_PREFIX = "-wvhttp-01-" +# SPEC: image endpoints. Canon WebView serves a single JPEG and a server-push MJPEG +# stream; on some firmware these are *different* CGIs (e.g. "image.cgi" vs +# "nphMotionJpeg"). They are kept as two constants so they can be corrected +# independently once BPE-7216-005 is confirmed -- see risk item R10. +DEFAULT_STILL_CGI = "image.cgi" +DEFAULT_STREAM_CGI = "image.cgi" + # SPEC: CGI endpoint names. Adjust to match BPE-7216-005. _CGI = { "open": "open.cgi", @@ -66,6 +73,7 @@ _CGI = { "info": "info.cgi", "control": "control.cgi", "preset": "preset.cgi", + "still": DEFAULT_STILL_CGI, } # SPEC: query-parameter names used by the control/info CGIs. Adjust per spec. @@ -413,3 +421,26 @@ class XCTransport(CameraTransport): def get_preset_state(self) -> int: return self._preset_state + + # -- single-frame capture ---------------------------------------------- + + def get_still_jpeg(self) -> bytes: + """Fetch one JPEG still over the open session (no persistent stream). + + Backs the device's ``jpeg_on_demand`` mode. Requires a session but *not* the + control privilege -- reading an image is not a control operation. + """ + if self._session_id is None: + raise CanonSessionError("cannot capture a still before connecting") + resp = self._request("still") + body = resp.body + # Cheap sanity check: a CGI that answers 200 with an HTML error page would + # otherwise be buffered as an undecodable "frame" and only fail much later, + # at decode time. Fail here instead, pointing at the endpoint (R10). + if not body.startswith(b"\xff\xd8"): + raise CanonCommandError( + f"{_CGI['still']} did not return a JPEG " + f"(got {len(body)} bytes starting {body[:16]!r}); verify the still " + "endpoint against BPE-7216-005" + ) + return body diff --git a/tests/tests_devices/test_canon_device.py b/tests/tests_devices/test_canon_device.py index bc119647..089b2610 100644 --- a/tests/tests_devices/test_canon_device.py +++ b/tests/tests_devices/test_canon_device.py @@ -1,6 +1,7 @@ """Tests for the CanonPTZCamera ophyd device (simulation mode, no hardware).""" import os +import time from unittest import mock import pytest @@ -130,6 +131,117 @@ def test_streaming_in_simulation_fills_buffer(): device.destroy() +# --------------------------------------------------------------------------- # +# Acquisition modes (network-footprint choice behind one grab API) # +# --------------------------------------------------------------------------- # + + +def test_default_acquisition_mode_is_the_network_friendly_one(cam): + assert cam._acquisition_mode == "jpeg_poll" + assert cam.stream_stats()["mode"] == "jpeg_poll" + + +def test_invalid_acquisition_mode_and_poll_rate_rejected(): + with pytest.raises(ValueError): + CanonPTZCamera(name="badmode", simulation=True, acquisition_mode="rtsp") + with pytest.raises(ValueError): + CanonPTZCamera(name="badrate", simulation=True, poll_rate=0) + + +def test_jpeg_poll_mode_fills_buffer_and_keeps_full_grab_api(): + device = CanonPTZCamera( + name="poll_sim", simulation=True, acquisition_mode="jpeg_poll", poll_rate=50.0 + ) + 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") + # History-based grabs stay available in this mode -- that is the whole point. + assert device.get_latest_n(2) + assert device.get_nearest(frame.wall_ts) is not None + stats = device.stream_stats() + assert stats["mode"] == "jpeg_poll" and stats["poll_rate_hz"] == 50.0 + assert stats["worker"]["establishment_time_s"] is not None + finally: + device.stop_stream() + device.destroy() + + +def test_jpeg_poll_uses_the_still_endpoint_not_the_stream(): + device = CanonPTZCamera( + name="poll_url", simulation=False, host="cam.test", acquisition_mode="jpeg_poll" + ) + source = device._make_source() + assert source.url == "http://cam.test:80/-wvhttp-01-/image.cgi" + assert source.rate == 2.0 + + +def test_jpeg_on_demand_starts_no_worker_and_fetches_per_grab(): + device = CanonPTZCamera(name="ondemand", simulation=True, acquisition_mode="jpeg_on_demand") + device.wait_for_connection() + try: + device.start_stream() # must be a no-op beyond preparing the buffer + assert device._worker is None + first = device.get_latest() + second = device.get_latest() + assert first.data.startswith(b"\xff\xd8") + # Each grab is a fresh fetch, not a re-read of one buffered frame. + assert first.data != second.data + assert "get_still_jpeg" in device.transport.call_log + finally: + device.destroy() + + +def test_jpeg_on_demand_needs_no_start_stream(): + """Zero idle footprint: nothing to start, so a bare grab must just work.""" + device = CanonPTZCamera(name="ondemand2", simulation=True, acquisition_mode="jpeg_on_demand") + device.wait_for_connection() + try: + assert device.get_latest() is not None + finally: + device.destroy() + + +def test_jpeg_on_demand_rejects_history_grabs_loudly(): + """No buffer-ahead means no history; answering with a stale frame would be worse.""" + device = CanonPTZCamera(name="ondemand3", simulation=True, acquisition_mode="jpeg_on_demand") + device.wait_for_connection() + try: + device.get_latest() + with pytest.raises(RuntimeError, match="jpeg_on_demand"): + device.get_latest_n(3) + with pytest.raises(RuntimeError, match="jpeg_on_demand"): + device.get_nearest(time.time()) + finally: + device.destroy() + + +def test_mjpeg_mode_still_available_for_high_rate_needs(): + device = CanonPTZCamera(name="mjpeg_sim", simulation=True, acquisition_mode="mjpeg") + device.wait_for_connection() + try: + device.start_stream() + assert device._worker.wait_for_first_frame(timeout=2.0) + assert device.stream_stats()["mode"] == "mjpeg" + assert "poll_rate_hz" not in device.stream_stats() + finally: + device.stop_stream() + device.destroy() + + +def test_mjpeg_mode_uses_the_stream_url_override(): + device = CanonPTZCamera( + name="mjpeg_url", + simulation=False, + host="cam.test", + acquisition_mode="mjpeg", + stream_url="http://cam.test/custom/stream", + ) + assert device._make_source().url == "http://cam.test/custom/stream" + + def test_non_simulation_requires_host(): with pytest.raises(ValueError): CanonPTZCamera(name="nohost", simulation=False) diff --git a/tests/tests_devices/test_canon_probe.py b/tests/tests_devices/test_canon_probe.py new file mode 100644 index 00000000..97c4a6e0 --- /dev/null +++ b/tests/tests_devices/test_canon_probe.py @@ -0,0 +1,296 @@ +"""End-to-end tests for the first-contact probe, against a mock Canon camera. + +The probe is the first thing anyone runs against real hardware, so its *success* +path has to be trustworthy before it is trusted at the beamline: a bug in the +"session established" branch would only surface with the camera in front of you. + +:class:`FakeCameraServer` is a real HTTP server that speaks enough of the assumed XC +protocol to exercise the whole probe. It doubles as a **rehearsal target** -- run this +file directly to serve a fake camera, then point the probe at it, so you know exactly +what a PASS looks like before hardware day: + + python tests/tests_devices/test_canon_probe.py # serves on :8400 + python -m csaxs_bec.devices.canon.probe --host 127.0.0.1 --port 8400 \ + --user operator --password secret +""" + +from __future__ import annotations + +import base64 +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +import pytest + +from csaxs_bec.devices.canon.probe import Probe, main + +USER = "operator" +PASSWORD = "secret" +PREFIX = "-wvhttp-01-" +SESSION_ID = "SESSION123" + +# Minimal JPEG-framed payload: the probe only checks the SOI marker. +FAKE_JPEG = b"\xff\xd8\xff\xe0" + b"\x00" * 64 + b"\xff\xd9" + +# Newline-delimited key=value, the format the transport's _parse_kv expects. +INFO_BODY = ( + "model=CR-N300\n" + "firmware=1.0.3\n" + "serial=FAKE12345\n" + "pan=1200\n" + "tilt=-300\n" + "zoom=100\n" + "focus=500\n" + "focusmode=manual\n" + "gain=0.0\n" + "iris=5.6\n" +) + + +class _Handler(BaseHTTPRequestHandler): + """Serves the subset of the assumed XC surface the probe exercises.""" + + require_auth = True + # When False, only "/" is served and every CGI 404s -- a camera that does not + # speak the protocol we assumed. + serve_cgi = True + prefix = PREFIX + + def log_message(self, *args): # silence the test run + pass + + def _authorised(self) -> bool: + if not self.require_auth: + return True + header = self.headers.get("Authorization", "") + if not header.startswith("Basic "): + return False + try: + decoded = base64.b64decode(header.split(None, 1)[1]).decode() + except Exception: # pylint: disable=broad-except + return False + return decoded == f"{USER}:{PASSWORD}" + + def _send(self, status: int, body: bytes, content_type: str) -> None: + self.send_response(status) + self.send_header("Content-Type", content_type) + self.send_header("Content-Length", str(len(body))) + self.send_header("Server", "Canon-Fake/1.0") + if status == 401: + self.send_header("WWW-Authenticate", 'Basic realm="Canon"') + self.end_headers() + self.wfile.write(body) + + # pylint: disable=invalid-name + def do_GET(self) -> None: + path = self.path.split("?", 1)[0] + + if path == "/": + self._send( + 200, + b"CR-N300 Network Camera", + "text/html", + ) + return + + if not self.serve_cgi or not path.startswith(f"/{self.prefix}/"): + self._send(404, b"not found", "text/plain") + return + + if not self._authorised(): + self._send(401, b"unauthorized", "text/plain") + return + + cgi = path[len(f"/{self.prefix}/") :] + if cgi == "open.cgi": + self._send(200, f"s={SESSION_ID}\n".encode(), "text/plain") + elif cgi in ("claim.cgi", "yield.cgi", "close.cgi"): + self._send(200, b"ok\n", "text/plain") + elif cgi == "info.cgi": + self._send(200, INFO_BODY.encode(), "text/plain") + elif cgi == "image.cgi": + self._send(200, FAKE_JPEG, "image/jpeg") + elif cgi in ("control.cgi", "preset.cgi"): + self._send(200, b"ok\n", "text/plain") + else: + self._send(404, b"unknown cgi", "text/plain") + + +class FakeCameraServer: + """Threaded HTTP server emulating a Canon camera. Use as a context manager.""" + + def __init__( + self, port: int = 0, require_auth: bool = True, serve_cgi: bool = True, prefix: str = PREFIX + ): + handler = type( + "_H", + (_Handler,), + {"require_auth": require_auth, "serve_cgi": serve_cgi, "prefix": prefix}, + ) + self._server = ThreadingHTTPServer(("127.0.0.1", port), handler) + self.port = self._server.server_address[1] + self._thread: threading.Thread | None = None + + def __enter__(self) -> FakeCameraServer: + self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) + self._thread.start() + return self + + def __exit__(self, *exc) -> None: + self._server.shutdown() + self._server.server_close() + if self._thread is not None: + self._thread.join(timeout=5) + + +@pytest.fixture +def camera(): + with FakeCameraServer() as server: + yield server + + +def _probe(server, auth=(USER, PASSWORD)) -> Probe: + probe = Probe("127.0.0.1", server.port, "http", auth) + assert probe.step_tcp() + probe.step_identity() + probe.step_endpoints() + probe.step_still() + probe.step_session_builtin() + probe.step_xc_session() + return probe + + +def test_probe_success_path_reports_a_session(camera): + """The branch that matters on hardware day: everything fits, session opens.""" + report = _probe(camera).report["steps"] + assert report["tcp"]["ok"] + assert report["xc_connect"]["ok"] + assert report["camera_info"]["model"] == "CR-N300" + assert report["camera_info"]["firmware"] == "1.0.3" + + +def test_probe_identifies_the_camera_web_ui(camera): + identity = _probe(camera).report["steps"]["identity"] + assert identity["/"]["status"] == 200 + assert "CR-N300 Network Camera" in identity["/"]["head"] + + +def test_probe_sweep_finds_the_right_prefix_and_endpoints(camera): + endpoints = _probe(camera).report["steps"]["endpoints"] + assert PREFIX in endpoints, "the sweep must identify the real CGI prefix (R1)" + names = {hit["name"] for hit in endpoints[PREFIX]} + assert {"open.cgi", "claim.cgi", "info.cgi", "image.cgi"} <= names + # A wrong prefix must be eliminated, not reported as a hit. + assert "cgi-bin" not in endpoints + + +def test_probe_flags_the_still_endpoint(camera): + """This is the hit that de-risks jpeg_poll and jpeg_on_demand together.""" + still = _probe(camera).report["steps"]["still"] + assert still, "a JPEG-serving endpoint must be flagged" + assert any(hit["name"] == "image.cgi" for hit in still) + + +def test_probe_without_credentials_still_discovers_paths(camera): + """401 confirms an endpoint exists -- path discovery must work before login.""" + probe = Probe("127.0.0.1", camera.port, "http", None) + assert probe.step_tcp() + probe.step_identity() + probe.step_endpoints() + endpoints = probe.report["steps"]["endpoints"] + assert PREFIX in endpoints + assert all(hit["status"] == 401 for hit in endpoints[PREFIX]) + # ... but nothing is mistaken for a real image. + probe.step_still() + assert not probe.report["steps"]["still"] + + +def test_builtin_session_step_works_without_the_package(camera): + """The step that must survive `scp probe.py` to a beamline box: stdlib only.""" + builtin = _probe(camera).report["steps"]["session_builtin"] + assert builtin[PREFIX]["session_id"] == SESSION_ID + + +def test_builtin_session_step_uses_the_discovered_prefix_not_the_default(): + """If the camera serves a prefix we didn't default to, step 5 must still find it.""" + with FakeCameraServer(prefix="cgi-bin") as server: + probe = _probe(server) + builtin = probe.report["steps"]["session_builtin"] + assert builtin["cgi-bin"]["session_id"] == SESSION_ID + # ...and the shipped transport, which uses the hardcoded default, does not. + assert probe.report["steps"]["xc_connect"]["ok"] is False + probe.summarise() + + +def test_session_id_parsing_covers_body_and_set_cookie(): + from csaxs_bec.devices.canon.probe import _parse_session_id + + assert _parse_session_id("s=ABC\n", "") == "ABC" + assert _parse_session_id("session=XY9\n", "") == "XY9" + assert _parse_session_id("id=Q1\n", "") == "Q1" + assert _parse_session_id("", "session=COOKIEVAL; path=/") == "COOKIEVAL" + assert _parse_session_id("model=CR-N300\n", "") is None + + +def test_probe_main_runs_end_to_end_and_writes_json(camera, tmp_path): + out = tmp_path / "report.json" + rc = main( + [ + "--host", + "127.0.0.1", + "--port", + str(camera.port), + "--user", + USER, + "--password", + PASSWORD, + "--json", + str(out), + ] + ) + assert rc == 0 + assert out.exists() + import json + + saved = json.loads(out.read_text()) + assert saved["steps"]["xc_connect"]["ok"] + + +def test_probe_survives_a_camera_that_speaks_nothing_we_expect(): + """The realistic bad case: reachable web UI, but none of our CGI guesses hold. + + This is the outcome that sends you to Stage 0.5.3 (harvest the protocol from the + camera's own web UI), so it must degrade cleanly rather than crash. + """ + with FakeCameraServer(require_auth=False, serve_cgi=False) as server: + probe = Probe("127.0.0.1", server.port, "http", None) + assert probe.step_tcp() + probe.step_identity() + # The web UI is reachable -- that alone is a real day-one result. + assert probe.report["steps"]["identity"]["/"]["status"] == 200 + probe.step_endpoints() + assert probe.report["steps"]["endpoints"] == {}, "no path should be claimed as a hit" + probe.step_still() + assert probe.report["steps"]["still"] == [] + # The XC attempt fails cleanly rather than raising. + probe.step_xc_session() + assert probe.report["steps"]["xc_connect"]["ok"] is False + probe.summarise() + + +if __name__ == "__main__": # rehearsal mode -- serve a fake camera and wait + import time + + with FakeCameraServer(port=8400) as srv: + print(f"Fake Canon camera on http://127.0.0.1:{srv.port} (user={USER} pw={PASSWORD})") + print("Probe it with:") + print( + f" python -m csaxs_bec.devices.canon.probe --host 127.0.0.1 " + f"--port {srv.port} --user {USER} --password {PASSWORD}" + ) + print("Ctrl-C to stop.") + try: + while True: + time.sleep(1) + except KeyboardInterrupt: + pass diff --git a/tests/tests_devices/test_canon_stream_worker.py b/tests/tests_devices/test_canon_stream_worker.py index c7e943df..06a68026 100644 --- a/tests/tests_devices/test_canon_stream_worker.py +++ b/tests/tests_devices/test_canon_stream_worker.py @@ -3,11 +3,14 @@ import threading import time +import pytest + from csaxs_bec.devices.canon.acquisition.ring_buffer import FrameRingBuffer from csaxs_bec.devices.canon.acquisition.stream_worker import ( FrameSource, IterableFrameSource, MJPEGHttpSource, + SingleJpegPollSource, StreamWorker, ) @@ -103,6 +106,97 @@ def test_mjpeg_source_splits_frames_without_decoding(): assert got == [b"\xff\xd8AAA\xff\xd9", b"\xff\xd8BBBB\xff\xd9"] +def test_poll_source_yields_one_frame_per_interval(): + calls = [] + + def fetch(): + calls.append(time.monotonic()) + return b"\xff\xd8" + str(len(calls)).encode() + b"\xff\xd9" + + src = SingleJpegPollSource(fetch=fetch, rate=100.0) + assert src.interval == pytest.approx(0.01) + src.open() + frames = [] + for frame in src.frames(): + frames.append(frame) + if len(frames) == 3: + src.close() + assert frames == [b"\xff\xd81\xff\xd9", b"\xff\xd82\xff\xd9", b"\xff\xd83\xff\xd9"] + + +def test_poll_source_feeds_buffer_through_the_same_worker(): + """The point of the seam: buffer + grab API are unchanged by the source swap.""" + buf = FrameRingBuffer(16) + counter = {"n": 0} + + def fetch(): + counter["n"] += 1 + return f"frame{counter['n']}".encode() + + worker = StreamWorker(SingleJpegPollSource(fetch=fetch, rate=50.0), buf, name="poll") + 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 buf.get_latest() is not None + # Buffered history is intact, so nearest-timestamp lookup still works. + assert buf.get_nearest(buf.get_latest().wall_ts) is buf.get_latest() + + +def test_poll_source_close_interrupts_a_slow_interval(): + """A 30 s poll interval must not delay stop() by 30 s.""" + buf = FrameRingBuffer(4) + worker = StreamWorker( + SingleJpegPollSource(fetch=lambda: b"slow", rate=1 / 30.0), buf, name="slowpoll" + ) + worker.start() + assert worker.wait_for_first_frame(timeout=2.0) + t0 = time.monotonic() + worker.stop(timeout=5.0) + assert time.monotonic() - t0 < 2.0 + assert not worker.running + + +def test_poll_source_rejects_bad_configuration(): + with pytest.raises(ValueError): + SingleJpegPollSource("http://x/image.cgi", rate=0) + with pytest.raises(ValueError): + SingleJpegPollSource() # neither url nor fetch + + +def test_poll_source_http_path_uses_opener_without_holding_a_connection(): + src = SingleJpegPollSource("http://cam/image.cgi", rate=10.0) + src.open() + assert src._opener is not None + + class FakeResp: + def __init__(self, data): + self._data = data + + def read(self): + return self._data + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + opened = [] + + class FakeOpener: + def open(self, url, timeout=None): + opened.append((url, timeout)) + return FakeResp(b"\xff\xd8JPEG\xff\xd9") + + src._opener = FakeOpener() + assert src._http_get() == b"\xff\xd8JPEG\xff\xd9" + assert opened == [("http://cam/image.cgi", 5.0)] + src.close() + assert src._opener is None + + def test_worker_thread_stops_cleanly_no_leak(): buf = FrameRingBuffer(8) before = threading.active_count() diff --git a/tests/tests_devices/test_canon_transport.py b/tests/tests_devices/test_canon_transport.py index 8877e4ff..d809793f 100644 --- a/tests/tests_devices/test_canon_transport.py +++ b/tests/tests_devices/test_canon_transport.py @@ -236,3 +236,68 @@ def test_xc_focus_mode_roundtrip_wire_tokens(): 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 + + +# --------------------------------------------------------------------------- # +# Single-frame capture (backs the jpeg_on_demand acquisition mode) # +# --------------------------------------------------------------------------- # + + +class BinaryOpener(RecordingOpener): + """Opener returning raw bytes for the still endpoint (JPEGs are not text).""" + + def __init__(self, still_body: bytes): + super().__init__(bodies={"open.cgi": "s=S1\n"}) + self.still_body = still_body + + def __call__(self, url, data, timeout, auth): + if "image.cgi" in url: + self.requests.append(url) + return HttpResponse(status=self.status, body=self.still_body, headers={}) + return super().__call__(url, data, timeout, auth) + + +def test_xc_get_still_jpeg_returns_body_and_carries_session(): + jpeg = b"\xff\xd8\xff\xe0payload\xff\xd9" + opener = BinaryOpener(jpeg) + t = make_xc(opener) + t.connect() + # A still is a read, not a control operation: no claim required. + assert t.get_still_jpeg() == jpeg + still_url = [u for u in opener.requests if "image.cgi" in u][-1] + assert "s=S1" in still_url + + +def test_xc_get_still_jpeg_rejects_a_non_jpeg_body(): + """An HTML error page served with HTTP 200 must fail here, not at decode time.""" + opener = BinaryOpener(b"error") + t = make_xc(opener) + t.connect() + with pytest.raises(CanonCommandError): + t.get_still_jpeg() + + +def test_xc_get_still_jpeg_requires_a_session(): + t = make_xc(BinaryOpener(b"\xff\xd8\xff\xd9")) + with pytest.raises(CanonSessionError): + t.get_still_jpeg() + + +def test_fake_get_still_jpeg_is_framed_and_distinct(fake): + first = fake.get_still_jpeg() + second = fake.get_still_jpeg() + assert first.startswith(b"\xff\xd8") and first.endswith(b"\xff\xd9") + assert first != second + + +def test_transport_still_capture_is_optional_not_mandatory(): + """A transport with no image path reports so clearly instead of failing obscurely.""" + from csaxs_bec.devices.canon.transport.base import CameraTransport + + class ControlOnlyTransport(FakeTransport): + get_still_jpeg = CameraTransport.get_still_jpeg + + t = ControlOnlyTransport() + t.connect() + with pytest.raises(CanonCommandError, match="does not support single-frame capture"): + t.get_still_jpeg()