feat(canon): add acquisition modes and a hardware first-contact probe (Phase 2)
Acquisition modes -- selectable network footprint behind one grab API --------------------------------------------------------------------- Adds `acquisition_mode` (deviceConfig): mjpeg | jpeg_poll | jpeg_on_demand, defaulting to jpeg_poll. Answers the standing bandwidth critique of holding 1-3 x 4K@30fps open continuously, without losing buffer-and-grab. - SingleJpegPollSource: polls a still endpoint at `poll_rate` Hz (default 2). Frames land in the same ring buffer, so the full grab API -- including get_nearest() timestamp correlation and fail-soft reconnect -- is unchanged. Pacing uses an interruptible Event.wait(), so a long poll interval cannot delay stop(). - jpeg_on_demand: no worker, no connection held; get_latest() does one synchronous fetch through the control session. get_latest_n()/get_nearest() raise rather than answer a timestamp query from a non-existent history -- a stale frame would mis-correlate images with scan points. - CameraTransport.get_still_jpeg(): concrete default on the ABC that raises, not an abstract method, so a control-only transport (e.g. VISCA-over-IP) need not implement an image path. XCTransport rejects a non-JPEG body up front, so a wrong endpoint fails at the grab, not later at decode. - Splits DEFAULT_STREAM_CGI / DEFAULT_STILL_CGI; extracts a shared urllib auth-handler helper used by both HTTP sources. probe.py -- read-only first-contact tool ---------------------------------------- The CGI prefix, endpoint names and axis encodings are guesses from Canon doc BPE-7216-005 (risk items R1-R10). The probe replaces "guess a table entry, retry against hardware" with one sweep reporting what the camera answers: TCP -> HTTP identity -> candidate prefixes x endpoints -> JPEG-serving endpoint -> session open. A 401 confirms a path even without credentials. Stdlib only, Python 3.7+, GETs only, one host. Runs standalone from a copied file (no BEC/venv/checkout), so it can be run on a beamline machine instead of tunnelling; step 5 opens a session using the *discovered* prefix and names the constant to change, step 6 additionally validates XCTransport when run from a checkout. Docs and tests -------------- - COMMISSIONING: new Stage 0.5 (first contact) covering address/credentials, the probe, harvesting the real protocol from the camera's own web UI via DevTools, and correcting the SPEC tables; per-mode Stage 7 with a bandwidth measurement; Stage 0.2 records the editable-install/worktree trap (and that verifying it from a worktree root silently lies). - README: acquisition-mode table; corrects an earlier overclaim that the SUB_FILE_EVENT/preview delivery hooks were complete -- they are not. - 76 tests (was 47), no hardware: poll source and its interruptibility, still capture incl. non-JPEG rejection, all three modes, and the probe end-to-end against a mock Canon camera that doubles as a rehearsal target. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
menzelandClaude Opus 5 committed
1 parent
699c6b146c
commit
7885b3f1d4
14 files changed
+1756
-117
No files matched your search
@@ -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
|
||||
|
||||
@@ -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 <ip>` 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"]`.
|
||||
<a name="r9"></a>**R9 — Stop command.** `control.cgi?stop=all|<axis>`. *Symptom:* stop
|
||||
doesn't halt motion. *Fix:* `_PARAM["stop"]` and `stop()`.
|
||||
|
||||
<a name="r10"></a>**R10 — Stream URL/format.** Default `…/<prefix>/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`).
|
||||
<a name="r10"></a>**R10 — Image endpoints (stream *and* still).** Two separate SPEC
|
||||
constants in `xc.py`, both defaulting to `image.cgi`; on real firmware they may differ
|
||||
(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 <user>@<beamline-host>:/tmp/
|
||||
ssh <user>@<beamline-host>
|
||||
export CANON_PROBE_USER=<user> CANON_PROBE_PASSWORD='<pw>' # not on argv: `ps` leaks it
|
||||
python3 /tmp/probe.py --host <cam-ip> --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://<ip>/`, 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 <user>:<pw> "http://<ip>/<prefix>/<still-cgi>" -o still.jpg
|
||||
curl -v --basic -u <user>:<pw> "http://<ip>/<prefix>/<still-cgi>" -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 <host>`, 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 <host>`, then
|
||||
`curl -v http://<host>:<port>/`. *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="<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
|
||||
|
||||
|
||||
@@ -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 <ip> --user <user> --password <pw>
|
||||
```
|
||||
|
||||
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 <user>@<beamline-host>:/tmp/
|
||||
ssh <user>@<beamline-host> 'CANON_PROBE_USER=u CANON_PROBE_PASSWORD=p python3 /tmp/probe.py --host <cam-ip>'
|
||||
```
|
||||
|
||||
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).
|
||||
```
|
||||
@@ -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",
|
||||
]
|
||||
@@ -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.
|
||||
|
||||
|
||||
@@ -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_<NAME>_USER`` / ``CANON_<NAME>_PASSWORD`` by default (``<NAME>`` = 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
|
||||
|
||||
@@ -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 <cam-ip>'
|
||||
|
||||
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 <cam-ip>
|
||||
|
||||
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 "<title>" in text.lower():
|
||||
start = text.lower().index("<title>") + 7
|
||||
end = text.lower().find("</title>", 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())
|
||||
@@ -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"
|
||||
)
|
||||
@@ -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"
|
||||
@@ -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
|
||||
@@ -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)
|
||||
|
||||
@@ -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"<html><head><title>CR-N300 Network Camera</title></head><body/></html>",
|
||||
"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
|
||||
@@ -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()
|
||||
|
||||
@@ -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"<html>error</html>")
|
||||
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()
|
||||
Reference in new issue
Block a user