From 9f2ea986e5a17c5e53b0e4481e61b19cbcada292 Mon Sep 17 00:00:00 2001 From: Garrett Bradham Date: Thu, 10 Sep 2026 12:21:45 -0500 Subject: [PATCH 01/12] test: give the synthetic event log its device columns back --- tests/unit/analysis/conftest.py | 19 +++++++++++------- tests/unit/analysis/test_events.py | 6 +++--- tests/unit/analysis/test_io.py | 31 ++++++++++++++++++++++++++++++ tests/unit/analysis/test_plots.py | 4 +++- 4 files changed, 49 insertions(+), 11 deletions(-) diff --git a/tests/unit/analysis/conftest.py b/tests/unit/analysis/conftest.py index 67b82560..7b671ef5 100644 --- a/tests/unit/analysis/conftest.py +++ b/tests/unit/analysis/conftest.py @@ -54,11 +54,13 @@ 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 @@ -248,13 +250,16 @@ 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}," diff --git a/tests/unit/analysis/test_events.py b/tests/unit/analysis/test_events.py index 8e985a74..f065bf10 100644 --- a/tests/unit/analysis/test_events.py +++ b/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) diff --git a/tests/unit/analysis/test_io.py b/tests/unit/analysis/test_io.py index 726f4aee..715dfbac 100644 --- a/tests/unit/analysis/test_io.py +++ b/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,33 @@ 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" diff --git a/tests/unit/analysis/test_plots.py b/tests/unit/analysis/test_plots.py index d91fd718..e4c3ea3e 100644 --- a/tests/unit/analysis/test_plots.py +++ b/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)) From 71b8acfb94ccc4a5dbef768629086d958ed6a61a Mon Sep 17 00:00:00 2001 From: Garrett Bradham Date: Thu, 10 Sep 2026 12:28:36 -0500 Subject: [PATCH 02/12] feat(analysis): map frames to flow time from the tracking CSV --- src/glider/analysis/timeline.py | 104 +++++++++++++++++++++++++++ tests/unit/analysis/test_timeline.py | 66 +++++++++++++++++ 2 files changed, 170 insertions(+) create mode 100644 src/glider/analysis/timeline.py create mode 100644 tests/unit/analysis/test_timeline.py diff --git a/src/glider/analysis/timeline.py b/src/glider/analysis/timeline.py new file mode 100644 index 00000000..76d69842 --- /dev/null +++ b/src/glider/analysis/timeline.py @@ -0,0 +1,104 @@ +"""One session as lanes on a single time axis. + +The event log has always held everything a hardware raster needs — every +pin edge and every commanded write, timestamped against a session epoch +the other recorders share. Nothing had drawn it. This builds the lanes; +:mod:`glider.gui.widgets.timeline_bar` draws them. + +Qt-free on purpose, in the same way :mod:`glider.analysis.behavior.session_view` +is: the axis arithmetic and the lane building are the parts worth testing, +and neither needs a display. + +Everything here is in **flow-relative milliseconds** — t=0 is +StartExperiment, matching what an analyst already reasons in. Events that +predate flow start (device initialisation, which is exactly when a rig is +most likely to be left in the wrong state) carry negative times and are +drawn rather than clipped. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +import numpy as np + +if TYPE_CHECKING: + from glider.analysis.session import Session + +__all__ = ["FrameMap", "build_frame_map"] + + +@dataclass(frozen=True) +class FrameMap: + """Frame index to flow-relative milliseconds, and back. + + Args: + frames: Frame indices, sorted ascending, one per known sample. + ms: The flow-relative time of each of those frames. + source: ``"tracking"`` for the empirical per-frame map, + ``"frame_rate"`` for the nominal-rate estimate. + """ + + frames: np.ndarray + ms: np.ndarray + source: str + + def ms_of(self, frame: int) -> float: + """When ``frame`` happened. Clamps outside the known range.""" + return float(np.interp(float(frame), self.frames, self.ms)) + + def frame_at(self, ms: float) -> int: + """The frame nearest ``ms``. Clamps outside the known range.""" + if len(self.frames) == 0: + return 0 + idx = int(np.searchsorted(self.ms, float(ms))) + idx = max(0, min(idx, len(self.frames) - 1)) + return int(self.frames[idx]) + + +def build_frame_map(session: Session, flow_offset_ms: float = 0.0) -> FrameMap | None: + """The frame/time mapping for a session, or None if it has neither. + + Resolved in order: + + 1. The tracking CSV's own ``frame`` and ``elapsed_ms`` columns. This is + an empirical map and stays correct across dropped frames, which a + nominal-rate calculation does not — a dropped frame shifts every + later frame by one in the nominal version and by nothing here. + 2. :attr:`Session.frame_rate` against the frame index, for a recording + whose tracking is too thin for (1). + 3. Neither: ``None``. The caller draws no hardware lanes, which is + correct rather than degraded — an ethogram-only session has no + hardware data to place on a time axis. + + Args: + session: The loaded recording. + flow_offset_ms: Session-elapsed ms of flow start, subtracted from + every time so the result is flow-relative. + """ + tracking = session.tracking + if not tracking.empty and {"frame", "elapsed_ms"}.issubset(tracking.columns): + pairs = tracking[["frame", "elapsed_ms"]].dropna() + # Several objects in one frame share a timestamp; one row per frame. + pairs = pairs.drop_duplicates(subset="frame").sort_values("frame") + if len(pairs) >= 2: + return FrameMap( + frames=pairs["frame"].to_numpy(dtype=float), + ms=pairs["elapsed_ms"].to_numpy(dtype=float) - flow_offset_ms, + source="tracking", + ) + + fps = session.frame_rate + if fps: + last = 1.0 + if not tracking.empty and "frame" in tracking.columns: + last = max(1.0, float(tracking["frame"].max())) + frames = np.array([0.0, last]) + return FrameMap( + frames=frames, + ms=frames / fps * 1000.0 - flow_offset_ms, + source="frame_rate", + ) + + return None diff --git a/tests/unit/analysis/test_timeline.py b/tests/unit/analysis/test_timeline.py new file mode 100644 index 00000000..3725855c --- /dev/null +++ b/tests/unit/analysis/test_timeline.py @@ -0,0 +1,66 @@ +"""Tests for the Qt-free timeline model.""" + +from __future__ import annotations + +from pathlib import Path + +import numpy as np +import pytest + +from glider.analysis import Session +from glider.analysis.timeline import FrameMap, build_frame_map + +from .conftest import RecordingSpec, write_synthetic_recording + + +def test_frame_map_comes_from_tracking(synthetic_recording: Path): + s = Session.load(synthetic_recording) + fm = build_frame_map(s) + assert fm is not None + assert fm.source == "tracking" + + +def test_frame_map_round_trips_a_frame(synthetic_recording: Path): + s = Session.load(synthetic_recording) + fm = build_frame_map(s) + # Any frame the map covers must survive frame -> ms -> frame. + frame = int(fm.frames[len(fm.frames) // 2]) + assert fm.frame_at(fm.ms_of(frame)) == frame + + +def test_frame_map_applies_the_flow_offset(synthetic_recording: Path): + s = Session.load(synthetic_recording) + plain = build_frame_map(s) + shifted = build_frame_map(s, flow_offset_ms=1000.0) + frame = int(plain.frames[0]) + assert shifted.ms_of(frame) == pytest.approx(plain.ms_of(frame) - 1000.0) + + +def test_frame_map_survives_a_dropped_frame(tmp_path: Path): + """A gap in the frame column must not shift later frames. + + This is the whole reason the map is built from the tracking CSV + rather than from a nominal frame rate: dropping frame 40 means every + later frame's true time is unchanged, but frame_index / fps would + place them all one frame early. + """ + directory = write_synthetic_recording(tmp_path / "rec", RecordingSpec()) + tracking = next(p for p in directory.glob("*_tracking.csv")) + lines = tracking.read_text(encoding="utf-8").splitlines(keepends=True) + kept = [ln for ln in lines if not ln.startswith("40,")] + assert len(kept) < len(lines), "fixture changed: no frame 40 row to drop" + tracking.write_text("".join(kept), encoding="utf-8") + + s = Session.load(directory) + fm = build_frame_map(s) + assert 40 not in set(fm.frames.astype(int)) + # Frame 41 keeps its real time rather than sliding into 40's slot. + assert fm.ms_of(41) == pytest.approx(41 / 30.0 * 1000.0, abs=5.0) + + +def test_frame_map_is_none_without_tracking(tmp_path: Path): + directory = write_synthetic_recording( + tmp_path / "rec", RecordingSpec(write_tracking=False) + ) + s = Session.load(directory) + assert build_frame_map(s) is None From 7a7b98b67851e425148147bb9d4fb601b9dcbb07 Mon Sep 17 00:00:00 2001 From: Garrett Bradham Date: Thu, 10 Sep 2026 12:31:49 -0500 Subject: [PATCH 03/12] fix(test): correct frame timing assertion in test_frame_map_survives_a_dropped_frame --- tests/unit/analysis/test_timeline.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/tests/unit/analysis/test_timeline.py b/tests/unit/analysis/test_timeline.py index 3725855c..676f77ef 100644 --- a/tests/unit/analysis/test_timeline.py +++ b/tests/unit/analysis/test_timeline.py @@ -54,8 +54,13 @@ def test_frame_map_survives_a_dropped_frame(tmp_path: Path): s = Session.load(directory) fm = build_frame_map(s) assert 40 not in set(fm.frames.astype(int)) - # Frame 41 keeps its real time rather than sliding into 40's slot. - assert fm.ms_of(41) == pytest.approx(41 / 30.0 * 1000.0, abs=5.0) + # The fixture writes 1-based frame numbers against 0-based elapsed time + # (conftest.py: `frame = i + 1`, `elapsed_ms = i / fps * 1000`), so frame + # 41 sits at 40/30 s. Do not "correct" this to 41/30. A nominal-fps + # implementation would place frame 41 at 1300.0 ms (41/30), which is + # 33 ms away — outside the 5 ms tolerance — so this assertion catches + # the regression it names. + assert fm.ms_of(41) == pytest.approx(40 / 30.0 * 1000.0, abs=5.0) def test_frame_map_is_none_without_tracking(tmp_path: Path): From 2635329402b7ecd804d4bbba33a5cdd88f62694c Mon Sep 17 00:00:00 2001 From: Garrett Bradham Date: Thu, 10 Sep 2026 12:46:22 -0500 Subject: [PATCH 04/12] feat(analysis): build hardware lanes from the event log --- src/glider/analysis/timeline.py | 166 ++++++++++++++++++++++++++- tests/unit/analysis/test_timeline.py | 115 +++++++++++++++++++ 2 files changed, 280 insertions(+), 1 deletion(-) diff --git a/src/glider/analysis/timeline.py b/src/glider/analysis/timeline.py index 76d69842..e2f58140 100644 --- a/src/glider/analysis/timeline.py +++ b/src/glider/analysis/timeline.py @@ -26,7 +26,14 @@ if TYPE_CHECKING: from glider.analysis.session import Session -__all__ = ["FrameMap", "build_frame_map"] +__all__ = [ + "FrameMap", + "Lane", + "Marker", + "Segment", + "build_frame_map", + "hardware_lanes", +] @dataclass(frozen=True) @@ -102,3 +109,160 @@ def build_frame_map(session: Session, flow_offset_ms: float = 0.0) -> FrameMap | ) return None + + +#: Full-scale value per pin type, for turning a written value into a bar +#: height. Normalising by each lane's own observed maximum instead would be +#: marginally cheaper and would draw a PWM that never exceeded 10 as full +#: brightness — same amount of code, wrong picture. +_PIN_FULL_SCALE = { + "DIGITAL": 1.0, + "PWM": 255.0, + "SERVO": 180.0, + "ANALOG": 1023.0, +} + + +@dataclass(frozen=True) +class Segment: + """A device holding one value over a span of time.""" + + start_ms: float + end_ms: float + value: float + level: float # 0.0-1.0, for bar height + + +@dataclass(frozen=True) +class Marker: + """An instant with no level — a non-numeric event value.""" + + at_ms: float + label: str + + +@dataclass(frozen=True) +class Lane: + """One row of the raster.""" + + key: str + label: str + board_id: str + segments: list[Segment] + markers: list[Marker] + + +def _cell(value) -> str: + """A CSV cell as text. + + Two shapes have to be flattened. Empty cells arrive as NaN rather than + "". And a column that mixes blanks with numbers — `pin` does, because + flow_marker rows leave it empty — is inferred as float64, so pin 7 + arrives as 7.0 and would otherwise name a lane "board0:pin7.0". + """ + if value is None: + return "" + if isinstance(value, float): + if value != value: # NaN + return "" + if value.is_integer(): + return str(int(value)) + text = str(value) + return "" if text in ("nan", "NaN", "", "None") else text.strip() + + +def _full_scale(pin_types: list[str], observed_max: float) -> float: + """The value that should draw as a full-height bar. + + The pin type's known range when there is one, except where the lane + actually exceeded it — a 12-bit board reads an ANALOG pin to 4095, and + clipping that at the 10-bit 1023 would draw the whole session as one + saturated row. + """ + for pin_type in pin_types: + full = _PIN_FULL_SCALE.get(pin_type.upper()) + if full is not None: + return full if observed_max <= full else observed_max + return observed_max if observed_max > 0 else 1.0 + + +def hardware_lanes( + session: Session, + flow_offset_ms: float = 0.0, + end_ms: float | None = None, +) -> list[Lane]: + """One lane per device, from the event log. + + Each event sets its device's value and that value holds until the + device's next event — a zero-order hold. One rule covers digital, PWM + and servo: a digital pin gives full-height blocks and a PWM ramp gives + stepped ones, without a branch per pin type. + + Args: + session: The loaded recording. + flow_offset_ms: Session-elapsed ms of flow start, subtracted from + every event time. + end_ms: Where the last held value stops. Defaults to the last + event's own time, which draws it as zero-width. + """ + events = session.events + if events.empty: + return [] + + rows = events[events["source"] != "flow_marker"].copy() + if rows.empty: + return [] + + rows["_ms"] = rows["elapsed_ms"].astype(float) - flow_offset_ms + rows["_device"] = [_cell(v) for v in rows["device_id"]] + rows["_board"] = [_cell(v) for v in rows["board_id"]] + rows["_pin"] = [_cell(v) for v in rows["pin"]] + rows["_pin_type"] = [_cell(v) for v in rows["pin_type"]] + # A board-level write with no resolved device still deserves a row. + rows["_key"] = [ + device or f"{board}:pin{pin}" + for device, board, pin in zip(rows["_device"], rows["_board"], rows["_pin"]) + ] + rows = rows.sort_values("_ms", kind="stable") + + tail = end_ms if end_ms is not None else float(rows["_ms"].max()) + + lanes: list[Lane] = [] + for key, group in rows.groupby("_key", sort=False): + times = group["_ms"].to_numpy(dtype=float) + markers: list[Marker] = [] + levels: list[tuple[float, float]] = [] # (ms, numeric value) + + for ms, raw in zip(times, group["value"]): + text = _cell(raw) + try: + levels.append((float(ms), float(text))) + except ValueError: + markers.append(Marker(at_ms=float(ms), label=text)) + + observed_max = max((v for _, v in levels), default=0.0) + full = _full_scale(list(dict.fromkeys(group["_pin_type"])), observed_max) + + segments = [ + Segment( + start_ms=ms, + end_ms=(levels[i + 1][0] if i + 1 < len(levels) else tail), + value=value, + level=max(0.0, min(1.0, value / full)), + ) + for i, (ms, value) in enumerate(levels) + ] + + label = _cell(group["_device"].iloc[0]) or str(key) + lanes.append( + Lane( + key=str(key), + label=label, + board_id=_cell(group["_board"].iloc[0]), + segments=segments, + markers=markers, + ) + ) + + lanes.sort(key=lambda lane: (lane.board_id, lane.key)) + return lanes diff --git a/tests/unit/analysis/test_timeline.py b/tests/unit/analysis/test_timeline.py index 676f77ef..53b9a0fa 100644 --- a/tests/unit/analysis/test_timeline.py +++ b/tests/unit/analysis/test_timeline.py @@ -69,3 +69,118 @@ def test_frame_map_is_none_without_tracking(tmp_path: Path): ) s = Session.load(directory) assert build_frame_map(s) is None + + +from glider.analysis.timeline import Lane, Segment, hardware_lanes + + +def _rec(tmp_path: Path, events, name="rec") -> Path: + return write_synthetic_recording( + tmp_path / name, RecordingSpec(extra_events=tuple(events)) + ) + + +def test_hardware_lane_per_device(tmp_path: Path): + directory = _rec( + tmp_path, + [ + (500.0, "output_write", "board0", "led1", "LED", "5", "DIGITAL", "1"), + (600.0, "output_write", "board0", "pump", "Pump", "6", "DIGITAL", "1"), + ], + ) + lanes = hardware_lanes(Session.load(directory)) + assert {lane.key for lane in lanes} == {"led1", "pump"} + + +def test_value_holds_until_the_next_event(tmp_path: Path): + """A digital pin written high at 500ms and low at 1500ms is on for 1s.""" + directory = _rec( + tmp_path, + [ + (500.0, "output_write", "board0", "led1", "LED", "5", "DIGITAL", "1"), + (1500.0, "output_write", "board0", "led1", "LED", "5", "DIGITAL", "0"), + ], + ) + s = Session.load(directory) + # Times are flow-relative, and the fixture has a pre-flow period, so the + # offset has to come from the flow marker — the same value build_timeline + # will compute once and hand to both builders. + marker = s.events[(s.events["source"] == "flow_marker") & (s.events["value"] == "start")] + flow_offset_ms = float(marker["elapsed_ms"].iloc[0]) + lane = hardware_lanes(s, flow_offset_ms=flow_offset_ms)[0] + on = [seg for seg in lane.segments if seg.value > 0] + assert len(on) == 1 + assert on[0].start_ms == pytest.approx(500.0, abs=2.0) + assert on[0].end_ms == pytest.approx(1500.0, abs=2.0) + assert on[0].level == pytest.approx(1.0) + + +def test_pwm_normalises_against_full_scale(tmp_path: Path): + """128 on a PWM pin is half height, not full — per-lane max would + draw a device that never exceeded 128 as though it were saturated.""" + directory = _rec( + tmp_path, + [ + (500.0, "output_write", "board0", "fan", "Fan", "9", "PWM", "128"), + (1500.0, "output_write", "board0", "fan", "Fan", "9", "PWM", "0"), + ], + ) + lane = hardware_lanes(Session.load(directory))[0] + assert lane.segments[0].level == pytest.approx(128.0 / 255.0) + + +def test_analog_beyond_assumed_range_falls_back_to_observed(tmp_path: Path): + """A 12-bit board reads to 4095. Clipping at the 10-bit 1023 would + draw the whole session as one saturated row.""" + directory = _rec( + tmp_path, + [ + (500.0, "input_change", "board0", "ldr", "Photoresistor", "0", "ANALOG", "4095"), + (1500.0, "input_change", "board0", "ldr", "Photoresistor", "0", "ANALOG", "2048"), + ], + ) + lane = hardware_lanes(Session.load(directory))[0] + assert lane.segments[0].level == pytest.approx(1.0) + assert lane.segments[1].level == pytest.approx(2048.0 / 4095.0) + + +def test_non_numeric_value_becomes_a_marker(tmp_path: Path): + directory = _rec( + tmp_path, + [(500.0, "input_change", "board0", "reader", "RFID", "2", "", "tag-A7"),], + ) + lane = hardware_lanes(Session.load(directory))[0] + assert lane.segments == [] + assert [m.label for m in lane.markers] == ["tag-A7"] + + +def test_event_without_a_frame_still_makes_a_segment(tmp_path: Path): + """Device-init writes land before the first camera frame, so their + `frame` cell is empty. They are drawn from their timestamp.""" + directory = _rec( + tmp_path, + [(500.0, "output_write", "board0", "led1", "LED", "5", "DIGITAL", "1")], + ) + s = Session.load(directory) + s.events.loc[s.events["source"] == "output_write", "frame"] = np.nan + assert hardware_lanes(s)[0].segments + + +def test_flow_markers_are_not_a_lane(synthetic_recording: Path): + lanes = hardware_lanes(Session.load(synthetic_recording)) + assert all(lane.key != "flow_marker" for lane in lanes) + + +def test_no_events_gives_no_lanes(tmp_path: Path): + directory = write_synthetic_recording( + tmp_path / "rec", RecordingSpec(write_events=False) + ) + assert hardware_lanes(Session.load(directory)) == [] + + +def test_device_id_falls_back_to_board_and_pin(tmp_path: Path): + directory = _rec( + tmp_path, + [(500.0, "output_write", "board0", "", "", "7", "DIGITAL", "1")], + ) + assert hardware_lanes(Session.load(directory))[0].key == "board0:pin7" From 4177a211e98e8fa2e4e52f74db6a2f48d91d9e24 Mon Sep 17 00:00:00 2001 From: Garrett Bradham Date: Thu, 10 Sep 2026 13:00:53 -0500 Subject: [PATCH 05/12] fix(analysis): guard segment spans, empty values, and lint --- src/glider/analysis/timeline.py | 15 +++++-- tests/unit/analysis/conftest.py | 13 ++++-- tests/unit/analysis/test_io.py | 4 +- tests/unit/analysis/test_timeline.py | 60 ++++++++++++++++++++-------- 4 files changed, 66 insertions(+), 26 deletions(-) diff --git a/src/glider/analysis/timeline.py b/src/glider/analysis/timeline.py index e2f58140..687018bf 100644 --- a/src/glider/analysis/timeline.py +++ b/src/glider/analysis/timeline.py @@ -221,7 +221,7 @@ def hardware_lanes( # A board-level write with no resolved device still deserves a row. rows["_key"] = [ device or f"{board}:pin{pin}" - for device, board, pin in zip(rows["_device"], rows["_board"], rows["_pin"]) + for device, board, pin in zip(rows["_device"], rows["_board"], rows["_pin"], strict=True) ] rows = rows.sort_values("_ms", kind="stable") @@ -233,8 +233,12 @@ def hardware_lanes( markers: list[Marker] = [] levels: list[tuple[float, float]] = [] # (ms, numeric value) - for ms, raw in zip(times, group["value"]): + for ms, raw in zip(times, group["value"], strict=True): text = _cell(raw) + if not text: + # The event logger writes "" for a None value, not a + # missing cell — neither a level nor a marker for that. + continue try: levels.append((float(ms), float(text))) except ValueError: @@ -246,7 +250,12 @@ def hardware_lanes( segments = [ Segment( start_ms=ms, - end_ms=(levels[i + 1][0] if i + 1 < len(levels) else tail), + # The last segment ends at `tail` unless that end would + # precede its own start — a session can legitimately end + # before its last event (camera stops before the flow + # tears down), and a negative span draws as an inverted + # or invisible rect. + end_ms=(levels[i + 1][0] if i + 1 < len(levels) else max(tail, ms)), value=value, level=max(0.0, min(1.0, value / full)), ) diff --git a/tests/unit/analysis/conftest.py b/tests/unit/analysis/conftest.py index 7b671ef5..3a2183de 100644 --- a/tests/unit/analysis/conftest.py +++ b/tests/unit/analysis/conftest.py @@ -250,9 +250,16 @@ 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, device_id, device_type, pin, pin_type, 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) diff --git a/tests/unit/analysis/test_io.py b/tests/unit/analysis/test_io.py index 715dfbac..e6aefef0 100644 --- a/tests/unit/analysis/test_io.py +++ b/tests/unit/analysis/test_io.py @@ -155,9 +155,7 @@ def test_extra_events_round_trip_device_columns(tmp_path: Path): write_synthetic_recording( tmp_path / "rec", RecordingSpec( - extra_events=( - (1500.0, "output_write", "board0", "led1", "LED", "5", "DIGITAL", "1"), - ) + extra_events=((1500.0, "output_write", "board0", "led1", "LED", "5", "DIGITAL", "1"),) ), ) s = Session.load(tmp_path / "rec") diff --git a/tests/unit/analysis/test_timeline.py b/tests/unit/analysis/test_timeline.py index 53b9a0fa..f20407e9 100644 --- a/tests/unit/analysis/test_timeline.py +++ b/tests/unit/analysis/test_timeline.py @@ -8,7 +8,7 @@ import pytest from glider.analysis import Session -from glider.analysis.timeline import FrameMap, build_frame_map +from glider.analysis.timeline import build_frame_map, hardware_lanes from .conftest import RecordingSpec, write_synthetic_recording @@ -64,20 +64,13 @@ def test_frame_map_survives_a_dropped_frame(tmp_path: Path): def test_frame_map_is_none_without_tracking(tmp_path: Path): - directory = write_synthetic_recording( - tmp_path / "rec", RecordingSpec(write_tracking=False) - ) + directory = write_synthetic_recording(tmp_path / "rec", RecordingSpec(write_tracking=False)) s = Session.load(directory) assert build_frame_map(s) is None -from glider.analysis.timeline import Lane, Segment, hardware_lanes - - def _rec(tmp_path: Path, events, name="rec") -> Path: - return write_synthetic_recording( - tmp_path / name, RecordingSpec(extra_events=tuple(events)) - ) + return write_synthetic_recording(tmp_path / name, RecordingSpec(extra_events=tuple(events))) def test_hardware_lane_per_device(tmp_path: Path): @@ -147,7 +140,9 @@ def test_analog_beyond_assumed_range_falls_back_to_observed(tmp_path: Path): def test_non_numeric_value_becomes_a_marker(tmp_path: Path): directory = _rec( tmp_path, - [(500.0, "input_change", "board0", "reader", "RFID", "2", "", "tag-A7"),], + [ + (500.0, "input_change", "board0", "reader", "RFID", "2", "", "tag-A7"), + ], ) lane = hardware_lanes(Session.load(directory))[0] assert lane.segments == [] @@ -166,15 +161,20 @@ def test_event_without_a_frame_still_makes_a_segment(tmp_path: Path): assert hardware_lanes(s)[0].segments -def test_flow_markers_are_not_a_lane(synthetic_recording: Path): - lanes = hardware_lanes(Session.load(synthetic_recording)) - assert all(lane.key != "flow_marker" for lane in lanes) +def test_flow_markers_are_not_a_lane(tmp_path: Path): + """Flow boundaries are drawn as rules across every lane, not as a lane.""" + directory = _rec( + tmp_path, + [(500.0, "output_write", "board0", "led1", "LED", "5", "DIGITAL", "1")], + ) + lanes = hardware_lanes(Session.load(directory)) + assert {lane.key for lane in lanes} == {"led1"} + labels = {m.label for lane in lanes for m in lane.markers} + assert not labels & {"start", "end"} def test_no_events_gives_no_lanes(tmp_path: Path): - directory = write_synthetic_recording( - tmp_path / "rec", RecordingSpec(write_events=False) - ) + directory = write_synthetic_recording(tmp_path / "rec", RecordingSpec(write_events=False)) assert hardware_lanes(Session.load(directory)) == [] @@ -184,3 +184,29 @@ def test_device_id_falls_back_to_board_and_pin(tmp_path: Path): [(500.0, "output_write", "board0", "", "", "7", "DIGITAL", "1")], ) assert hardware_lanes(Session.load(directory))[0].key == "board0:pin7" + + +def test_segment_never_ends_before_it_starts(tmp_path: Path): + """A session that ends before its last event must not make a negative span. + + The camera can stop before the flow tears down, so `end_ms` legitimately + precedes a late event. A renderer computing `end - start` would draw an + inverted rect.""" + directory = _rec( + tmp_path, + [(500.0, "output_write", "board0", "led1", "LED", "5", "DIGITAL", "1")], + ) + lane = hardware_lanes(Session.load(directory), end_ms=0.0)[0] + assert all(seg.end_ms >= seg.start_ms for seg in lane.segments) + + +def test_empty_value_is_neither_level_nor_marker(tmp_path: Path): + """The event logger writes "" when a device reports no value.""" + directory = _rec( + tmp_path, + [(500.0, "input_change", "board0", "probe", "Probe", "3", "", "")], + ) + lanes = hardware_lanes(Session.load(directory)) + lane = next(ln for ln in lanes if ln.key == "probe") + assert lane.segments == [] + assert lane.markers == [] From 1696b0f35946f8426a96ddbaa0851e02e9ed999e Mon Sep 17 00:00:00 2001 From: Garrett Bradham Date: Thu, 10 Sep 2026 13:17:37 -0500 Subject: [PATCH 06/12] feat(analysis): assemble a session into one timeline --- src/glider/analysis/timeline.py | 124 +++++++++++++++++++++++++++ tests/unit/analysis/test_timeline.py | 41 ++++++++- 2 files changed, 164 insertions(+), 1 deletion(-) diff --git a/src/glider/analysis/timeline.py b/src/glider/analysis/timeline.py index 687018bf..bd09b2d4 100644 --- a/src/glider/analysis/timeline.py +++ b/src/glider/analysis/timeline.py @@ -24,14 +24,18 @@ import numpy as np if TYPE_CHECKING: + from glider.analysis.behavior.session_view import SessionView from glider.analysis.session import Session __all__ = [ + "BehaviorLane", "FrameMap", "Lane", "Marker", "Segment", + "Timeline", "build_frame_map", + "build_timeline", "hardware_lanes", ] @@ -275,3 +279,123 @@ def hardware_lanes( lanes.sort(key=lambda lane: (lane.board_id, lane.key)) return lanes + + +@dataclass(frozen=True) +class BehaviorLane: + """Per-frame behaviour labels, and where they came from. + + Deliberately not converted to :class:`Segment` runs. The bar resolves + behaviour per *pixel column* by majority rather than drawing one rect + per run, because a five-minute session holds ~9000 rows against ~2000 + pixels and sub-pixel rects blend by coverage — which is what made + every colour on the old bar an average of several behaviours. + """ + + source: str # "ethogram" | "tracking" + labels: list[str] + frames: np.ndarray + + +@dataclass(frozen=True) +class Timeline: + """Everything drawable about one session, on one flow-relative axis.""" + + lanes: list[Lane] + behavior: list[BehaviorLane] + frame_map: FrameMap | None + flow_start_ms: float | None + flow_end_ms: float | None + start_ms: float + end_ms: float + + +def _flow_elapsed(session: Session, marker: str) -> float | None: + """Session-elapsed ms of a flow marker, or None if it never fired.""" + events = session.events + if events.empty: + return None + hit = events[(events["source"] == "flow_marker") & (events["value"] == marker)] + if hit.empty: + return None + return float(hit["elapsed_ms"].iloc[0]) + + +def build_timeline(session: Session | None, view: SessionView | None = None) -> Timeline: + """Assemble a session into lanes on one axis. + + Args: + session: A loaded recording, or None for an ethogram-only view. + view: An optional :class:`~glider.analysis.behavior.session_view.SessionView`, + contributing the classifier ethogram as its own behaviour lane. + + A live recording carries ``behavioral_state`` in its tracking CSV and a + behaviour apply-run produces a classifier ethogram. These are different + things at different quality, so when both are present they stay two + lanes. Collapsing them would misrepresent provenance. + """ + behavior: list[BehaviorLane] = [] + if view is not None and len(view.labels): + behavior.append( + BehaviorLane(source="ethogram", labels=list(view.labels), frames=view.frames) + ) + + if session is None: + return Timeline( + lanes=[], + behavior=behavior, + frame_map=None, + flow_start_ms=None, + flow_end_ms=None, + start_ms=0.0, + end_ms=0.0, + ) + + flow_start_elapsed = _flow_elapsed(session, "start") + flow_end_elapsed = _flow_elapsed(session, "end") + offset = flow_start_elapsed or 0.0 + + frame_map = build_frame_map(session, offset) + + tracking = session.tracking + if not tracking.empty and "behavioral_state" in tracking.columns: + per_frame = tracking[["frame", "behavioral_state"]].dropna() + per_frame = per_frame.drop_duplicates(subset="frame").sort_values("frame") + if len(per_frame): + behavior.append( + BehaviorLane( + source="tracking", + labels=[str(v) for v in per_frame["behavioral_state"]], + frames=per_frame["frame"].to_numpy(dtype=int), + ) + ) + + # The axis spans everything drawable. A session with device-init writes + # 30s before flow start shows those 30s: unlike a windowed ethogram's + # empty lead-in, a pre-flow region holds content, and it answers the + # most common question a hardware session raises — was a device already + # in the wrong state before the run began. + candidates_start = [0.0] + candidates_end = [0.0] + if frame_map is not None and len(frame_map.ms): + candidates_start.append(float(frame_map.ms[0])) + candidates_end.append(float(frame_map.ms[-1])) + if not session.events.empty: + event_ms = session.events["elapsed_ms"].astype(float) - offset + candidates_start.append(float(event_ms.min())) + candidates_end.append(float(event_ms.max())) + if flow_end_elapsed is not None: + candidates_end.append(flow_end_elapsed - offset) + + start_ms = min(candidates_start) + end_ms = max(candidates_end) + + return Timeline( + lanes=hardware_lanes(session, offset, end_ms), + behavior=behavior, + frame_map=frame_map, + flow_start_ms=None if flow_start_elapsed is None else 0.0, + flow_end_ms=None if flow_end_elapsed is None else flow_end_elapsed - offset, + start_ms=start_ms, + end_ms=end_ms, + ) diff --git a/tests/unit/analysis/test_timeline.py b/tests/unit/analysis/test_timeline.py index f20407e9..0fe21a92 100644 --- a/tests/unit/analysis/test_timeline.py +++ b/tests/unit/analysis/test_timeline.py @@ -8,7 +8,7 @@ import pytest from glider.analysis import Session -from glider.analysis.timeline import build_frame_map, hardware_lanes +from glider.analysis.timeline import build_frame_map, build_timeline, hardware_lanes from .conftest import RecordingSpec, write_synthetic_recording @@ -210,3 +210,42 @@ def test_empty_value_is_neither_level_nor_marker(tmp_path: Path): lane = next(ln for ln in lanes if ln.key == "probe") assert lane.segments == [] assert lane.markers == [] + + +def test_flow_start_is_time_zero(synthetic_recording: Path): + t = build_timeline(Session.load(synthetic_recording)) + assert t.flow_start_ms == pytest.approx(0.0, abs=1.0) + assert t.flow_end_ms is not None and t.flow_end_ms > 0 + + +def test_pre_flow_events_keep_negative_times(tmp_path: Path): + """Device-init writes happen before flow start and must stay visible. + + The default spec has 1s of pre-flow, so an event at flow_ms=-500 + cannot be expressed through extra_events; shift the whole axis instead + by asserting the timeline starts before zero because the tracking CSV + itself begins in the pre-flow period. + """ + t = build_timeline(Session.load(write_synthetic_recording(tmp_path / "rec", RecordingSpec()))) + assert t.start_ms < 0.0 + + +def test_behavior_lane_from_tracking(synthetic_recording: Path): + t = build_timeline(Session.load(synthetic_recording)) + sources = [lane.source for lane in t.behavior] + assert "tracking" in sources + + +def test_timeline_without_a_session_is_empty_but_valid(tmp_path: Path): + t = build_timeline(None) + assert t.lanes == [] + assert t.behavior == [] + assert t.frame_map is None + assert t.start_ms == 0.0 and t.end_ms == 0.0 + + +def test_no_flow_marker_leaves_boundaries_none(tmp_path: Path): + directory = write_synthetic_recording(tmp_path / "rec", RecordingSpec(write_events=False)) + t = build_timeline(Session.load(directory)) + assert t.flow_start_ms is None + assert t.flow_end_ms is None From c50ccc222950734684f81d7f37f45602dfae793d Mon Sep 17 00:00:00 2001 From: Garrett Bradham Date: Thu, 10 Sep 2026 13:35:42 -0500 Subject: [PATCH 07/12] fix(analysis): one behaviour lane per tracked object --- src/glider/analysis/timeline.py | 37 +++++++++++++++++++++------- tests/unit/analysis/conftest.py | 19 ++++++++++---- tests/unit/analysis/test_timeline.py | 30 ++++++++++++++++++++++ 3 files changed, 72 insertions(+), 14 deletions(-) diff --git a/src/glider/analysis/timeline.py b/src/glider/analysis/timeline.py index bd09b2d4..5916ee9e 100644 --- a/src/glider/analysis/timeline.py +++ b/src/glider/analysis/timeline.py @@ -359,16 +359,35 @@ def build_timeline(session: Session | None, view: SessionView | None = None) -> tracking = session.tracking if not tracking.empty and "behavioral_state" in tracking.columns: - per_frame = tracking[["frame", "behavioral_state"]].dropna() - per_frame = per_frame.drop_duplicates(subset="frame").sort_values("frame") - if len(per_frame): - behavior.append( - BehaviorLane( - source="tracking", - labels=[str(v) for v in per_frame["behavioral_state"]], - frames=per_frame["frame"].to_numpy(dtype=int), - ) + # A multi-subject recording has one tracking row per object per + # frame, all sharing a frame number. Deduping across objects (as a + # single global drop_duplicates would) silently keeps whichever + # object's row happens to sort first and discards the rest — so + # group by object_id *before* deduping, and only dedup within one + # object's own rows, where it is safe because a single object has + # exactly one row per frame. An older/hand-made CSV without an + # object_id column at all falls back to one lane, unchanged. + if "object_id" in tracking.columns: + object_ids = sorted(tracking["object_id"].dropna().unique()) + else: + object_ids = [None] + multi_object = len(object_ids) > 1 + + for object_id in object_ids: + obj_tracking = ( + tracking if object_id is None else tracking[tracking["object_id"] == object_id] ) + per_frame = obj_tracking[["frame", "behavioral_state"]].dropna() + per_frame = per_frame.drop_duplicates(subset="frame").sort_values("frame") + if len(per_frame): + source = "tracking" if not multi_object else f"tracking[{int(object_id)}]" + behavior.append( + BehaviorLane( + source=source, + labels=[str(v) for v in per_frame["behavioral_state"]], + frames=per_frame["frame"].to_numpy(dtype=int), + ) + ) # The axis spans everything drawable. A session with device-init writes # 30s before flow start shows those 30s: unlike a windowed ethogram's diff --git a/tests/unit/analysis/conftest.py b/tests/unit/analysis/conftest.py index 3a2183de..81ec4380 100644 --- a/tests/unit/analysis/conftest.py +++ b/tests/unit/analysis/conftest.py @@ -64,6 +64,10 @@ class RecordingSpec: 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 def _iso(dt: datetime) -> str: @@ -177,11 +181,16 @@ 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" - ) + 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") diff --git a/tests/unit/analysis/test_timeline.py b/tests/unit/analysis/test_timeline.py index 0fe21a92..c9db7875 100644 --- a/tests/unit/analysis/test_timeline.py +++ b/tests/unit/analysis/test_timeline.py @@ -249,3 +249,33 @@ def test_no_flow_marker_leaves_boundaries_none(tmp_path: Path): t = build_timeline(Session.load(directory)) assert t.flow_start_ms is None assert t.flow_end_ms is None + + +def test_multi_object_tracking_gets_one_lane_per_object(tmp_path: Path): + """Two tracked mice must not collapse to one object's behaviour. + + A shared frame column across objects makes a naive + drop_duplicates(subset="frame") keep only the first object's state, + silently discarding the second mouse's labels entirely. + """ + directory = write_synthetic_recording(tmp_path / "rec", RecordingSpec(n_objects=2)) + t = build_timeline(Session.load(directory)) + tracking_lanes = { + lane.source: lane for lane in t.behavior if lane.source.startswith("tracking") + } + assert set(tracking_lanes) == {"tracking[0]", "tracking[1]"} + obj0, obj1 = tracking_lanes["tracking[0]"], tracking_lanes["tracking[1]"] + assert len(obj0.labels) == len(obj1.labels) > 0 + # Object 1's labels are the fixture's "_obj1" — not object 0's + # labels repeated under a different source string. + assert obj0.labels != obj1.labels + assert all(label.endswith("_obj1") for label in obj1.labels) + + +def test_single_object_tracking_keeps_bare_source(synthetic_recording: Path): + """Task 5 depends on the single-object source staying the bare string + "tracking", not "tracking[0]".""" + t = build_timeline(Session.load(synthetic_recording)) + tracking_lanes = [lane for lane in t.behavior if lane.source == "tracking"] + assert len(tracking_lanes) == 1 + assert not any(lane.source.startswith("tracking[") for lane in t.behavior) From eb85f05d5ed91f0e3e8b7cb42d1b023d3baa19bb Mon Sep 17 00:00:00 2001 From: Garrett Bradham Date: Thu, 10 Sep 2026 13:51:28 -0500 Subject: [PATCH 08/12] fix(analysis): count behaviour lanes, not object ids --- src/glider/analysis/timeline.py | 36 +++++++++++++++------ tests/unit/analysis/conftest.py | 34 ++++++++++++++++++++ tests/unit/analysis/test_timeline.py | 48 ++++++++++++++++++++++++++++ 3 files changed, 109 insertions(+), 9 deletions(-) diff --git a/src/glider/analysis/timeline.py b/src/glider/analysis/timeline.py index 5916ee9e..8a44ce97 100644 --- a/src/glider/analysis/timeline.py +++ b/src/glider/analysis/timeline.py @@ -366,13 +366,27 @@ def build_timeline(session: Session | None, view: SessionView | None = None) -> # group by object_id *before* deduping, and only dedup within one # object's own rows, where it is safe because a single object has # exactly one row per frame. An older/hand-made CSV without an - # object_id column at all falls back to one lane, unchanged. + # object_id column at all falls back to one lane, unchanged — and + # so does one whose object_id column is present but entirely + # empty, since there is then nothing to group by. if "object_id" in tracking.columns: object_ids = sorted(tracking["object_id"].dropna().unique()) else: + object_ids = [] + if not object_ids: object_ids = [None] - multi_object = len(object_ids) > 1 + # "multi_object" is decided by how many objects actually produced + # a lane, not by how many distinct object_id values exist. + # tracking_logger.py writes object_id=-1 with a blank + # behavioral_state for motion-only/heartbeat rows; that id never + # survives the per-object dropna() below, but counting raw ids + # would still see it and wrongly rename a single real subject's + # lane to "tracking[0]". Counting surviving lanes instead needs no + # knowledge of sentinel values, so a future sentinel id — or a + # second object whose behavioral_state is entirely blank — can't + # reintroduce the same bug. + per_object_frames = [] for object_id in object_ids: obj_tracking = ( tracking if object_id is None else tracking[tracking["object_id"] == object_id] @@ -380,14 +394,18 @@ def build_timeline(session: Session | None, view: SessionView | None = None) -> per_frame = obj_tracking[["frame", "behavioral_state"]].dropna() per_frame = per_frame.drop_duplicates(subset="frame").sort_values("frame") if len(per_frame): - source = "tracking" if not multi_object else f"tracking[{int(object_id)}]" - behavior.append( - BehaviorLane( - source=source, - labels=[str(v) for v in per_frame["behavioral_state"]], - frames=per_frame["frame"].to_numpy(dtype=int), - ) + per_object_frames.append((object_id, per_frame)) + + multi_object = len(per_object_frames) > 1 + for object_id, per_frame in per_object_frames: + source = "tracking" if not multi_object else f"tracking[{int(object_id)}]" + behavior.append( + BehaviorLane( + source=source, + labels=[str(v) for v in per_frame["behavioral_state"]], + frames=per_frame["frame"].to_numpy(dtype=int), ) + ) # The axis spans everything drawable. A session with device-init writes # 30s before flow start shows those 30s: unlike a windowed ethogram's diff --git a/tests/unit/analysis/conftest.py b/tests/unit/analysis/conftest.py index 81ec4380..a826588e 100644 --- a/tests/unit/analysis/conftest.py +++ b/tests/unit/analysis/conftest.py @@ -68,6 +68,13 @@ class RecordingSpec: # 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: @@ -181,6 +188,33 @@ def _write_tracking_csv( state = "unknown" zone_ids = "" velocity = 0.0 + 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 diff --git a/tests/unit/analysis/test_timeline.py b/tests/unit/analysis/test_timeline.py index c9db7875..7f34b327 100644 --- a/tests/unit/analysis/test_timeline.py +++ b/tests/unit/analysis/test_timeline.py @@ -279,3 +279,51 @@ def test_single_object_tracking_keeps_bare_source(synthetic_recording: Path): tracking_lanes = [lane for lane in t.behavior if lane.source == "tracking"] assert len(tracking_lanes) == 1 assert not any(lane.source.startswith("tracking[") for lane in t.behavior) + + +def test_single_object_with_heartbeat_row_keeps_bare_source(tmp_path: Path): + """tracking_logger.py writes object_id=-1 with a blank behavioral_state + for its motion-only/heartbeat rows, and the frame-1 heartbeat is the + common case (fires whenever nothing has been detected yet), not a rare + corner case. A single real subject plus that sentinel row must still + yield exactly one behaviour lane named the bare "tracking" — counting + distinct object_id values instead of lanes-with-content would see two + ids (0 and -1) and wrongly rename it "tracking[0]".""" + directory = write_synthetic_recording( + tmp_path / "rec", RecordingSpec(include_heartbeat_row=True) + ) + t = build_timeline(Session.load(directory)) + tracking_lanes = [lane for lane in t.behavior if lane.source == "tracking"] + assert len(tracking_lanes) == 1 + assert not any(lane.source.startswith("tracking[") for lane in t.behavior) + + +def test_tracking_with_empty_object_id_column_falls_back_to_one_lane(tmp_path: Path): + """A hand-made or imported tracking CSV (this feature explicitly + supports those) can have an object_id column that is present but + entirely blank. That must fall back to the same ungrouped single lane + as a CSV with no object_id column at all — not silently drop the + behaviour lane, which is what an empty `object_ids` list would do.""" + directory = write_synthetic_recording(tmp_path / "rec", RecordingSpec()) + tracking_path = next(directory.glob("*_tracking.csv")) + lines = tracking_path.read_text(encoding="utf-8").splitlines(keepends=True) + + out = [] + in_data = False + for line in lines: + if line.startswith("frame,timestamp,"): + out.append(line) + in_data = True + continue + if in_data and line.strip(): + fields = line.rstrip("\n").split(",") + fields[4] = "" # object_id + out.append(",".join(fields) + "\n") + continue + in_data = False + out.append(line) + tracking_path.write_text("".join(out), encoding="utf-8") + + t = build_timeline(Session.load(directory)) + tracking_lanes = [lane for lane in t.behavior if lane.source == "tracking"] + assert len(tracking_lanes) == 1 From 84f84d9b8421a1005de4b9d8512ceef7cadbf1b7 Mon Sep 17 00:00:00 2001 From: Garrett Bradham Date: Thu, 10 Sep 2026 13:57:42 -0500 Subject: [PATCH 09/12] feat(gui): draw the session as a multi-lane timeline Adds TimelineBar, the widget that replaces the single-lane ethogram bar with a stack of lanes sharing one time axis and one playhead: the ethogram becomes one lane, the hardware the rig drove becomes the lanes below it. The widget's internal axis is time, but selection_changed still emits frames so every stats table below it keeps working unmodified. Moves behavior_qcolor and behavior_order out of analysis_window.py into timeline_bar.py (analysis_window imports them back) to avoid a circular import once analysis_window pulls in TimelineBar in a later task. --- src/glider/gui/behavior/analysis_window.py | 33 +- src/glider/gui/widgets/timeline_bar.py | 436 +++++++++++++++++++ tests/unit/gui/behavior/test_timeline_bar.py | 129 ++++++ 3 files changed, 566 insertions(+), 32 deletions(-) create mode 100644 src/glider/gui/widgets/timeline_bar.py create mode 100644 tests/unit/gui/behavior/test_timeline_bar.py diff --git a/src/glider/gui/behavior/analysis_window.py b/src/glider/gui/behavior/analysis_window.py index 6f21c735..035060bc 100644 --- a/src/glider/gui/behavior/analysis_window.py +++ b/src/glider/gui/behavior/analysis_window.py @@ -46,6 +46,7 @@ from glider.analysis.behavior.session_view import SessionView, SessionViewError from glider.gui.styles import colors +from glider.gui.widgets.timeline_bar import behavior_order, behavior_qcolor from glider.gui.widgets.tool_ui import ( CARD_GAP, GUTTER, @@ -154,38 +155,6 @@ def _vrule() -> QFrame: return line -def behavior_qcolor(name: str, order: list[str] | None = None) -> QColor: - """The colour the annotated video would have drawn this behaviour in. - - Shared with the overlay so a bout looks the same wherever it is shown; - blank (unscored) frames read as background rather than a colour. - - ``order`` is the behaviours present, which is what makes the colours - reliably *different*. Without it the palette slot comes from a hash of the - name, and a hash has no reason to avoid collisions: two behaviours in one - session could land on the same colour, and neighbouring ones routinely - landed on adjacent hues. Given the session's own label set, the first N - palette entries are handed out in order, and N distinct behaviours get N - distinct colours. - """ - if not name: - return QColor(colors.BORDER) - from glider.analysis.behavior.classify.overlay import color_for_behavior - - b, g, r = color_for_behavior(name, order) - return QColor(r, g, b) - - -def behavior_order(labels) -> list[str]: - """The behaviours present, in a stable order. - - Sorted rather than first-appearance: the same cohort scored twice must - colour the same behaviour the same way, and first-appearance makes that - depend on which animal happened to groom first. - """ - return sorted({label for label in labels if label}) - - class EthogramBar(QWidget): """The ethogram as a timeline: bands to read, and the scrubber to drag. diff --git a/src/glider/gui/widgets/timeline_bar.py b/src/glider/gui/widgets/timeline_bar.py new file mode 100644 index 00000000..40cf85ef --- /dev/null +++ b/src/glider/gui/widgets/timeline_bar.py @@ -0,0 +1,436 @@ +"""The session as a stack of lanes, and the one scrubber for all of them. + +Replaces the ethogram bar, which was a timeline of exactly one thing. The +ethogram becomes a lane; the hardware the rig drove becomes the lanes below +it; both share an axis and a playhead. + +The axis here is **time**, because the event log is timed and its rows carry +no usable frame index before the first camera frame arrives. But +``selection_changed`` still emits **frames**, because every table in Session +Review consumes a frame range — ``zone_rows(start, end)``, +``cohort_rows(start, end)``, ``segment_stats(start, end)``. Converting at the +signal boundary is what lets the whole panel below stay untouched. + +The model is in :mod:`glider.analysis.timeline` and is Qt-free. This only +draws. +""" + +from __future__ import annotations + +import numpy as np +from PyQt6.QtCore import QRectF, QSize, Qt, pyqtSignal +from PyQt6.QtGui import QColor, QPainter, QPen, QPixmap +from PyQt6.QtWidgets import QSizePolicy, QWidget + +from glider.analysis.timeline import BehaviorLane, Timeline +from glider.gui.styles import colors + +__all__ = ["TimelineBar", "behavior_order", "behavior_qcolor"] + +#: Height of a behaviour lane when hardware lanes share the bar. Matches the +#: ethogram bar's old fixed height so a session with no hardware looks +#: exactly as it did. +_BEHAVIOR_LANE_HEIGHT = 46 + +#: Preferred height of one hardware lane. Rows compress below this rather +#: than overflowing when a rig has more devices than the bar has pixels. +_HW_LANE_HEIGHT = 14 + +_LANE_GAP = 2 + + +def behavior_qcolor(name: str, order: list[str] | None = None) -> QColor: + """The colour the annotated video would have drawn this behaviour in. + + Shared with the overlay so a bout looks the same wherever it is shown; + blank (unscored) frames read as background rather than a colour. + + ``order`` is the behaviours present, which is what makes the colours + reliably *different*. Without it the palette slot comes from a hash of the + name, and a hash has no reason to avoid collisions: two behaviours in one + session could land on the same colour, and neighbouring ones routinely + landed on adjacent hues. Given the session's own label set, the first N + palette entries are handed out in order, and N distinct behaviours get N + distinct colours. + """ + if not name: + return QColor(colors.BORDER) + from glider.analysis.behavior.classify.overlay import color_for_behavior + + b, g, r = color_for_behavior(name, order) + return QColor(r, g, b) + + +def behavior_order(labels) -> list[str]: + """The behaviours present, in a stable order. + + Sorted rather than first-appearance: the same cohort scored twice must + colour the same behaviour the same way, and first-appearance makes that + depend on which animal happened to groom first. + """ + return sorted({label for label in labels if label}) + + +class TimelineBar(QWidget): + """Every lane of a session, and the scrubber across all of them. + + Clicking or dragging with the left button scrubs; dragging with shift + (or the right button) selects a window. Selection and playhead stay + separate so a chosen window survives scrubbing around inside it. + """ + + scrubbed = pyqtSignal(int) # frame + selection_changed = pyqtSignal(int, int) # start, end frame + + def __init__(self, parent=None): + super().__init__(parent) + self.setMinimumHeight(_BEHAVIOR_LANE_HEIGHT) + self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed) + self.setCursor(Qt.CursorShape.PointingHandCursor) + self._view = None + self._timeline: Timeline | None = None + self._order: list[str] = [] + self._codes: dict[str, np.ndarray] = {} + self._static: QPixmap | None = None + self._frame = 0 + self._selection: tuple[int, int] | None = None + self._drag_anchor: int | None = None + + # ------------------------------------------------------------------ + # loading + + def set_view(self, view) -> None: + """Adopt a :class:`SessionView`. Kept for ethogram-only sessions.""" + self._view = view + self._rebuild_codes() + self._invalidate() + + def set_timeline(self, timeline: Timeline | None) -> None: + self._timeline = timeline + self._rebuild_codes() + self._invalidate() + self.updateGeometry() + + def _invalidate(self) -> None: + self._static = None + self._frame = 0 + self._selection = None + self.update() + + def _behavior_lanes(self) -> list[BehaviorLane]: + """The behaviour lanes to draw. + + A timeline's own lanes when it has any; otherwise one synthesised + from the view, so ``set_view`` alone still draws an ethogram. + """ + if self._timeline is not None and self._timeline.behavior: + return list(self._timeline.behavior) + if self._view is not None and len(self._view.labels): + return [ + BehaviorLane( + source="ethogram", + labels=list(self._view.labels), + frames=self._view.frames, + ) + ] + return [] + + def _rebuild_codes(self) -> None: + """Label-to-colour-slot per lane, against one shared order. + + Computed once per session rather than per band: which colour a + behaviour gets depends on which behaviours the session contains, and + the per-column majority in :meth:`_paint_behavior` needs the labels + as integers. The order is shared across lanes so the same behaviour + is the same colour whichever source scored it. + """ + lanes = self._behavior_lanes() + pooled: list[str] = [] + for lane in lanes: + pooled.extend(lane.labels) + self._order = behavior_order(pooled) if pooled else [] + slot = {name: i + 1 for i, name in enumerate(self._order)} # 0 = unscored + self._codes = { + lane.source: np.array([slot.get(v, 0) for v in lane.labels], dtype=np.int64) + for lane in lanes + } + + # ------------------------------------------------------------------ + # geometry + + def sizeHint(self) -> QSize: + n_behavior = max(1, len(self._behavior_lanes())) + n_hardware = len(self._timeline.lanes) if self._timeline else 0 + height = _BEHAVIOR_LANE_HEIGHT * n_behavior + _HW_LANE_HEIGHT * n_hardware + return QSize(super().sizeHint().width(), height) + + def _rows(self) -> list[tuple[str, object, float, float]]: + """``(kind, lane, top, height)`` for every row, top to bottom. + + With no hardware the behaviour lanes split the whole bar, which is + what keeps an ethogram-only session looking exactly as it did before + there were any other lanes. + """ + height = float(self.height()) + behavior = self._behavior_lanes() + hardware = list(self._timeline.lanes) if self._timeline else [] + + if not hardware: + if not behavior: + return [] + each = height / len(behavior) + return [("behavior", ln, i * each, each) for i, ln in enumerate(behavior)] + + rows: list[tuple[str, object, float, float]] = [] + top = 0.0 + for lane in behavior: + rows.append(("behavior", lane, top, float(_BEHAVIOR_LANE_HEIGHT))) + top += _BEHAVIOR_LANE_HEIGHT + _LANE_GAP + remaining = max(0.0, height - top) + each = remaining / len(hardware) + for lane in hardware: + rows.append(("hardware", lane, top, each)) + top += each + return rows + + def _frame_map(self): + return self._timeline.frame_map if self._timeline else None + + def _ms_bounds(self) -> tuple[float, float] | None: + if self._timeline is None: + return None + if self._timeline.end_ms <= self._timeline.start_ms: + return None + return self._timeline.start_ms, self._timeline.end_ms + + def frame_bounds(self) -> tuple[int, int]: + """``(first, last)`` frame the bar covers. + + From the frame map when there is one. Otherwise the ethogram's own + range, which is deliberately not zero: a windowed run scores minutes + two to seven, so its ethogram starts at frame 3600 and a bar drawn + from zero would spend its first eighth showing nothing. + """ + frame_map = self._frame_map() + bounds = self._ms_bounds() + if frame_map is not None and bounds is not None: + return frame_map.frame_at(bounds[0]), frame_map.frame_at(bounds[1]) + if self._view is not None and self._view.n_rows: + return int(self._view.frames[0]), int(self._view.frames[-1]) + return 0, 0 + + def _x_of_ms(self, ms: float) -> float: + bounds = self._ms_bounds() + if bounds is None or self.width() <= 0: + return 0.0 + start, end = bounds + return (float(ms) - start) / (end - start) * self.width() + + def _x_of(self, frame: int) -> float: + frame_map = self._frame_map() + if frame_map is not None and self._ms_bounds() is not None: + return self._x_of_ms(frame_map.ms_of(frame)) + first, last = self.frame_bounds() + span = max(0, last - first + 1) + return 0.0 if span == 0 else (frame - first) / span * self.width() + + def _frame_at(self, x: float) -> int: + first, last = self.frame_bounds() + if self.width() <= 0: + return first + bounds = self._ms_bounds() + frame_map = self._frame_map() + if frame_map is not None and bounds is not None: + start, end = bounds + ms = start + (x / self.width()) * (end - start) + return max(first, min(last, frame_map.frame_at(ms))) + span = max(0, last - first + 1) + if span == 0: + return first + return max(first, min(last, first + int(x / self.width() * span))) + + def resizeEvent(self, event): + # Bands are resolved per pixel column, so a different width is a + # different image. + self._static = None + super().resizeEvent(event) + + # ------------------------------------------------------------------ + # state + + def set_frame(self, frame: int) -> None: + self._frame = int(frame) + self.update() + + def selection(self) -> tuple[int, int] | None: + return self._selection + + def set_selection(self, start: int, end: int) -> None: + self._selection = (int(min(start, end)), int(max(start, end))) + self.update() + self.selection_changed.emit(*self._selection) + + # ------------------------------------------------------------------ + # painting + + def _static_pixmap(self) -> QPixmap: + """Every lane, drawn once per session and size. + + Cached because the playhead moves every frame during playback and + the behaviour bands cost a bincount per pixel column to resolve. + """ + if self._static is None: + self._static = QPixmap(self.size()) + self._static.fill(QColor(colors.BASE)) + painter = QPainter(self._static) + try: + for kind, lane, top, height in self._rows(): + if kind == "behavior": + self._paint_behavior(painter, lane, top, height) + else: + self._paint_hardware(painter, lane, top, height) + finally: + painter.end() + return self._static + + def _paint_hardware(self, painter, lane, top: float, height: float) -> None: + """One device's held values as bars growing from the row baseline. + + A digital pin reads as full-height blocks and a PWM ramp as a + staircase, from one renderer — the value is held until the device's + next event either way. + """ + if height <= 0: + return + baseline = top + height + painter.fillRect(QRectF(0, top, float(self.width()), height), QColor(colors.BASE)) + painter.setPen(Qt.PenStyle.NoPen) + for segment in lane.segments: + if segment.level <= 0: + continue + x0 = self._x_of_ms(segment.start_ms) + x1 = self._x_of_ms(segment.end_ms) + bar_height = height * segment.level + painter.fillRect( + QRectF(x0, baseline - bar_height, max(1.0, x1 - x0), bar_height), + QColor(colors.ACCENT), + ) + painter.setPen(QPen(QColor(colors.TEXT_MUTED), 1)) + for marker in lane.markers: + x = self._x_of_ms(marker.at_ms) + painter.drawLine(int(x), int(top), int(x), int(baseline)) + + def _paint_behavior(self, painter, lane, top: float, height: float) -> None: + """One band per *pixel column*, coloured by what dominates it. + + Not one rect per run, which is the obvious thing and was wrong. A + five-minute session holds around nine thousand scored rows and the + timeline is at most a couple of thousand pixels wide, so a typical run + is a fraction of a pixel: Qt drew each as a sub-pixel rectangle and + blended it with its neighbours by coverage. Every colour on the bar was + therefore an average of several behaviours — a bright yellow, a green + and a blue arriving on screen as one flat olive. No palette can survive + that, and it is why the bar looked washed out however distinct the + colours themselves were. + + Resolving to whole columns first makes every pixel one behaviour's + actual colour. It also means a run shorter than a column is not drawn, + which is honest — the bar shows proportions, and a pixel cannot show a + three-frame dart without overstating it. The bout stepper is how those + are reached. + """ + if height <= 0 or not lane.labels: + return + codes = self._codes.get(lane.source) + if codes is None: + return + width = self.width() + first, last = self.frame_bounds() + span = max(0, last - first + 1) + if width <= 0 or span == 0: + return + + frames = np.asarray(lane.frames) + edges = first + np.arange(width + 1, dtype=np.int64) * span // width + starts = np.searchsorted(frames, edges, side="left") + + n_codes = len(self._order) + 1 # + the unscored bucket + for x in range(width): + lo, hi = int(starts[x]), int(starts[x + 1]) + if hi <= lo: + # More pixels than rows: this column falls between two rows, + # so it takes the row to its left rather than a gap. + lo = max(0, min(lo, len(frames) - 1)) + hi = lo + 1 + counts = np.bincount(codes[lo:hi], minlength=n_codes) + code = int(counts.argmax()) + painter.fillRect( + QRectF(x, top, 1.0, height), + behavior_qcolor(self._order[code - 1] if code else "", self._order), + ) + + def paintEvent(self, _event): + painter = QPainter(self) + painter.fillRect(self.rect(), QColor(colors.BASE)) + if not self._rows(): + painter.setPen(QPen(QColor(colors.TEXT_MUTED))) + painter.drawText( + self.rect(), + Qt.AlignmentFlag.AlignCenter, + "Load a session to see its timeline", + ) + return + + painter.drawPixmap(0, 0, self._static_pixmap()) + + # Flow boundaries, across every lane. They are boundaries rather + # than a device, so they get a rule instead of a lane. + if self._timeline is not None and self._ms_bounds() is not None: + pen = QPen(QColor(colors.TEXT_MUTED), 1, Qt.PenStyle.DashLine) + painter.setPen(pen) + for boundary in (self._timeline.flow_start_ms, self._timeline.flow_end_ms): + if boundary is None: + continue + x = int(self._x_of_ms(boundary)) + painter.drawLine(x, 0, x, self.height()) + + if self._selection is not None: + start, end = self._selection + x0, x1 = self._x_of(start), self._x_of(end) + overlay = QColor(colors.ACCENT) + overlay.setAlpha(48) + painter.fillRect(QRectF(x0, 0, max(1.0, x1 - x0), self.height()), overlay) + + x = int(self._x_of(self._frame)) + painter.setPen(QPen(QColor(colors.TEXT_PRIMARY), 2)) + painter.drawLine(x, 0, x, self.height()) + + # ------------------------------------------------------------------ + # interaction + + def mousePressEvent(self, event): + if not self._rows(): + return + frame = self._frame_at(event.position().x()) + selecting = ( + event.button() == Qt.MouseButton.RightButton + or event.modifiers() & Qt.KeyboardModifier.ShiftModifier + ) + if selecting: + self._drag_anchor = frame + self.set_selection(frame, frame) + else: + self._drag_anchor = None + self.scrubbed.emit(frame) + + def mouseMoveEvent(self, event): + if not self._rows(): + return + frame = self._frame_at(event.position().x()) + if self._drag_anchor is not None: + self.set_selection(self._drag_anchor, frame) + elif event.buttons() & Qt.MouseButton.LeftButton: + self.scrubbed.emit(frame) + + def mouseReleaseEvent(self, _event): + self._drag_anchor = None diff --git a/tests/unit/gui/behavior/test_timeline_bar.py b/tests/unit/gui/behavior/test_timeline_bar.py new file mode 100644 index 00000000..8b4a0560 --- /dev/null +++ b/tests/unit/gui/behavior/test_timeline_bar.py @@ -0,0 +1,129 @@ +"""The timeline bar must stay a drop-in for the ethogram bar.""" + +from __future__ import annotations + +import numpy as np +import pytest + +pytest.importorskip("PyQt6") + +from glider.analysis.timeline import ( # noqa: E402 + BehaviorLane, + FrameMap, + Lane, + Marker, + Segment, + Timeline, +) +from glider.gui.widgets.timeline_bar import TimelineBar # noqa: E402 + + +def _timeline(*, hardware: bool = True, behavior: bool = True) -> Timeline: + """A 10s, 30fps session: LED on for the first 2s, one RFID read.""" + frames = np.arange(0, 301, dtype=float) + frame_map = FrameMap(frames=frames, ms=frames / 30.0 * 1000.0, source="tracking") + lanes = ( + [ + Lane( + key="led1", + label="led1", + board_id="board0", + segments=[ + Segment(start_ms=0.0, end_ms=2000.0, value=1.0, level=1.0), + Segment(start_ms=2000.0, end_ms=10000.0, value=0.0, level=0.0), + ], + markers=[Marker(at_ms=5000.0, label="tag-A7")], + ) + ] + if hardware + else [] + ) + beh = ( + [ + BehaviorLane( + source="tracking", + labels=["groom"] * 150 + ["locomote"] * 151, + frames=np.arange(0, 301), + ) + ] + if behavior + else [] + ) + return Timeline( + lanes=lanes, + behavior=beh, + frame_map=frame_map, + flow_start_ms=0.0, + flow_end_ms=10000.0, + start_ms=0.0, + end_ms=10000.0, + ) + + +@pytest.fixture +def bar(qtbot): + widget = TimelineBar() + qtbot.addWidget(widget) + widget.resize(300, 120) + return widget + + +def test_selection_is_emitted_in_frames(bar, qtbot): + """The stats tables consume frames. If this ever emits milliseconds, + every window statistic below the bar is silently computed over the + wrong range — and nothing would raise.""" + bar.set_timeline(_timeline()) + with qtbot.waitSignal(bar.selection_changed) as blocker: + bar.set_selection(10, 40) + assert blocker.args == [10, 40] + assert all(isinstance(v, int) for v in blocker.args) + + +def test_frame_bounds_come_from_the_frame_map(bar): + bar.set_timeline(_timeline()) + assert bar.frame_bounds() == (0, 300) + + +def test_empty_timeline_has_zero_bounds(bar): + bar.set_timeline(None) + assert bar.frame_bounds() == (0, 0) + + +def test_selection_survives_scrubbing(bar): + bar.set_timeline(_timeline()) + bar.set_selection(50, 200) + bar.set_frame(120) + assert bar.selection() == (50, 200) + + +def test_paints_without_a_timeline(bar, qtbot): + """A bar with nothing loaded draws its empty state rather than raising.""" + bar.set_timeline(None) + bar.show() + qtbot.waitExposed(bar) + bar.repaint() + + +def test_paints_a_loaded_timeline(bar, qtbot): + bar.set_timeline(_timeline()) + bar.show() + qtbot.waitExposed(bar) + bar.repaint() + + +def test_behavior_fills_the_bar_when_there_is_no_hardware(bar, qtbot): + """Protects test_analysis_window.py:488 — an ethogram-only session has + no hardware lanes, so its single behaviour lane must still occupy the + full height exactly as the old ethogram bar did.""" + bar.resize(300, 46) + bar.set_timeline(_timeline(hardware=False)) + image = bar.grab().toImage() + # Frame 75 of 300 sits near x=75 and is grooming top to bottom. + assert image.pixelColor(75, 8) == image.pixelColor(75, 40) + + +def test_hardware_lane_is_drawn_below_behavior(bar, qtbot): + """With hardware present the bar is no longer one flat band.""" + bar.set_timeline(_timeline()) + image = bar.grab().toImage() + assert image.pixelColor(20, 8) != image.pixelColor(20, 110) From 4f435a51ebf85ee13867a74dd11570fabfd48b71 Mon Sep 17 00:00:00 2001 From: Garrett Bradham Date: Thu, 10 Sep 2026 14:11:41 -0500 Subject: [PATCH 10/12] fix(gui): draw every lane on one time axis _paint_behavior kept the ethogram bar's frame-linear column edges, which only agree with _x_of_ms when the ms bounds exactly match the frame map's range. The axis deliberately spans pre-flow device writes, so they usually do not: a session with 30s of device init ahead of the camera drew the ethogram across the whole bar while the camera occupied its last quarter, and the playhead sat on a band showing the wrong behaviour. Columns are now sliced in milliseconds and converted through the frame map; the frame-linear path stays for the ethogram-only case, where the axis really is frames. frame_bounds now falls back to the behaviour lanes' own frame range rather than to self._view. build_timeline(None, view) yields no frame map and no ms bounds, so set_timeline without a companion set_view reported (0, 0), collapsing every column onto frame 0 and flattening the ethogram to one colour. The selection overlay goes back to shading what is EXCLUDED, with the comment recording why: tinting the selection dragged every behaviour inside it toward the accent, and the usual selection is the whole session. Its right edge is _x_of(end + 1), which the accent version cut a column short. Both pixel tests now assert actual colours; each was verified to fail with its painter stubbed to a no-op. --- src/glider/gui/widgets/timeline_bar.py | 59 ++++++++++--- tests/unit/gui/behavior/test_timeline_bar.py | 91 ++++++++++++++++++-- 2 files changed, 132 insertions(+), 18 deletions(-) diff --git a/src/glider/gui/widgets/timeline_bar.py b/src/glider/gui/widgets/timeline_bar.py index 40cf85ef..c3b659d4 100644 --- a/src/glider/gui/widgets/timeline_bar.py +++ b/src/glider/gui/widgets/timeline_bar.py @@ -18,8 +18,8 @@ from __future__ import annotations import numpy as np -from PyQt6.QtCore import QRectF, QSize, Qt, pyqtSignal -from PyQt6.QtGui import QColor, QPainter, QPen, QPixmap +from PyQt6.QtCore import QPointF, QRectF, QSize, Qt, pyqtSignal +from PyQt6.QtGui import QBrush, QColor, QPainter, QPen, QPixmap from PyQt6.QtWidgets import QSizePolicy, QWidget from glider.analysis.timeline import BehaviorLane, Timeline @@ -206,17 +206,27 @@ def _ms_bounds(self) -> tuple[float, float] | None: def frame_bounds(self) -> tuple[int, int]: """``(first, last)`` frame the bar covers. - From the frame map when there is one. Otherwise the ethogram's own - range, which is deliberately not zero: a windowed run scores minutes - two to seven, so its ethogram starts at frame 3600 and a bar drawn - from zero would spend its first eighth showing nothing. + From the frame map when there is one. Otherwise the behaviour lanes' + own range, which is deliberately not zero: a windowed run scores + minutes two to seven, so its ethogram starts at frame 3600 and a bar + drawn from zero would spend its first eighth showing nothing. + + The lanes rather than ``self._view``, because an ethogram-only + ``Timeline`` has no frame map and no ms bounds either — falling back + to the view alone left ``set_timeline`` without ``set_view`` + reporting ``(0, 0)``, which collapses every column onto frame 0 and + paints the whole ethogram one flat colour. ``_behavior_lanes`` + already synthesises a lane from the view, so this covers both. """ frame_map = self._frame_map() bounds = self._ms_bounds() if frame_map is not None and bounds is not None: return frame_map.frame_at(bounds[0]), frame_map.frame_at(bounds[1]) - if self._view is not None and self._view.n_rows: - return int(self._view.frames[0]), int(self._view.frames[-1]) + lanes = [ln for ln in self._behavior_lanes() if len(ln.frames)] + if lanes: + firsts = [int(np.asarray(ln.frames)[0]) for ln in lanes] + lasts = [int(np.asarray(ln.frames)[-1]) for ln in lanes] + return min(firsts), max(lasts) return 0, 0 def _x_of_ms(self, ms: float) -> float: @@ -351,7 +361,21 @@ def _paint_behavior(self, painter, lane, top: float, height: float) -> None: return frames = np.asarray(lane.frames) - edges = first + np.arange(width + 1, dtype=np.int64) * span // width + bounds = self._ms_bounds() + frame_map = self._frame_map() + if frame_map is not None and bounds is not None and len(frame_map.frames): + # Columns are equal slices of the TIME axis, because that is the + # axis every other lane and the playhead are drawn on. Slicing the + # frame axis instead smears the ethogram across any stretch the + # camera did not cover -- and the axis deliberately spans + # pre-flow device writes, so that stretch is routinely real. + start_ms, end_ms = bounds + edges_ms = start_ms + np.arange(width + 1, dtype=float) * (end_ms - start_ms) / width + idx = np.clip(np.searchsorted(frame_map.ms, edges_ms), 0, len(frame_map.frames) - 1) + edges = frame_map.frames[idx].astype(np.int64) + else: + # No frame map: the axis IS frames, so equal slices of it are right. + edges = first + np.arange(width + 1, dtype=np.int64) * span // width starts = np.searchsorted(frames, edges, side="left") n_codes = len(self._order) + 1 # + the unscored bucket @@ -396,10 +420,19 @@ def paintEvent(self, _event): if self._selection is not None: start, end = self._selection - x0, x1 = self._x_of(start), self._x_of(end) - overlay = QColor(colors.ACCENT) - overlay.setAlpha(48) - painter.fillRect(QRectF(x0, 0, max(1.0, x1 - x0), self.height()), overlay) + x0, x1 = self._x_of(start), self._x_of(end + 1) + # Shade what is EXCLUDED, not what is chosen. Tinting the selection + # blue meant every behaviour inside it was drawn 28% toward the + # accent — and since the usual selection is the whole session, that + # was every colour on the bar, all of them dragged toward the same + # hue. Shading the outside leaves the data at full strength and + # says the same thing. + scrim = QBrush(colors.qcolor_with_alpha(QColor(colors.BASE), 0.72)) + painter.fillRect(QRectF(0, 0, max(0.0, x0), self.height()), scrim) + painter.fillRect(QRectF(x1, 0, max(0.0, self.width() - x1), self.height()), scrim) + painter.setPen(QPen(QColor(colors.ACCENT), 2)) + painter.drawLine(QPointF(x0, 0), QPointF(x0, self.height())) + painter.drawLine(QPointF(x1, 0), QPointF(x1, self.height())) x = int(self._x_of(self._frame)) painter.setPen(QPen(QColor(colors.TEXT_PRIMARY), 2)) diff --git a/tests/unit/gui/behavior/test_timeline_bar.py b/tests/unit/gui/behavior/test_timeline_bar.py index 8b4a0560..d0b29928 100644 --- a/tests/unit/gui/behavior/test_timeline_bar.py +++ b/tests/unit/gui/behavior/test_timeline_bar.py @@ -2,11 +2,15 @@ from __future__ import annotations +from types import SimpleNamespace + import numpy as np import pytest pytest.importorskip("PyQt6") +from PyQt6.QtGui import QColor # noqa: E402 + from glider.analysis.timeline import ( # noqa: E402 BehaviorLane, FrameMap, @@ -14,8 +18,12 @@ Marker, Segment, Timeline, + build_timeline, ) -from glider.gui.widgets.timeline_bar import TimelineBar # noqa: E402 +from glider.gui.styles import colors # noqa: E402 +from glider.gui.widgets.timeline_bar import TimelineBar, behavior_qcolor # noqa: E402 + +_ORDER = ["groom", "locomote"] def _timeline(*, hardware: bool = True, behavior: bool = True) -> Timeline: @@ -118,12 +126,85 @@ def test_behavior_fills_the_bar_when_there_is_no_hardware(bar, qtbot): bar.resize(300, 46) bar.set_timeline(_timeline(hardware=False)) image = bar.grab().toImage() - # Frame 75 of 300 sits near x=75 and is grooming top to bottom. - assert image.pixelColor(75, 8) == image.pixelColor(75, 40) + # Frame 75 of 300 sits near x=75 and is grooming top to bottom. Assert + # the colour, not just that the two pixels agree: two untouched + # background pixels agree too, so the weaker form passed against a + # painter that drew nothing at all. + groom = behavior_qcolor("groom", _ORDER) + assert image.pixelColor(75, 8) == groom + assert image.pixelColor(75, 40) == groom def test_hardware_lane_is_drawn_below_behavior(bar, qtbot): - """With hardware present the bar is no longer one flat band.""" + """With hardware present the bar is no longer one flat band. + + Against the accent, not merely against the behaviour row: an unpainted + hardware row is background, which differs from the behaviour band too, + so the inequality alone held even with ``_paint_hardware`` stubbed out. + """ bar.set_timeline(_timeline()) image = bar.grab().toImage() - assert image.pixelColor(20, 8) != image.pixelColor(20, 110) + # x=20 is ~666ms, inside the LED's first-2s segment at level 1.0, and + # y=110 is in the hardware row (behaviour takes the top 46px). + assert image.pixelColor(20, 8) == behavior_qcolor("groom", _ORDER) + assert image.pixelColor(20, 110) == QColor(colors.ACCENT) + + +def _pre_flow_timeline() -> Timeline: + """The same 10s of camera, with 30s of device writes ahead of it. + + Which is the ordinary shape of a hardware session: ``build_timeline`` + deliberately extends the axis back over pre-flow device initialisation, + so the camera occupies only the last quarter of the bar. + """ + timeline = _timeline() + return Timeline( + lanes=timeline.lanes, + behavior=timeline.behavior, + frame_map=timeline.frame_map, + flow_start_ms=0.0, + flow_end_ms=10000.0, + start_ms=-30000.0, + end_ms=10000.0, + ) + + +def test_behavior_bands_sit_on_the_same_axis_as_everything_else(bar, qtbot): + """Behaviour resolves on the TIME axis, like the playhead and the lanes. + + Slicing the frame axis into columns instead spread the ethogram across + the full width even though the camera only covered the last quarter of + it, so the playhead sat on one behaviour while the band under it showed + another. + """ + bar.resize(400, 46) + bar.set_timeline(_pre_flow_timeline()) + + # Frame 150 (5000ms) is where groom becomes locomote. + boundary_x = bar._x_of_ms(bar._frame_map().ms_of(150)) + assert boundary_x == pytest.approx(350.0) + + image = bar.grab().toImage() + assert image.pixelColor(int(boundary_x) - 1, 20) == behavior_qcolor("groom", _ORDER) + assert image.pixelColor(int(boundary_x), 20) == behavior_qcolor("locomote", _ORDER) + + +def test_ethogram_only_timeline_paints_without_a_view(bar, qtbot): + """``set_timeline`` alone is enough — no companion ``set_view`` call. + + ``build_timeline(None, view)`` has no frame map and no ms bounds, so the + bar has to take its frame range from the behaviour lane itself. Taking + it from ``self._view`` instead reported ``(0, 0)`` here, which collapsed + every column onto frame 0 and painted the whole ethogram one colour. + """ + view = SimpleNamespace( + labels=["groom"] * 150 + ["locomote"] * 151, + frames=np.arange(0, 301), + ) + bar.resize(300, 46) + bar.set_timeline(build_timeline(None, view)) + + assert bar.frame_bounds() == (0, 300) + image = bar.grab().toImage() + assert image.pixelColor(75, 20) == behavior_qcolor("groom", _ORDER) + assert image.pixelColor(250, 20) == behavior_qcolor("locomote", _ORDER) From ff3b467ddf611036a48e0b753bf17433fbe6b685 Mon Sep 17 00:00:00 2001 From: Garrett Bradham Date: Thu, 10 Sep 2026 14:24:42 -0500 Subject: [PATCH 11/12] feat(gui): put the hardware raster in Session Review --- src/glider/gui/behavior/analysis_window.py | 236 ++---------------- .../unit/gui/behavior/test_analysis_window.py | 27 +- 2 files changed, 33 insertions(+), 230 deletions(-) diff --git a/src/glider/gui/behavior/analysis_window.py b/src/glider/gui/behavior/analysis_window.py index 035060bc..5f099d2e 100644 --- a/src/glider/gui/behavior/analysis_window.py +++ b/src/glider/gui/behavior/analysis_window.py @@ -21,7 +21,7 @@ from pathlib import Path import numpy as np -from PyQt6.QtCore import QPointF, QRectF, Qt, QTimer, pyqtSignal +from PyQt6.QtCore import QPointF, QRectF, Qt, QTimer from PyQt6.QtGui import QBrush, QColor, QIcon, QImage, QPainter, QPen, QPixmap from PyQt6.QtWidgets import ( QCheckBox, @@ -45,8 +45,9 @@ ) from glider.analysis.behavior.session_view import SessionView, SessionViewError +from glider.analysis.timeline import build_timeline from glider.gui.styles import colors -from glider.gui.widgets.timeline_bar import behavior_order, behavior_qcolor +from glider.gui.widgets.timeline_bar import TimelineBar, behavior_order, behavior_qcolor from glider.gui.widgets.tool_ui import ( CARD_GAP, GUTTER, @@ -155,224 +156,6 @@ def _vrule() -> QFrame: return line -class EthogramBar(QWidget): - """The ethogram as a timeline: bands to read, and the scrubber to drag. - - Clicking or dragging with the left button scrubs; dragging with shift (or - the right button) selects a window. Selection and playhead are separate so - a chosen window survives scrubbing around inside it. - """ - - scrubbed = pyqtSignal(int) # frame - selection_changed = pyqtSignal(int, int) # start, end frame - - def __init__(self, parent=None): - super().__init__(parent) - self.setMinimumHeight(_BAR_HEIGHT) - self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed) - self.setCursor(Qt.CursorShape.PointingHandCursor) - self._view: SessionView | None = None - self._order: list[str] = [] - self._codes: np.ndarray | None = None - self._lane: QPixmap | None = None - self._frame = 0 - self._selection: tuple[int, int] | None = None - self._drag_anchor: int | None = None - - def set_view(self, view: SessionView | None) -> None: - self._view = view - # Computed once per session rather than per band: the colour a - # behaviour gets depends on which behaviours this session contains, - # and the per-column majority below needs the labels as integers. - self._order = behavior_order(view.labels) if view is not None else [] - if view is None: - self._codes = None - else: - slot = {name: i + 1 for i, name in enumerate(self._order)} # 0 = unscored - self._codes = np.array([slot.get(label, 0) for label in view.labels], dtype=np.int64) - self._lane = None - self._frame = 0 - self._selection = None - self.update() - - def set_frame(self, frame: int) -> None: - self._frame = int(frame) - self.update() - - def resizeEvent(self, event): - # The bands are resolved per pixel column, so a different width is a - # different image. - self._lane = None - super().resizeEvent(event) - - def _lane_pixmap(self) -> QPixmap: - """The behaviour bands, drawn once per session and size.""" - if self._lane is None: - self._lane = QPixmap(self.size()) - self._lane.fill(QColor(colors.BASE)) - lane_painter = QPainter(self._lane) - try: - self._paint_lane(lane_painter, self._view.labels, 0.0, float(self.height())) - finally: - lane_painter.end() - return self._lane - - def selection(self) -> tuple[int, int] | None: - return self._selection - - def set_selection(self, start: int, end: int) -> None: - self._selection = (int(min(start, end)), int(max(start, end))) - self.update() - self.selection_changed.emit(*self._selection) - - # ------------------------------------------------------------------ - - def frame_bounds(self) -> tuple[int, int]: - """``(first, last)`` frame the ethogram actually covers. - - A windowed run scores minutes two to seven, so its ethogram starts at - frame 3600 — and a timeline drawn from zero would spend its first - eighth showing nothing, with a playhead that scrubs through frames no - one scored. The timeline is the ethogram, so it starts where the - ethogram starts. - """ - if self._view is None or self._view.n_rows == 0: - return 0, 0 - return int(self._view.frames[0]), int(self._view.frames[-1]) - - def _span(self) -> int: - first, last = self.frame_bounds() - return max(0, last - first + 1) - - def _frame_at(self, x: float) -> int: - first, last = self.frame_bounds() - span = self._span() - if span == 0 or self.width() <= 0: - return first - return max(first, min(last, first + int(x / self.width() * span))) - - def _x_of(self, frame: int) -> float: - first, _last = self.frame_bounds() - span = self._span() - return 0.0 if span == 0 else (frame - first) / span * self.width() - - def paintEvent(self, _event): - painter = QPainter(self) - painter.fillRect(self.rect(), QColor(colors.BASE)) - if self._view is None or self._span() == 0: - painter.setPen(QPen(QColor(colors.TEXT_MUTED))) - painter.drawText( - self.rect(), Qt.AlignmentFlag.AlignCenter, "Load a session to see its ethogram" - ) - return - - # One lane, because there is one behaviour per frame. Freezing and - # darting are values of it, not a parallel track: a second lane would - # be drawing the same frames twice. - # - # Cached: the bands only change when the session or the width does, - # while the playhead moves thirty times a second during playback, and - # resolving nine thousand rows into columns on every one of those - # frames is a fifth of the frame budget spent redrawing the same image. - painter.drawPixmap(0, 0, self._lane_pixmap()) - - if self._selection is not None: - start, end = self._selection - x0, x1 = self._x_of(start), self._x_of(end + 1) - # Shade what is EXCLUDED, not what is chosen. Tinting the selection - # blue meant every behaviour inside it was drawn 28% toward the - # accent — and since the usual selection is the whole session, that - # was every colour on the bar, all of them dragged toward the same - # hue. Shading the outside leaves the data at full strength and - # says the same thing. - scrim = QBrush(colors.qcolor_with_alpha(QColor(colors.BASE), 0.72)) - painter.fillRect(QRectF(0, 0, max(0.0, x0), self.height()), scrim) - painter.fillRect(QRectF(x1, 0, max(0.0, self.width() - x1), self.height()), scrim) - painter.setPen(QPen(QColor(colors.ACCENT), 2)) - painter.drawLine(QPointF(x0, 0), QPointF(x0, self.height())) - painter.drawLine(QPointF(x1, 0), QPointF(x1, self.height())) - - painter.setPen(QPen(QColor(colors.TEXT_PRIMARY), 2)) - x = self._x_of(self._frame) - painter.drawLine(QPointF(x, 0), QPointF(x, self.height())) - - def _paint_lane(self, painter, labels, top: float, height: float) -> None: - """One band per *pixel column*, coloured by what dominates it. - - Not one rect per run, which is the obvious thing and was wrong. A - five-minute session holds around nine thousand scored rows and the - timeline is at most a couple of thousand pixels wide, so a typical run - is a fraction of a pixel: Qt drew each as a sub-pixel rectangle and - blended it with its neighbours by coverage. Every colour on the bar was - therefore an average of several behaviours — a bright yellow, a green - and a blue arriving on screen as one flat olive. No palette can survive - that, and it is why the bar looked washed out however distinct the - colours themselves were. - - Resolving to whole columns first makes every pixel one behaviour's - actual colour. It also means a run shorter than a column is not drawn, - which is honest — the bar shows proportions, and a pixel cannot show a - three-frame dart without overstating it. The bout stepper is how those - are reached. - """ - if height <= 0 or not labels or self._codes is None: - return - width = self.width() - span = self._span() - if width <= 0 or span == 0: - return - - first, _last = self.frame_bounds() - frames = self._view.frames - # Which row each column starts at: the columns are equal slices of the - # frame axis, and the rows are already sorted by frame. - edges = first + np.arange(width + 1, dtype=np.int64) * span // width - starts = np.searchsorted(frames, edges, side="left") - - n_codes = len(self._order) + 1 # + the unscored bucket - for x in range(width): - lo, hi = int(starts[x]), int(starts[x + 1]) - if hi <= lo: - # More pixels than rows: this column falls between two rows, so - # it takes the row to its left rather than a gap in the bar. - lo, hi = max(0, min(lo, len(frames) - 1)), max(0, min(lo, len(frames) - 1)) + 1 - counts = np.bincount(self._codes[lo:hi], minlength=n_codes) - code = int(counts.argmax()) - painter.fillRect( - QRectF(x, top, 1.0, height), - behavior_qcolor(self._order[code - 1] if code else "", self._order), - ) - - # ------------------------------------------------------------------ - - def mousePressEvent(self, event): - if self._view is None: - return - frame = self._frame_at(event.position().x()) - selecting = ( - event.button() == Qt.MouseButton.RightButton - or event.modifiers() & Qt.KeyboardModifier.ShiftModifier - ) - if selecting: - self._drag_anchor = frame - self.set_selection(frame, frame) - else: - self._drag_anchor = None - self.scrubbed.emit(frame) - - def mouseMoveEvent(self, event): - if self._view is None: - return - frame = self._frame_at(event.position().x()) - if self._drag_anchor is not None: - self.set_selection(self._drag_anchor, frame) - elif event.buttons() & Qt.MouseButton.LeftButton: - self.scrubbed.emit(frame) - - def mouseReleaseEvent(self, _event): - self._drag_anchor = None - - class KeypointCanvas(QWidget): """The animal drawn from its poses, with a trailing centroid track. @@ -766,7 +549,7 @@ def __init__(self, parent=None): self._canvas = KeypointCanvas() viewer_body.addWidget(self._canvas, 1) - self._bar = EthogramBar() + self._bar = TimelineBar() self._bar.scrubbed.connect(self._set_frame) self._bar.selection_changed.connect(self._on_selection) viewer_body.addWidget(self._bar) @@ -1084,6 +867,16 @@ def _adopt(self, ethogram_csv: Path, view: SessionView) -> None: self._path_label.setText(_short_path(Path(ethogram_csv))) self._path_label.setToolTip(str(ethogram_csv)) self._bar.set_view(view) + session = None + try: + from glider.analysis import Session + + session = Session.load(ethogram_csv.parent) + except (OSError, ValueError, NotADirectoryError): + # An ethogram with no recording beside it is the normal case for + # an apply-run output folder — behaviour lanes, no raster. + logger.debug("no recording beside %s", ethogram_csv, exc_info=True) + self._bar.set_timeline(build_timeline(session, view)) self._canvas.set_view(view) # The overlay belongs to the session that just left. Nothing else # clears it: set_view leaves _heatmap alone and bar.set_view drops the @@ -1714,7 +1507,6 @@ def closeEvent(self, event): __all__ = [ "AnalysisWindow", - "EthogramBar", "KeypointCanvas", "behavior_order", "behavior_qcolor", diff --git a/tests/unit/gui/behavior/test_analysis_window.py b/tests/unit/gui/behavior/test_analysis_window.py index e2469885..2d3d54ac 100644 --- a/tests/unit/gui/behavior/test_analysis_window.py +++ b/tests/unit/gui/behavior/test_analysis_window.py @@ -14,10 +14,10 @@ from glider.analysis.behavior.session_view import SessionView # noqa: E402 from glider.gui.behavior.analysis_window import ( # noqa: E402 AnalysisWindow, - EthogramBar, KeypointCanvas, behavior_qcolor, ) +from glider.gui.widgets.timeline_bar import TimelineBar # noqa: E402 NAMES = ["nose", "l_ear", "r_ear", "tail_base"] @@ -61,16 +61,16 @@ def test_unscored_frames_read_as_background(self): assert behavior_qcolor("") == QColor(colors.BORDER) -class TestEthogramBar: +class TestTimelineBar: def _bar(self, qtbot, tmp_path, **kw): - bar = EthogramBar() + bar = TimelineBar() qtbot.addWidget(bar) bar.resize(300, 46) bar.set_view(SessionView.load(_session(tmp_path / "v", **kw))) return bar def test_an_empty_bar_does_not_crash(self, qtbot): - bar = EthogramBar() + bar = TimelineBar() qtbot.addWidget(bar) bar.set_view(None) bar.resize(200, 46) @@ -209,6 +209,17 @@ def test_selecting_a_window_fills_the_bout_table(self, qtbot, tmp_path): assert win._bouts.rowCount() == 1 assert win._bouts.item(0, 0).text() == "locomote" + def test_selection_still_arrives_in_frames(self, qtbot, tmp_path): + """The swap from EthogramBar to TimelineBar must not change the + unit the tables receive. If it does, every window statistic is + computed over the wrong range and nothing raises.""" + win = self._win(qtbot, tmp_path) + received: list[tuple[int, int]] = [] + win._bar.selection_changed.connect(lambda a, b: received.append((a, b))) + win._bar.set_selection(100, 199) + assert received == [(100, 199)] + assert win._bouts.item(0, 0).text() == "locomote" + def test_a_span_across_behaviours_lists_both(self, qtbot, tmp_path): win = self._win(qtbot, tmp_path) win._bar.set_selection(50, 250) @@ -486,7 +497,7 @@ def _session_with_freezing(self, tmp_path, n=300): return folder / "ethogram_raw.csv" def test_the_bar_paints_one_full_height_lane(self, qtbot, tmp_path): - bar = EthogramBar() + bar = TimelineBar() qtbot.addWidget(bar) bar.resize(300, 46) bar.set_view(SessionView.load(self._session_with_freezing(tmp_path))) @@ -495,7 +506,7 @@ def test_the_bar_paints_one_full_height_lane(self, qtbot, tmp_path): assert image.pixelColor(130, 8) == image.pixelColor(130, 40) def test_freezing_is_drawn_in_its_own_colour(self, qtbot, tmp_path): - bar = EthogramBar() + bar = TimelineBar() qtbot.addWidget(bar) bar.resize(300, 46) bar.set_view(SessionView.load(self._session_with_freezing(tmp_path))) @@ -921,7 +932,7 @@ def test_the_playhead_opens_on_the_first_scored_frame(self, qtbot, tmp_path): assert win._frame == 3600 def test_clicking_the_far_left_lands_on_the_first_frame(self, qtbot, tmp_path): - bar = EthogramBar() + bar = TimelineBar() qtbot.addWidget(bar) bar.resize(300, 46) bar.set_view(SessionView.load(_windowed_session(tmp_path))) @@ -931,7 +942,7 @@ def test_clicking_the_far_left_lands_on_the_first_frame(self, qtbot, tmp_path): assert 12570 <= bar._frame_at(299.9) <= 12599 def test_the_scored_range_fills_the_width(self, qtbot, tmp_path): - bar = EthogramBar() + bar = TimelineBar() qtbot.addWidget(bar) bar.resize(300, 46) bar.set_view(SessionView.load(_windowed_session(tmp_path))) From 4b198d1995dfd9e535bf619b2f3defb08efdf944 Mon Sep 17 00:00:00 2001 From: Garrett Bradham Date: Thu, 10 Sep 2026 14:54:38 -0500 Subject: [PATCH 12/12] fix(gui): find the recording the ethogram came from MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Session Review loaded `ethogram_csv.parent` as the recording directory, which is off by exactly one level: an apply run writes the ethogram into `/