From 994195f92c85871f96ce651af022e7a3b1336fa1 Mon Sep 17 00:00:00 2001 From: gbradham Date: Thu, 20 Aug 2026 22:58:05 -0500 Subject: [PATCH] feat(vision): rehearse a closed loop from a recording A closed-loop experiment has a lot of links -- pose inference, the behavior model, the node watching for a behavior, the canvas wiring, the device binding, the peripheral -- and any one of them can be wrong in a way that looks fine until an animal is in the box. This plays a recording through the live path instead of the camera, so the whole chain runs on footage where you already know what the animal did. The hardware fires for real, which is the point: a rehearsal that faked the output would not tell you the stimulator works. It needed almost nothing new downstream. _handle_frame_input is a single funnel that every frame already goes through, so a pump emitting _frame_received with the camera's (frame, timestamp) shape gets classification, node triggers and BLE writes unchanged. It emits that signal rather than calling _on_frame, deliberately skipping the _preview_active guard: a rehearsal must not require an open camera, since most machines running one have none. Frames are never skipped. compute_features computes velocity and acceleration over unit frame spacing, so a dropped frame doubles the apparent displacement across the gap and inflates exactly the kinematics the model keys on -- a pump that skipped to keep up would report confident, wrong behavior. It reports its worst lag instead, which is also the number worth reading: a run that ends a second behind will miss stimulus timing on a live animal by the same margin. Both speeds classify identically, for the same reason: unit frame spacing means the playback rate cannot change what the model sees. They answer different questions. Real time answers whether inference keeps up and what the end-to-end latency is; as-fast-as-possible answers whether any of it is wired up, sooner. Offering only the first would have made the wiring check needlessly slow, and offering only the second would have made the latency question unanswerable. A consumer that raises ends the run rather than being swallowed, because continuing would deliver a gap-riddled stream -- the one thing this must never produce. --- CHANGELOG.md | 1 + docs-site/camera-behavior/rehearsal.md | 76 +++++ mkdocs.yml | 1 + src/glider/gui/panels/camera_panel.py | 129 ++++++++ src/glider/vision/video_pump.py | 232 +++++++++++++++ tests/unit/gui/test_behavior_input_editor.py | 4 + tests/unit/gui/test_camera_panel_rehearsal.py | 176 +++++++++++ tests/unit/vision/test_video_pump.py | 275 ++++++++++++++++++ 8 files changed, 894 insertions(+) create mode 100644 docs-site/camera-behavior/rehearsal.md create mode 100644 src/glider/vision/video_pump.py create mode 100644 tests/unit/gui/test_camera_panel_rehearsal.py create mode 100644 tests/unit/vision/test_video_pump.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 5eb34a1..9d5ed40 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **Rehearse a closed loop from a recording.** **Camera → Live behavior → Rehearse from video…** plays a clip through the live path instead of the camera: the classifier runs, the nodes fire, and the hardware is driven for real, on footage where you already know what the animal did. Frames are never skipped — the live feature extractor uses unit frame spacing, so a dropped frame inflates exactly the kinematics the model keys on — so a run that cannot keep up reports its worst lag instead, which is also the number that says whether the rig will hold up live. Real-time and as-fast-as-possible modes classify identically (`compute_features` never reads fps); they differ only in whether they can answer "does inference keep up?". - **Maimu moves out of core into an installable plugin, `glider-maimu`.** The device and node are unchanged; what changed is that they are no longer bundled. One lab's stimulator does not belong in every install, and moving it made it the first real consumer of the plugin node extension points — which is the only way to find out whether they are good enough. Two gaps surfaced doing it and are fixed here: a plugin node could not declare its canvas ports (the editor now reads them off the node's own `NodeDefinition` instead of falling back to a generic one-in-one-out, which also fixes every built-in node missing from the hand-written table), and a plugin node could not be bound to a device (a node now declares `REQUIRES_DEVICE`, which `HardwareNode` sets, instead of the editor matching a hardcoded list of type names). - **Maimu BLE stimulator** — a `Maimu` device type with the peripheral's GATT layout built in (**Add Device → Maimu → Scan**, no UUIDs to paste), and a **Maimu** node in the library's I/O section offering Mode (On / Off / Pulse) with a period and duration, instead of a Device Action node writing `"500,10"` by hand. Unlike the generic BLE devices, its shutdown writes `off` before disconnecting — the firmware runs a pulse autonomously, so an emergency stop that only dropped the link would leave the device stimulating. - **DeepLabCut and SLEAP pose models** — a `PoseBackend` seam behind the three places GLIDER loaded a pose net lets DeepLabCut single-animal and SLEAP single-instance models drive live camera inference, offline classification, and batch runs. They run through `onnxruntime` (new `glider[pose-onnx]` extra), after a one-time export performed in your own DeepLabCut or SLEAP environment with `tools/export_pose_onnx.py` — GLIDER never imports `deeplabcut` or `sleap`, which is what keeps it installable on Python 3.11-3.13 and on the Pi. Ultralytics YOLO `.pt` weights are unchanged: that path is a pass-through over the same code as before. Multi-animal architectures, including SLEAP top-down and bottom-up, are rejected by name rather than silently scoring one arbitrary animal. diff --git a/docs-site/camera-behavior/rehearsal.md b/docs-site/camera-behavior/rehearsal.md new file mode 100644 index 0000000..c53e21d --- /dev/null +++ b/docs-site/camera-behavior/rehearsal.md @@ -0,0 +1,76 @@ +# Rehearsing a closed loop from a recording + +A closed-loop experiment — stimulate *because* the animal did something — has a +lot of links: pose inference, the behavior model, the node that watches for a +behavior, the wiring on the canvas, the device binding, and the hardware itself. +Any one of them can be wrong in a way that looks fine until an animal is in the +box. + +**Rehearsal mode plays a recording through the live path instead of the camera.** +The classifier runs, the nodes fire, and the hardware is driven for real — on +footage where you already know what the animal did. + +!!! warning "The hardware really fires" + This is the point: a rehearsal that faked the output would not tell you the + stimulator works. Don't rehearse with an animal connected. + +## Running one + +1. Load your pose model and behavior model in **Camera → Live behavior**, set the + keypoint names, and press **Start**. The classifier has to be running — a + rehearsal with nothing listening does nothing, and the panel says so. +2. Build the flow you'd run for real: a **Behavior Input** node watching the + behavior you care about, wired to whatever it triggers. +3. Press **Rehearse from video…** and pick a clip. + +The status line reports progress, and at the end it tells you the worst lag. + +## Real time or as fast as possible + +Both classify the recording **identically**. Feature values come from +`compute_features`, which uses unit frame spacing and never reads fps — so the +playback rate cannot change what the model sees. The two modes exist because +they answer different questions: + +| Mode | Answers | +| --- | --- | +| **Real time** | Does inference keep up on this machine, and how long does a stimulus take to arrive after a behavior starts? | +| **As fast as possible** | Is any of this wired up correctly? Same answer, sooner. | + +Use *as fast as possible* while you're still fixing wiring, and *real time* for +the run you actually believe. + +## Reading the result + +**"kept up with real time"** — inference is fast enough on this machine for this +footage. The rig should behave the same live. + +**"worst lag N ms — inference did not keep up"** — the model is slower than the +frame rate. A stimulus will arrive late on a live animal by roughly that margin, +on top of the confirmation delay from the Behavior Input node's `min_frames` +setting (at 30 fps, 5 frames is about 167 ms). Either is fine if it is small +against the behavior you are studying, and neither is fine if it isn't. + +Frames are **never skipped**, whatever happens. The live feature extractor +computes velocity and acceleration over unit frame spacing, so a dropped frame +doubles the apparent displacement across the gap and inflates the kinematics the +model keys on. A rehearsal that dropped frames to keep up would report confident, +wrong behavior — so it reports lag instead. + +## What it does and doesn't prove + +**Does:** the model loads and classifies; `min_frames` is tuned for your actual +footage; node wiring; the device binds; the peripheral connects; the commands +sent are the ones you meant; end-to-end latency. + +**Doesn't:** camera setup, lighting, or whether inference keeps up on your +*camera's* resolution and frame rate — unless the clip was recorded on the same +camera at the same settings. It's worth using footage from the rig you're about +to run. + +## Getting a clip + +Any recording the model was trained to handle. A session recorded by GLIDER's own +video recorder is ideal, because it matches the camera and settings you'll use. +Pick one where the behavior you're triggering on actually occurs — a rehearsal on +footage with no freezing tells you nothing about a freeze trigger. diff --git a/mkdocs.yml b/mkdocs.yml index c7bf3fe..c12a199 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -103,6 +103,7 @@ nav: - Tracking: camera-behavior/tracking.md - Pose Models: camera-behavior/pose-models.md - Behavior Analysis: camera-behavior/behavior.md + - Closed-loop Rehearsal: camera-behavior/rehearsal.md - Runner Mode: - runner/index.md - The Runner Screen: runner/runner.md diff --git a/src/glider/gui/panels/camera_panel.py b/src/glider/gui/panels/camera_panel.py index a4790d3..7e409d2 100644 --- a/src/glider/gui/panels/camera_panel.py +++ b/src/glider/gui/panels/camera_panel.py @@ -306,6 +306,8 @@ class CameraPanel(QWidget): # into the BehaviorInferenceWorker's thread via automatic QueuedConnections. _behavior_init_requested = pyqtSignal(str, str, list) # pkl, pt, keypoint names _behavior_frame_requested = pyqtSignal(object) # FrameData -> worker.process_frame + # PumpStats from the rehearsal pump's thread -> the GUI thread. + _rehearsal_finished = pyqtSignal(object) def __init__( self, @@ -345,6 +347,7 @@ def __init__( self._behavior_thread: QThread | None = None self._behavior_worker: Any | None = None self._behavior_running = False + self._rehearsal_pump: Any | None = None # Video-file source state (offline tracking) from glider.vision.video_source import VideoFileSource @@ -598,6 +601,44 @@ def _setup_ui(self) -> None: self._live_behavior_btn.clicked.connect(self._toggle_live_behavior) behavior_layout.addWidget(self._live_behavior_btn) + # --- rehearsal: drive the live path from a recording ------------------ + rehearsal_line = QFrame() + rehearsal_line.setFrameShape(QFrame.Shape.HLine) + rehearsal_line.setProperty("textRole", "muted") + behavior_layout.addWidget(rehearsal_line) + + rehearsal_row = QHBoxLayout() + self._rehearse_btn = QPushButton("Rehearse from video\u2026") + self._rehearse_btn.setEnabled(False) + self._rehearse_btn.setToolTip( + "Play a recording through the live path instead of the camera.\n" + "Classification, node triggers and hardware all run for real." + ) + self._rehearse_btn.clicked.connect(self._toggle_rehearsal) + rehearsal_row.addWidget(self._rehearse_btn, 1) + + self._rehearse_speed = QComboBox() + # Feature values do not depend on playback rate -- compute_features uses + # unit frame spacing -- so both modes classify identically. They answer + # different questions, which is why both exist. + self._rehearse_speed.addItem("Real time", 1.0) + self._rehearse_speed.addItem("As fast as possible", 0.0) + self._rehearse_speed.setToolTip( + "Real time answers 'does inference keep up, and what is the " + "latency?'.\nAs fast as possible answers 'is this wired up?' " + "sooner.\nBoth classify the recording identically." + ) + rehearsal_row.addWidget(self._rehearse_speed) + behavior_layout.addLayout(rehearsal_row) + + self._rehearse_status = QLabel( + "Start live behavior first \u2014 a rehearsal with no classifier " + "running does nothing." + ) + self._rehearse_status.setProperty("textRole", "muted") + self._rehearse_status.setWordWrap(True) + behavior_layout.addWidget(self._rehearse_status) + layout.addWidget(behavior_group) # Set up scroll area @@ -615,6 +656,7 @@ def _connect_signals(self) -> None: # Connect thread-safe signals for UI updates (main thread) self._frame_received.connect(self._handle_frame_input) + self._rehearsal_finished.connect(self._on_rehearsal_finished_main) self._multi_frame_received.connect(self._handle_multi_frame_input) # Connect CV worker signals @@ -646,6 +688,88 @@ def _connect_signals(self) -> None: self._run_fps = FpsMeter() self._run_frames_done = 0 + # --- rehearsal from a recording --------------------------------------- + + def _toggle_rehearsal(self) -> None: + if self._rehearsal_pump is not None and self._rehearsal_pump.is_running: + self._stop_rehearsal() + else: + self._start_rehearsal() + + def _start_rehearsal(self) -> None: + """Play a recording through the live path, hardware and all.""" + from PyQt6.QtWidgets import QFileDialog + + from glider.vision.video_pump import VideoPump + + path, _ = QFileDialog.getOpenFileName( + self, "Rehearse from video", "", "Videos (*.mp4 *.avi *.mov *.mkv);;All Files (*)" + ) + if not path: + return + + speed = self._rehearse_speed.currentData() + self._rehearsal_pump = VideoPump( + path, + self._on_rehearsal_frame, + speed=speed, + on_finished=self._on_rehearsal_finished, + ) + if not self._rehearsal_pump.start(): + self._rehearsal_pump = None + self._rehearse_status.setText(f"Could not open {Path(path).name}.") + return + + self._rehearse_btn.setText("Stop rehearsal") + self._rehearse_speed.setEnabled(False) + self._rehearse_status.setText( + f"Rehearsing {Path(path).name} \u2014 hardware is being driven for real." + ) + logger.info("Rehearsal started from %s", path) + + def _stop_rehearsal(self) -> None: + pump = self._rehearsal_pump + if pump is not None: + pump.stop() + self._rehearsal_pump = None + self._rehearse_btn.setText("Rehearse from video\u2026") + self._rehearse_speed.setEnabled(True) + logger.info("Rehearsal stopped") + + def _on_rehearsal_frame(self, frame: np.ndarray, timestamp: float) -> None: + """Hand a recorded frame to the live path. + + Called from the pump's thread, exactly as ``_on_frame`` is called from + the camera's capture thread, and doing the same thing: copy, then emit + so the work happens on the GUI thread. Emitting ``_frame_received`` + rather than calling ``_on_frame`` deliberately skips its + ``_preview_active`` guard -- a rehearsal does not need a camera open, + and requiring one would mean it could not run on a machine that has + none. + """ + self._frame_received.emit(FrameData(frame=frame.copy(), timestamp=timestamp)) + + def _on_rehearsal_finished(self, stats: Any) -> None: + """Report how it went. Called from the pump's thread.""" + self._rehearsal_finished.emit(stats) + + def _on_rehearsal_finished_main(self, stats: Any) -> None: + self._rehearse_btn.setText("Rehearse from video\u2026") + self._rehearse_speed.setEnabled(True) + self._rehearsal_pump = None + + lag_ms = stats.max_lag_s * 1000 + summary = f"Rehearsal finished: {stats.frames_delivered} frames" + if stats.max_lag_s > 0: + # The number that says whether the rig will hold up live: a run + # that ended a second behind will miss stimulus timing on an animal + # by the same margin. + summary += f", worst lag {lag_ms:.0f} ms \u2014 inference did not keep up" + elif self._rehearse_speed.currentData(): + summary += ", kept up with real time" + self._rehearse_status.setText(summary) + logger.info("%s", summary) + def _handle_frame_input(self, frame_data: FrameData) -> None: """Decide whether to process frame with CV or update UI immediately.""" # Fan out EVERY frame to the live-behavior worker (independent of the CV @@ -1602,6 +1726,8 @@ def _on_behavior_ready(self) -> None: if worker is None: return self._preview.set_behavior_vocab(worker.classes) + self._rehearse_btn.setEnabled(True) + self._rehearse_status.setText("Ready to rehearse from a recording.") # Also hand the vocabulary to the flow side, so a Behavior Input node's # properties can offer the behaviors this model actually emits instead # of a free-text box where a typo means "never fires". @@ -1711,6 +1837,9 @@ def closeEvent(self, event): # Tear down any in-flight tracking run and release the scrub video source. self._teardown_run_thread() self.stop_live_behavior() + if self._rehearsal_pump is not None: + self._rehearsal_pump.stop() + self._rehearsal_pump = None self._video_source.release() if self._preview_active: diff --git a/src/glider/vision/video_pump.py b/src/glider/vision/video_pump.py new file mode 100644 index 0000000..778d568 --- /dev/null +++ b/src/glider/vision/video_pump.py @@ -0,0 +1,232 @@ +"""Feed a recorded video into the live pipeline, as if it were a camera. + +Lets a whole closed loop be rehearsed off a recording: pose inference, behavior +classification, the nodes that trigger on it, and the real hardware they drive. +The point is to find out whether the rig fires correctly *before* an animal is +in the box, using footage where you already know what the animal did. + +It hands frames to the same callback ``CameraManager`` streams to, with the +same ``(frame, timestamp)`` shape and the same wall-clock ``time.time()`` +timestamps, so nothing downstream can tell the difference. Everything the live +path does -- classification, node triggers, BLE writes -- happens for real. + +**Frames are never skipped.** That is the one hard rule. The live feature +extractor computes velocity and acceleration with ``np.gradient`` over *unit +frame spacing* (``compute_features`` never reads fps), so a dropped frame +doubles the apparent displacement across the gap and inflates the kinematics +that the behavior model was trained on. Falling behind is reported instead; a +rehearsal that quietly skipped frames would produce confident, wrong behavior. + +**Speed does not change what gets classified.** Because features use unit frame +spacing, the same recording yields identical features at any playback rate. +Speed changes only two things, and they are the reasons both modes exist: + +``speed=1.0`` + Real time. Answers "does inference keep up on this machine, and what is the + end-to-end latency from behavior onset to stimulus?" -- the questions that + decide whether the rig works. + +``speed=0`` (as fast as the decoder and model allow) + Answers "is any of this wired up correctly?" in a fraction of the runtime. + Same classifications, same commands, wrong timing. +""" + +from __future__ import annotations + +import logging +import threading +import time +from collections.abc import Callable +from dataclasses import dataclass +from pathlib import Path + +import numpy as np + +from glider.vision.video_source import VideoFileSource + +logger = logging.getLogger(__name__) + +FrameCallback = Callable[[np.ndarray, float], None] + + +@dataclass(frozen=True) +class PumpStats: + """A snapshot of how the rehearsal is going.""" + + frames_delivered: int + total_frames: int + #: Seconds behind the real-time schedule right now. Always 0 when unpaced. + lag_s: float + #: Worst lag seen this run. This is the number that says whether inference + #: keeps up: a rehearsal that ends seconds behind will miss stimulus timing + #: on a live animal by the same margin. + max_lag_s: float + finished: bool + + @property + def progress(self) -> float: + """0.0-1.0, or 0.0 when the frame count is unknown.""" + if self.total_frames <= 0: + return 0.0 + return min(1.0, self.frames_delivered / self.total_frames) + + +class VideoPump: + """Play a video file into ``on_frame``, in order, without skipping. + + Args: + path: The video to play. + on_frame: Called with ``(frame, timestamp)`` per frame, from the pump's + own thread -- exactly like ``CameraManager``'s stream callback. A + Qt consumer should marshal to the GUI thread, as the camera path + already does. + speed: Playback rate relative to the recording. 1.0 is real time; 0 + means as fast as possible. Values above 1 are allowed and are + honest, because features do not depend on wall-clock rate. + on_finished: Called once when the video ends or the pump is stopped, + with the final :class:`PumpStats`. Also from the pump's thread. + """ + + def __init__( + self, + path: Path | str, + on_frame: FrameCallback, + *, + speed: float = 1.0, + on_finished: Callable[[PumpStats], None] | None = None, + ) -> None: + self._path = Path(path) + self._on_frame = on_frame + self._speed = max(0.0, float(speed)) + self._on_finished = on_finished + + self._source = VideoFileSource() + self._thread: threading.Thread | None = None + self._stop = threading.Event() + + self._lock = threading.Lock() + self._delivered = 0 + self._lag = 0.0 + self._max_lag = 0.0 + self._finished = False + + # -- lifecycle --------------------------------------------------------- + + @property + def fps(self) -> float: + return self._source.fps + + @property + def total_frames(self) -> int: + return self._source.frame_count + + @property + def is_running(self) -> bool: + return self._thread is not None and self._thread.is_alive() + + def start(self) -> bool: + """Open the video and begin playing. False if it cannot be read.""" + if self.is_running: + logger.warning("VideoPump: already running") + return False + if not self._source.load(self._path): + logger.error("VideoPump: could not open %s", self._path) + return False + + self._stop.clear() + with self._lock: + self._delivered = 0 + self._lag = 0.0 + self._max_lag = 0.0 + self._finished = False + + self._thread = threading.Thread(target=self._run, name="glider-video-pump", daemon=True) + self._thread.start() + logger.info( + "VideoPump: playing %s (%d frames @ %.2f fps, speed=%s)", + self._path.name, + self.total_frames, + self.fps, + "max" if self._speed == 0 else f"{self._speed:g}x", + ) + return True + + def stop(self, timeout: float = 5.0) -> None: + """Stop playing and wait for the thread to wind down. + + Safe to call when not running, and safe to call from the finished + callback. The thread is a daemon, so a wedged decoder cannot keep the + application alive. + """ + self._stop.set() + thread = self._thread + if thread is not None and thread is not threading.current_thread(): + thread.join(timeout=timeout) + self._thread = None + self._source.release() + + def stats(self) -> PumpStats: + with self._lock: + return PumpStats( + frames_delivered=self._delivered, + total_frames=self._source.frame_count, + lag_s=self._lag, + max_lag_s=self._max_lag, + finished=self._finished, + ) + + # -- the pump ---------------------------------------------------------- + + def _run(self) -> None: + try: + self._pump() + except Exception: + logger.exception("VideoPump: playback failed") + finally: + with self._lock: + self._finished = True + stats = self.stats() + self._source.release() + if self._on_finished is not None: + try: + self._on_finished(stats) + except Exception: + logger.exception("VideoPump: finished callback raised") + + def _pump(self) -> None: + fps = self._source.fps or 30.0 + interval = 1.0 / (fps * self._speed) if self._speed > 0 else 0.0 + started = time.perf_counter() + + for index, frame in self._source.frames(): + if self._stop.is_set(): + logger.info("VideoPump: stopped after %d frames", index) + return + + if interval: + # Schedule against the start, not the previous frame, so a slow + # frame does not push every later one back by the same amount. + wait = (started + index * interval) - time.perf_counter() + if wait > 0: + # Interruptible: a stop during a wait takes effect at once + # rather than after the frame's worth of sleep. + if self._stop.wait(wait): + return + else: + self._record_lag(-wait) + + # Deliberately NOT wrapped in try/except. A callback that raises is + # a broken consumer, and continuing would deliver a gap-riddled + # stream that reads as an animal moving in jumps. _run logs it and + # ends the rehearsal, which is the honest outcome. + self._on_frame(frame, time.time()) + + with self._lock: + self._delivered = index + 1 + + logger.info("VideoPump: reached the end of %s", self._path.name) + + def _record_lag(self, seconds: float) -> None: + with self._lock: + self._lag = seconds + self._max_lag = max(self._max_lag, seconds) diff --git a/tests/unit/gui/test_behavior_input_editor.py b/tests/unit/gui/test_behavior_input_editor.py index cd674c2..b741f67 100644 --- a/tests/unit/gui/test_behavior_input_editor.py +++ b/tests/unit/gui/test_behavior_input_editor.py @@ -253,6 +253,8 @@ def test_loading_a_model_publishes_its_vocabulary_to_the_bus(qtbot): panel._preview = SimpleNamespace(set_behavior_vocab=lambda names: None) panel._behavior_running = False panel._live_behavior_btn = SimpleNamespace(setText=lambda t: None, setEnabled=lambda e: None) + panel._rehearse_btn = SimpleNamespace(setEnabled=lambda e: None) + panel._rehearse_status = SimpleNamespace(setText=lambda t: None) panel._on_behavior_ready() @@ -269,6 +271,8 @@ def test_a_panel_with_no_bus_still_goes_live(qtbot): panel._preview = SimpleNamespace(set_behavior_vocab=lambda names: None) panel._behavior_running = False panel._live_behavior_btn = SimpleNamespace(setText=lambda t: None, setEnabled=lambda e: None) + panel._rehearse_btn = SimpleNamespace(setEnabled=lambda e: None) + panel._rehearse_status = SimpleNamespace(setText=lambda t: None) panel._on_behavior_ready() diff --git a/tests/unit/gui/test_camera_panel_rehearsal.py b/tests/unit/gui/test_camera_panel_rehearsal.py new file mode 100644 index 0000000..febb10e --- /dev/null +++ b/tests/unit/gui/test_camera_panel_rehearsal.py @@ -0,0 +1,176 @@ +"""The rehearsal has to be indistinguishable from a camera, downstream. + +The whole value of playing a recording through the live path is that nothing +downstream treats it specially: the same classifier runs, the same nodes fire, +the same hardware is written to. If the frames took a different route -- or +arrived with a different shape of timestamp -- a green rehearsal would prove +nothing about the live run it is standing in for. +""" + +from __future__ import annotations + +import time +from types import SimpleNamespace + +import numpy as np +import pytest + +from glider.gui.panels.camera_panel import CameraPanel, FrameData + +pytestmark = pytest.mark.usefixtures("qtbot") + + +def _panel(): + """A panel with only what the rehearsal path touches.""" + panel = CameraPanel.__new__(CameraPanel) # skip the heavy __init__ + panel._rehearsal_pump = None + panel._behavior_running = True + return panel + + +def test_a_rehearsal_frame_goes_down_the_camera_funnel(qtbot): + """_frame_received is what the camera emits and what _handle_frame_input + listens on. A rehearsal that used any other route would bypass the + behavior worker, which is the entire point of running one.""" + panel = _panel() + received: list[FrameData] = [] + panel._frame_received = SimpleNamespace(emit=received.append) + + frame = np.full((4, 4, 3), 7, dtype=np.uint8) + panel._on_rehearsal_frame(frame, 1234.5) + + assert len(received) == 1 + assert isinstance(received[0], FrameData) + assert received[0].timestamp == 1234.5 + assert int(received[0].frame[0, 0, 0]) == 7 + + +def test_the_frame_is_copied(qtbot): + """The pump reuses decoder buffers; handing the original across a thread + boundary would let it change under the consumer.""" + panel = _panel() + received: list[FrameData] = [] + panel._frame_received = SimpleNamespace(emit=received.append) + + frame = np.zeros((2, 2, 3), dtype=np.uint8) + panel._on_rehearsal_frame(frame, time.time()) + frame[:] = 99 # the decoder moves on + + assert int(received[0].frame[0, 0, 0]) == 0 + + +def test_it_does_not_require_an_open_camera(qtbot): + """_on_frame drops everything unless _preview_active. A rehearsal has to + work on a machine with no camera at all -- which is most of them.""" + panel = _panel() + panel._preview_active = False + received: list[FrameData] = [] + panel._frame_received = SimpleNamespace(emit=received.append) + + panel._on_rehearsal_frame(np.zeros((2, 2, 3), dtype=np.uint8), 1.0) + + assert len(received) == 1 + + +# --- the end-to-end claim ----------------------------------------------------- + + +async def test_a_recording_drives_real_hardware(qtbot, monkeypatch, tmp_path): + """The claim the feature makes: play a video, the stimulator fires. + + Everything between the frames and the GATT write is the real thing -- the + live signal bus, the Behavior Input node, the exec connection, the device. + Only the two genuine externals are faked: pose/behavior inference (which + would need a trained model and weights) and bleak. + """ + import sys + from unittest.mock import MagicMock + + written: list[bytes] = [] + + class _FakeClient: + def __init__(self, address): + self.address = address + self.is_connected = False + + async def connect(self): + self.is_connected = True + + async def disconnect(self): + self.is_connected = False + + async def write_gatt_char(self, char, data, response=False): + written.append(bytes(data)) + + bleak = MagicMock(name="bleak") + bleak.BleakClient = _FakeClient + monkeypatch.setitem(sys.modules, "bleak", bleak) + + from glider.core.flow_engine import FlowEngine + from glider.core.hardware_manager import HardwareManager + from glider.core.live_signals import BehaviorEvent, LiveSignalBus + from glider.hal.mock_board import MockBoard + from glider.nodes.vision import register_behavior_nodes + + maimu = pytest.importorskip( + "glider_maimu", reason="glider-maimu is not installed; the stimulus half needs it" + ) + + HardwareManager.register_driver("mock", MockBoard) + hardware = HardwareManager() + hardware._boards["b1"] = MockBoard() + monkeypatch.setitem( + __import__("glider.hal.base_device", fromlist=["DEVICE_REGISTRY"]).DEVICE_REGISTRY, + "Maimu", + maimu.MaimuDevice, + ) + hardware.add_device_multi_pin( + "stim", "Maimu", "b1", pins={}, name="Stim", settings={"address": "AA:BB"} + ) + await hardware.initialize_device("stim") + + bus = LiveSignalBus() + engine = FlowEngine(hardware) + engine.set_live_signals(bus) + register_behavior_nodes(engine) + monkeypatch.setitem(FlowEngine._node_registry, "Maimu", maimu.MaimuNode) + + watcher = engine.create_node(node_id="w", node_type="BehaviorInput", position=(0.0, 0.0)) + watcher.target_behavior = "freezing" + watcher.min_frames = 3 + stim = engine.create_node(node_id="s", node_type="Maimu", position=(1.0, 0.0), device_id="stim") + stim.mode = "pulse" + stim.period_ms = 500 + stim.duration_s = 10 + engine.create_connection( + connection_id="c", + from_node_id="w", + from_output=2, + to_node_id="s", + to_input=0, + connection_type="exec", + ) + await engine.start() + + import asyncio + + deadline = asyncio.get_running_loop().time() + 2.0 + while not watcher._subscribed and asyncio.get_running_loop().time() < deadline: + await asyncio.sleep(0.01) + + # The stand-in for pose + behavior inference: every frame of this clip is a + # freeze, which is what the recording would contain. + panel = _panel() + panel._frame_received = SimpleNamespace( + emit=lambda fd: bus.publish_behavior(BehaviorEvent(behavior="freezing")) + ) + + for _ in range(5): + panel._on_rehearsal_frame(np.zeros((2, 2, 3), dtype=np.uint8), time.time()) + + deadline = asyncio.get_running_loop().time() + 2.0 + while not written and asyncio.get_running_loop().time() < deadline: + await asyncio.sleep(0.01) + await engine.stop() + + assert written == [b"500,10"], "a recorded freeze did not reach the stimulator" diff --git a/tests/unit/vision/test_video_pump.py b/tests/unit/vision/test_video_pump.py new file mode 100644 index 0000000..37e9f4e --- /dev/null +++ b/tests/unit/vision/test_video_pump.py @@ -0,0 +1,275 @@ +"""The rehearsal pump must deliver every frame, in order, and admit when it lags. + +Those three are not stylistic. The live feature extractor computes velocity and +acceleration over *unit frame spacing*, so a skipped frame doubles the apparent +displacement across the gap and inflates exactly the kinematics the behavior +model keys on. A pump that quietly dropped frames would produce confident, +wrong behavior -- and a rehearsal exists to be believed. +""" + +from __future__ import annotations + +import threading +import time + +import numpy as np +import pytest + +from glider.vision.video_pump import VideoPump + + +class _FakeSource: + """Stands in for VideoFileSource: a fixed number of numbered frames.""" + + def __init__(self, count=10, fps=30.0): + self.frame_count = count + self.fps = fps + self.released = False + self.loaded_path = None + self._count = count + + def load(self, path): + self.loaded_path = path + return True + + def frames(self): + for index in range(self._count): + # The frame's own index is written into the pixels, so a test can + # prove ordering and completeness from what the callback received. + yield index, np.full((2, 2, 3), index, dtype=np.uint8) + + def release(self): + self.released = True + + +@pytest.fixture +def pump_factory(monkeypatch): + """Build a pump over a fake source, returning (pump, received, source).""" + + def _make(count=10, fps=30.0, speed=0.0, on_frame=None, **kwargs): + source = _FakeSource(count=count, fps=fps) + monkeypatch.setattr("glider.vision.video_pump.VideoFileSource", lambda: source) + received: list[tuple[int, float]] = [] + + def _record(frame, timestamp): + received.append((int(frame[0, 0, 0]), timestamp)) + + pump = VideoPump("clip.mp4", on_frame or _record, speed=speed, **kwargs) + return pump, received, source + + return _make + + +def _run_to_completion(pump, timeout=5.0): + done = threading.Event() + original = pump._on_finished + + def _finished(stats): + if original: + original(stats) + done.set() + + pump._on_finished = _finished + assert pump.start() + assert done.wait(timeout), "the pump never finished" + pump.stop() + + +# --- the hard rule: every frame, in order ------------------------------------- + + +def test_every_frame_is_delivered_in_order(pump_factory): + pump, received, _ = pump_factory(count=25) + + _run_to_completion(pump) + + assert [index for index, _ in received] == list(range(25)) + + +def test_no_frame_is_skipped_even_when_the_consumer_is_slow(pump_factory): + """The failure this pump exists to avoid. A slow consumer must make the + rehearsal late, never gap-toothed.""" + slow_calls = [] + + def _slow(frame, timestamp): + time.sleep(0.004) # far longer than the 1/1000s frame interval below + slow_calls.append(int(frame[0, 0, 0])) + + pump, _received, _ = pump_factory(count=20, fps=1000.0, speed=1.0, on_frame=_slow) + + _run_to_completion(pump) + + assert slow_calls == list(range(20)) + + +def test_it_reports_lag_rather_than_catching_up_by_skipping(pump_factory): + def _slow(frame, timestamp): + time.sleep(0.004) + + pump, _received, _ = pump_factory(count=15, fps=1000.0, speed=1.0, on_frame=_slow) + + _run_to_completion(pump) + + stats = pump.stats() + assert stats.frames_delivered == 15 + assert stats.max_lag_s > 0, "a pump that could not keep up reported no lag" + + +def test_an_unpaced_run_reports_no_lag(pump_factory): + """Lag is meaningless without a schedule; reporting some would be noise.""" + pump, _received, _ = pump_factory(count=10, speed=0.0) + + _run_to_completion(pump) + + assert pump.stats().max_lag_s == 0.0 + + +# --- timestamps --------------------------------------------------------------- + + +def test_timestamps_are_wall_clock_like_the_camera(pump_factory): + """CameraManager passes time.time(); anything downstream measuring elapsed + real time has to see the same kind of number from a rehearsal.""" + before = time.time() + pump, received, _ = pump_factory(count=5) + + _run_to_completion(pump) + + after = time.time() + assert all(before <= ts <= after for _, ts in received) + + +def test_timestamps_increase(pump_factory): + pump, received, _ = pump_factory(count=10) + + _run_to_completion(pump) + + stamps = [ts for _, ts in received] + assert stamps == sorted(stamps) + + +# --- pacing ------------------------------------------------------------------- + + +def test_real_time_pacing_takes_about_the_clip_length(pump_factory): + """A 10-frame clip at 50 fps is 0.2 s. Real-time mode has to actually + take that long, or 'does inference keep up' is unanswerable.""" + pump, _received, _ = pump_factory(count=10, fps=50.0, speed=1.0) + + started = time.perf_counter() + _run_to_completion(pump) + elapsed = time.perf_counter() - started + + assert 0.12 < elapsed < 0.9, f"took {elapsed:.3f}s for a 0.2s clip" + + +def test_unpaced_is_much_faster_than_real_time(pump_factory): + pump, _received, _ = pump_factory(count=40, fps=50.0, speed=0.0) + + started = time.perf_counter() + _run_to_completion(pump) + elapsed = time.perf_counter() - started + + assert elapsed < 0.4, f"unpaced run took {elapsed:.3f}s for a 0.8s clip" + + +def test_speed_multiplies_the_rate(pump_factory): + pump, _received, _ = pump_factory(count=20, fps=50.0, speed=4.0) + + started = time.perf_counter() + _run_to_completion(pump) + elapsed = time.perf_counter() - started + + assert elapsed < 0.35, f"4x of a 0.4s clip took {elapsed:.3f}s" + + +# --- lifecycle ---------------------------------------------------------------- + + +def test_stopping_midway_stops_promptly(pump_factory): + """A rehearsal drives real hardware. Stop has to mean stop, not 'after the + rest of the clip'.""" + seen = threading.Event() + + def _on_frame(frame, timestamp): + seen.set() + + pump, _received, _ = pump_factory(count=100000, fps=1000.0, speed=1.0, on_frame=_on_frame) + assert pump.start() + assert seen.wait(2.0) + + started = time.perf_counter() + pump.stop() + + assert time.perf_counter() - started < 1.0 + assert not pump.is_running + + +def test_a_finished_run_reports_finished(pump_factory): + captured = {} + pump, _received, _ = pump_factory(count=5, on_finished=lambda s: captured.update(final=s)) + + _run_to_completion(pump) + + assert captured["final"].finished is True + assert captured["final"].frames_delivered == 5 + assert captured["final"].progress == 1.0 + + +def test_an_unreadable_file_fails_to_start(monkeypatch): + class _Unloadable: + frame_count = 0 + fps = 30.0 + + def load(self, path): + return False + + def release(self): + pass + + monkeypatch.setattr("glider.vision.video_pump.VideoFileSource", lambda: _Unloadable()) + pump = VideoPump("missing.mp4", lambda f, t: None) + + assert pump.start() is False + assert not pump.is_running + + +def test_starting_twice_is_refused(pump_factory): + pump, _received, _ = pump_factory(count=100000, fps=1000.0, speed=1.0) + assert pump.start() + try: + assert pump.start() is False + finally: + pump.stop() + + +def test_stop_is_safe_before_start(pump_factory): + pump, _received, _ = pump_factory() + pump.stop() # must not raise + assert not pump.is_running + + +def test_a_raising_consumer_ends_the_run_rather_than_gapping_it(pump_factory): + """Swallowing the error would deliver a stream with holes in it, which is + the one thing this pump must never produce.""" + delivered = [] + + def _explode(frame, timestamp): + delivered.append(int(frame[0, 0, 0])) + if len(delivered) == 3: + raise RuntimeError("consumer is broken") + + pump, _received, _ = pump_factory(count=50, on_frame=_explode) + + _run_to_completion(pump) + + assert delivered == [0, 1, 2], "the pump kept going after its consumer broke" + assert pump.stats().finished is True + + +def test_the_source_is_released(pump_factory): + pump, _received, source = pump_factory(count=5) + + _run_to_completion(pump) + + assert source.released is True