menzelandClaude Opus 4.8
2cbe3737ce
CI for csaxs_bec / test (push) Successful in 1m31s
Custom (non-EPICS) ophyd device for the Canon CR-N300 PTZ remote camera. Phase 1 (implemented): standalone control that works with the imaging stack absent. PTZ/zoom/focus exposed as positioners (move()/set() -> status, stop(), spec limits: pan +/-170 deg, tilt -30..+100 deg, 20x zoom); camera parameters (exposure/iris, gain, white balance, focus mode, presets) via read()/describe() with read-back verification; Manual + Continuous/Face/Tracking AF focus modes. Control plane (transport/, pure stdlib, no ophyd): - CameraTransport ABC; XCTransport implements the Canon XC Control Protocol (session open/claim/yield, HTTP/CGI) with an injectable opener; FakeTransport is a full in-memory simulator for offline mode and tests. Spec-defined CGI paths/params/axis encodings (BPE-7216-005) are isolated and flagged for hardware verification. Imaging plane (acquisition/, pure stdlib, no ophyd, not a control dependency): - FrameRingBuffer (bounded, thread-safe, stores compressed frames) and a persistent, fail-soft, instrumented StreamWorker. Phase-2 grab API (get_latest/get_latest_n/get_nearest, decode-on-grab) seamed in the device. Credentials come from env vars (CANON_<NAME>_USER/_PASSWORD), never YAML/Redis. Adds example device config (real + simulation), a README, and pytest coverage (transport mocked, ring buffer, stream worker, device logic) — no hardware required. No new third-party dependencies (Phase-2 decode uses cv2, already a csaxs_bec dependency). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
84 lines
2.4 KiB
Python
84 lines
2.4 KiB
Python
"""Tests for the Canon acquisition ring buffer (pure stdlib, no hardware)."""
|
|
|
|
import threading
|
|
|
|
import pytest
|
|
|
|
from csaxs_bec.devices.canon.acquisition.ring_buffer import Frame, FrameRingBuffer
|
|
|
|
|
|
def test_rejects_bad_maxlen():
|
|
with pytest.raises(ValueError):
|
|
FrameRingBuffer(0)
|
|
|
|
|
|
def test_put_stamps_sequence_and_timestamps():
|
|
clock = iter([1.0, 2.0, 3.0])
|
|
wall = iter([100.0, 200.0, 300.0])
|
|
buf = FrameRingBuffer(
|
|
8, source_id="cam", time_func=lambda: next(clock), wall_time_func=lambda: next(wall)
|
|
)
|
|
f0 = buf.put(b"a")
|
|
f1 = buf.put(b"bb")
|
|
assert isinstance(f0, Frame)
|
|
assert (f0.seq, f0.monotonic_ts, f0.wall_ts, f0.source_id) == (0, 1.0, 100.0, "cam")
|
|
assert (f1.seq, f1.nbytes) == (1, 2)
|
|
|
|
|
|
def test_get_latest_empty_and_filled():
|
|
buf = FrameRingBuffer(4)
|
|
assert buf.get_latest() is None
|
|
buf.put(b"x")
|
|
buf.put(b"y")
|
|
assert buf.get_latest().data == b"y"
|
|
|
|
|
|
def test_overwrites_oldest_and_counts_drops():
|
|
buf = FrameRingBuffer(3)
|
|
for i in range(5):
|
|
buf.put(bytes([i]))
|
|
snap = buf.snapshot()
|
|
assert [f.data for f in snap] == [b"\x02", b"\x03", b"\x04"] # oldest two evicted
|
|
stats = buf.stats()
|
|
assert stats["size"] == 3
|
|
assert stats["total_received"] == 5
|
|
assert stats["dropped_unread"] == 2
|
|
|
|
|
|
def test_get_latest_n():
|
|
buf = FrameRingBuffer(5)
|
|
for i in range(4):
|
|
buf.put(bytes([i]))
|
|
assert [f.data for f in buf.get_latest_n(2)] == [b"\x02", b"\x03"]
|
|
assert len(buf.get_latest_n(10)) == 4 # capped at available
|
|
assert buf.get_latest_n(0) == []
|
|
|
|
|
|
def test_get_nearest_wall_and_monotonic():
|
|
mono = iter([1.0, 2.0, 3.0])
|
|
wall = iter([10.0, 20.0, 30.0])
|
|
buf = FrameRingBuffer(8, time_func=lambda: next(mono), wall_time_func=lambda: next(wall))
|
|
buf.put(b"a")
|
|
buf.put(b"b")
|
|
buf.put(b"c")
|
|
assert buf.get_nearest(19.0).data == b"b" # wall clock default
|
|
assert buf.get_nearest(2.9, clock="monotonic").data == b"c"
|
|
empty = FrameRingBuffer(2)
|
|
assert empty.get_nearest(0.0) is None
|
|
|
|
|
|
def test_thread_safety_under_concurrent_writers():
|
|
buf = FrameRingBuffer(1000)
|
|
|
|
def worker(tag):
|
|
for i in range(200):
|
|
buf.put(f"{tag}-{i}".encode())
|
|
|
|
threads = [threading.Thread(target=worker, args=(t,)) for t in range(5)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
assert buf.stats()["total_received"] == 1000
|
|
assert len(buf) == 1000
|