feat(gui): a hardware raster in Session Review

tests/unit/analysis/conftest.py

@@ -54,14 +54,27 @@ class RecordingSpec:
     state_velocities: dict[str, float] = field(
         default_factory=lambda: {"resting": 0.0, "active": 5.0, "locomotion": 8.0}
     )
-    # Extra event rows to append after the flow_marker[start] but before
-    # flow_marker[end]. Each entry is (flow_ms, source, board_id, pin, value).
-    # Example: ((1000.0, "output_write", "board0", "5", "1"),) writes an LED-on
-    # event 1s after flow start.
-    extra_events: tuple[tuple[float, str, str, str, str], ...] = ()
+    # Extra event rows appended after flow_marker[start] and before
+    # flow_marker[end]. Each entry is
+    # (flow_ms, source, board_id, device_id, device_type, pin, pin_type, value)
+    # matching the event log's column order exactly. Example:
+    #   ((1000.0, "output_write", "board0", "led1", "LED", "5", "DIGITAL", "1"),)
+    # writes an LED-on event 1s after flow start.
+    extra_events: tuple[tuple[float, str, str, str, str, str, str, str], ...] = ()
     write_tracking: bool = True
     write_data: bool = True
     write_events: bool = True
+    # Tracked objects per frame. >1 writes one tracking row per object per
+    # frame (object_id 0..n-1), matching a real multi-subject recording
+    # where every frame has one row per tracked animal.
+    n_objects: int = 1
+    # Opt-in: prepend a motion-only "heartbeat" row (object_id=-1, blank
+    # behavioral_state) to frame 1 of the tracking CSV, matching the shape
+    # tracking_logger.py writes when `_frame_count == 1` and no object has
+    # been detected yet (its most common firing, not a corner case — see
+    # tracking_logger.py around line 828). Default False keeps the default
+    # recording shape byte-identical.
+    include_heartbeat_row: bool = False
 
 
 def _iso(dt: datetime) -> str:
@@ -175,11 +188,43 @@ def _write_tracking_csv(
                 state = "unknown"
                 zone_ids = ""
                 velocity = 0.0
-            f.write(
-                f"{i + 1},{_iso(t_dt)},{elapsed_ms:.1f},{flow_cell},0,mouse,"
-                f"{bx:.1f},{by:.1f},{bbox_w:.1f},{bbox_h:.1f},0.900,"
-                f"{cx:.1f},{cy:.1f},0.00,0.00,0.00,{zone_ids},{state},{velocity:.2f}\n"
-            )
+            if spec.include_heartbeat_row and i == 0:
+                # Mirrors tracking_logger.py's heartbeat write (~line 828):
+                # object_id=-1, class="heartbeat", zeroed bbox, and every
+                # field past confidence left blank, including
+                # behavioral_state.
+                heartbeat_row = [
+                    i + 1,
+                    _iso(t_dt),
+                    f"{elapsed_ms:.1f}",
+                    flow_cell,
+                    -1,
+                    "heartbeat",
+                    0,
+                    0,
+                    0,
+                    0,
+                    "0.000",
+                    "",
+                    "",
+                    "",
+                    "",
+                    "",
+                    "",
+                    "",
+                    "",
+                ]
+                f.write(",".join(str(v) for v in heartbeat_row) + "\n")
+            for obj in range(spec.n_objects):
+                # Object 0's state matches the single-object fixture
+                # exactly; other objects get a distinct label so tests
+                # can tell one object's lane from another's.
+                obj_state = state if obj == 0 else f"{state}_obj{obj}"
+                f.write(
+                    f"{i + 1},{_iso(t_dt)},{elapsed_ms:.1f},{flow_cell},{obj},mouse,"
+                    f"{bx:.1f},{by:.1f},{bbox_w:.1f},{bbox_h:.1f},0.900,"
+                    f"{cx:.1f},{cy:.1f},0.00,0.00,0.00,{zone_ids},{obj_state},{velocity:.2f}\n"
+                )
 
         f.write("\n")
         f.write(f"# End Time,{_iso(flow_end_dt)}\n")
@@ -248,13 +293,23 @@ def _write_events_csv(
         )
         # Extra synthetic events (e.g., output_write at known flow times) so
         # event_triggered tests have something to bind to.
-        for flow_ms, source, board_id, pin, value in spec.extra_events:
+        for (
+            flow_ms,
+            source,
+            board_id,
+            device_id,
+            device_type,
+            pin,
+            pin_type,
+            value,
+        ) in spec.extra_events:
             event_dt = flow_start_dt + timedelta(milliseconds=flow_ms)
             event_elapsed = (event_dt - _BASE_DATETIME).total_seconds() * 1000
             frame = spec.n_pre_flow_frames + int(flow_ms / 1000.0 * spec.fps)
             f.write(
                 f"{frame},{_iso(event_dt)},{event_elapsed:.1f},"
-                f"{source},{board_id},,{pin},,{pin},{value}\n"
+                f"{source},{board_id},{device_id},{device_type},"
+                f"{pin},{pin_type},{value}\n"
             )
         f.write(
             f"{spec.n_pre_flow_frames + spec.n_post_flow_frames},"

tests/unit/analysis/test_events.py

@@ -59,7 +59,7 @@ def test_event_triggered_extracts_per_trial_window(tmp_path: Path):
     """An LED-on event at flow t=1500ms should yield a window of tracking
     frames spanning [-1000, 5000] ms around it."""
     spec = RecordingSpec(
-        extra_events=((1500.0, "output_write", "board0", "5", "1"),),
+        extra_events=((1500.0, "output_write", "board0", "led1", "LED", "5", "DIGITAL", "1"),),
     )
     write_synthetic_recording(tmp_path / "rec", spec)
     s = Session.load(tmp_path / "rec")
@@ -80,8 +80,8 @@ def test_event_triggered_extracts_per_trial_window(tmp_path: Path):
 def test_event_triggered_handles_multiple_trials(tmp_path: Path):
     spec = RecordingSpec(
         extra_events=(
-            (1000.0, "output_write", "board0", "5", "1"),
-            (2500.0, "output_write", "board0", "5", "1"),
+            (1000.0, "output_write", "board0", "led1", "LED", "5", "DIGITAL", "1"),
+            (2500.0, "output_write", "board0", "led1", "LED", "5", "DIGITAL", "1"),
         ),
     )
     write_synthetic_recording(tmp_path / "rec", spec)

tests/unit/analysis/test_io.py

@@ -6,6 +6,7 @@
 
 import pytest
 
+from glider.analysis import Session
 from glider.analysis._io import discover, parse_csv
 
 from .conftest import RecordingSpec, write_synthetic_recording
@@ -140,3 +141,31 @@ def test_parse_csv_returns_empty_metadata_when_no_header(tmp_path: Path):
     # row, so it becomes a key with empty value.
     assert "GLIDER Tracking Data" in metadata
     assert len(df) == 1
+
+
+def test_extra_events_round_trip_device_columns(tmp_path: Path):
+    """The fixture's event rows must land in the columns its header names.
+
+    Regression: the writer emitted seven cells against a seven-column
+    header in the wrong order, so device_type received the pin number and
+    device_id was never populated at all.
+    """
+    from .conftest import RecordingSpec, write_synthetic_recording
+
+    write_synthetic_recording(
+        tmp_path / "rec",
+        RecordingSpec(
+            extra_events=((1500.0, "output_write", "board0", "led1", "LED", "5", "DIGITAL", "1"),)
+        ),
+    )
+    s = Session.load(tmp_path / "rec")
+    row = s.events[s.events["source"] == "output_write"].iloc[0]
+    assert row["board_id"] == "board0"
+    assert row["device_id"] == "led1"
+    assert row["device_type"] == "LED"
+    # The events CSV mixes this numeric-looking "5" with blank pin cells
+    # on the flow_marker rows in the same column, so pandas infers the
+    # column as float64 — it round-trips as 5.0, not the string "5".
+    assert row["pin"] == 5.0
+    assert row["pin_type"] == "DIGITAL"
+    assert str(row["value"]) == "1"

tests/unit/analysis/test_plots.py

@@ -113,7 +113,9 @@ def test_plot_event_triggered_smoke(tmp_path: Path):
     """Generate a recording with an LED event, slice around it, plot."""
     import matplotlib.pyplot as plt
 
-    spec = RecordingSpec(extra_events=((1500.0, "output_write", "board0", "5", "1"),))
+    spec = RecordingSpec(
+        extra_events=((1500.0, "output_write", "board0", "led1", "LED", "5", "DIGITAL", "1"),)
+    )
     write_synthetic_recording(tmp_path / "rec", spec)
     s = Session.load(tmp_path / "rec")
     eta = s.event_triggered(source="output_write", window_ms=(-500.0, 1000.0))