From 40637cc1a1b21eceffc83b06f24b9567e0945f4d Mon Sep 17 00:00:00 2001 From: gbradham Date: Thu, 29 Oct 2026 16:06:57 -0500 Subject: [PATCH] feat(review): show the drawn arena, an N x N grid, and how much was explored Session Review now finds each session's arena (a saved _arena.json, else the batch's pose_calibration.json) and outlines it on the video. An Arena menu draws or loads one and saves it beside the session. The Range tab gets an Exploration section: pick a grid (Off, 2x2 ... 20x20), and it reports cells entered, % explored and the time to enter every cell. The grid is laid out in floor centimetres and mapped through the arena's homography, so cells cover equal floor under a tilted camera; round and plus-maze arenas count only cells on the floor. Visited cells are shaded, and the numbers reach the epoch table and the CSV exports. --- src/glider/analysis/behavior/exploration.py | 298 ++++++++++++++++++ src/glider/analysis/epochs.py | 18 +- src/glider/gui/behavior/analysis_window.py | 187 ++++++++++- src/glider/gui/review/inspector.py | 45 +++ src/glider/gui/review/viewer.py | 70 ++++ .../analysis/behavior/test_exploration.py | 174 ++++++++++ .../unit/gui/behavior/test_analysis_window.py | 96 ++++++ 7 files changed, 882 insertions(+), 6 deletions(-) create mode 100644 src/glider/analysis/behavior/exploration.py create mode 100644 tests/unit/analysis/behavior/test_exploration.py diff --git a/src/glider/analysis/behavior/exploration.py b/src/glider/analysis/behavior/exploration.py new file mode 100644 index 00000000..b1c36ed9 --- /dev/null +++ b/src/glider/analysis/behavior/exploration.py @@ -0,0 +1,298 @@ +"""How much of the arena the animal covered, on a grid laid over the drawn floor. + +Exploration is the share of an N x N grid of floor cells the animal entered. +The grid is laid out in centimetres on the floor and mapped into the image +through the arena's homography, so under a tilted camera the far cells are +drawn smaller -- they are the same patch of floor as the near ones, which is +what makes "7 of 9 cells" mean the same thing in every video. A grid drawn +straight over the pixels would give the near wall more floor per cell than the +far wall, and a coverage figure that depends on where the camera sat. + +A round arena keeps only the cells whose centre is on the floor, so a corner +of the bounding box that is empty air never counts against the animal; a plus +maze likewise keeps the cells on its arms. + +Qt-free, so the epoch table, an export and a notebook all get the same number. +""" + +from __future__ import annotations + +import json +import logging +from dataclasses import dataclass +from pathlib import Path + +import numpy as np + +from glider.analysis.behavior.spatial import CENTROID, tracking_frame + +logger = logging.getLogger(__name__) + +__all__ = [ + "ARENA_SUFFIX", + "Exploration", + "arena_file_for", + "cell_polygon", + "explore", + "find_session_arena", + "floor_cells", + "grid_segments", + "read_arena", +] + +#: Saved-arena filename suffix, the spelling the project layout already reads +#: (``ProjectSession.arena``). +ARENA_SUFFIX = "_arena.json" + +_MASTER = "pose_calibration.json" + + +# ---------------------------------------------------------------------- +# finding the arena +# ---------------------------------------------------------------------- + + +def read_arena(path: Path | str): + """The :class:`ArenaCalibration` in an arena file, or None. + + Accepts both the bare calibration and the document ``tools/arena_zones.py`` + writes, which wraps it under ``arena`` beside its derived numbers. + """ + from glider.vision.arena import ArenaCalibration + + try: + data = json.loads(Path(path).read_text(encoding="utf-8")) + except (OSError, ValueError): + return None + if not isinstance(data, dict): + return None + for candidate in (data.get("arena"), data): + if not isinstance(candidate, dict) or "corners" not in candidate: + continue + try: + return ArenaCalibration.from_dict(candidate) + except (KeyError, TypeError, ValueError): + continue + return None + + +def _session_folder(session_path: Path) -> Path: + return session_path if session_path.is_dir() else session_path.parent + + +def arena_file_for(session_path: Path | str) -> Path: + """Where an arena drawn in Session Review is saved for this session.""" + folder = _session_folder(Path(session_path)) + return folder / f"{folder.name}{ARENA_SUFFIX}" + + +def find_session_arena(session_path: Path | str, view=None): + """``(arena, source)`` for a session, or None when none was drawn. + + Looked for first as a saved arena beside the session (named after its + folder or its video), then in the master ``pose_calibration.json`` an apply + run read, matched by the session's video. + """ + from glider.analysis.behavior.session_view import _find_upward + + folder = _session_folder(Path(session_path)) + video = getattr(view, "video_path", None) + stems = [folder.name] + ([Path(video).stem] if video is not None else []) + beside = [folder / f"{stem}{ARENA_SUFFIX}" for stem in dict.fromkeys(stems)] + if video is not None: + beside.append(Path(video).with_name(f"{Path(video).stem}{ARENA_SUFFIX}")) + for candidate in beside: + if candidate.is_file(): + arena = read_arena(candidate) + if arena is not None: + return arena, candidate + logger.info("ignoring unreadable arena %s", candidate) + + master = _find_upward(folder, _MASTER) + if master is None: + return None + from glider.vision.calibration_set import CalibrationSet, CalibrationSetError + + try: + loaded = CalibrationSet.load(master, known_videos=[Path(video)] if video else None) + except (CalibrationSetError, OSError) as e: + logger.info("no usable calibration in %s: %s", master, e) + return None + if video is not None: + arena = loaded.get_arena(video) + if arena is not None: + return arena, master + # The video was found somewhere other than where the batch saw it; the + # session folder is named after it, so match by stem -- but only a unique + # match, never a guess between two. + matches = [a for v, a in loaded.arenas.items() if v.stem in stems] + return (matches[0], master) if len(matches) == 1 else None + + +# ---------------------------------------------------------------------- +# grid geometry +# ---------------------------------------------------------------------- + + +def _to_cm(arena, normalised: np.ndarray) -> tuple[np.ndarray, np.ndarray]: + """Floor centimetres for ``(n, 2)`` normalised image points; NaN where unmappable.""" + pts = np.asarray(normalised, dtype=np.float64).reshape(-1, 2) + h = arena.homography() + w = pts[:, 0] * h[2, 0] + pts[:, 1] * h[2, 1] + h[2, 2] + with np.errstate(divide="ignore", invalid="ignore"): + x = (pts[:, 0] * h[0, 0] + pts[:, 1] * h[0, 1] + h[0, 2]) / w + y = (pts[:, 0] * h[1, 0] + pts[:, 1] * h[1, 1] + h[1, 2]) / w + # Past the horizon a point maps to the floor's far side, flipped; it is + # not on the floor at all. + bad = ~(w > 0) + x[bad] = np.nan + y[bad] = np.nan + return x, y + + +def _cell_size(arena, n: int) -> tuple[float, float]: + return arena.width_cm / n, arena.height_cm / n + + +def floor_cells(arena, n: int) -> np.ndarray: + """``(n, n)`` bool, row = y, column = x: the cells that count as floor.""" + cw, ch = _cell_size(arena, n) + xs = (np.arange(n) + 0.5) * cw + ys = (np.arange(n) + 0.5) * ch + gx, gy = np.meshgrid(xs, ys) + return ~np.asarray(arena.outside_cm(gx, gy), dtype=bool) + + +def grid_segments(arena, n: int, samples: int = 48) -> list[list[tuple[float, float]]]: + """The grid's interior lines as normalised image polylines, clipped to the floor. + + Each line is sampled along its length and broken where it leaves the + floor, so the same code clips a round arena's chords and a plus maze's + arms; a homography keeps lines straight, so the samples cost nothing in + accuracy. + """ + if n < 2: + return [] + w, h = arena.width_cm, arena.height_cm + t = np.linspace(0.0, 1.0, samples) + lines_cm = [(np.full_like(t, w * k / n), t * h) for k in range(1, n)] + lines_cm += [(t * w, np.full_like(t, h * k / n)) for k in range(1, n)] + out: list[list[tuple[float, float]]] = [] + for xs, ys in lines_cm: + # A whisker of margin, so a line along an arm's edge is not dropped + # for sitting exactly on the boundary. + on = ~np.asarray(arena.outside_cm(xs, ys, 1e-6), dtype=bool) + run: list[tuple[float, float]] = [] + for i in range(len(t)): + if on[i]: + run.append((float(xs[i]), float(ys[i]))) + elif run: + out.append(run) + run = [] + if run: + out.append(run) + return [arena.to_image(run) for run in out if len(run) > 1] + + +def cell_polygon(arena, n: int, col: int, row: int) -> list[tuple[float, float]]: + """One cell's four corners, as normalised image points.""" + cw, ch = _cell_size(arena, n) + x0, y0 = col * cw, row * ch + return arena.to_image([(x0, y0), (x0 + cw, y0), (x0 + cw, y0 + ch), (x0, y0 + ch)]) + + +# ---------------------------------------------------------------------- +# the statistic +# ---------------------------------------------------------------------- + + +@dataclass(frozen=True) +class Exploration: + """Grid coverage over one window. + + ``seconds`` is ``(n, n)``, row = y and column = x, the time in each cell, + NaN for a cell that is not floor. ``full_coverage_s`` is how long into the + window the last unvisited cell was first entered, or None if some cell + never was. + """ + + n: int + floor: np.ndarray + seconds: np.ndarray + full_coverage_s: float | None + + @property + def cells(self) -> int: + return int(self.floor.sum()) + + @property + def visited(self) -> np.ndarray: + return self.floor & (np.nan_to_num(self.seconds) > 0) + + @property + def cells_visited(self) -> int: + return int(self.visited.sum()) + + @property + def fraction(self) -> float: + return self.cells_visited / self.cells if self.cells else 0.0 + + +def explore( + view, + arena, + n: int, + *, + point: str = CENTROID, + start_frame: int | None = None, + end_frame: int | None = None, +) -> Exploration: + """How many of the arena's ``n x n`` cells the animal entered in a window. + + A cell counts once the body point has been in it for a single scored + frame. Raises :class:`~glider.analysis.behavior.spatial.SpatialError` + without poses, and ``ValueError`` without the frame size, since pixels + cannot be placed on the floor without it. + """ + if n < 1: + raise ValueError(f"A grid needs at least one cell a side, got {n}") + if not view.resolution or not all(view.resolution): + raise ValueError("this session's frame size is unknown, so the arena cannot be placed") + frame = tracking_frame(view, None, point=point) + if start_frame is not None or end_frame is not None: + low = -np.inf if start_frame is None else start_frame + high = np.inf if end_frame is None else end_frame + frame = frame[(frame["frame"] >= low) & (frame["frame"] <= high)].reset_index(drop=True) + + floor = floor_cells(arena, n) + seconds = np.where(floor, 0.0, np.nan) + if frame.empty: + return Exploration(n, floor, seconds, None) + + frames = frame["frame"].to_numpy() + # Each row stands for `stride` frames: rows arrive every `stride` frames. + stride = max(1, int(np.median(np.diff(frames)))) if len(frames) > 1 else 1 + row_s = stride / view.fps if view.fps else 0.0 + + width, height = view.resolution + norm = np.column_stack( + [frame["center_x"].to_numpy() / width, frame["center_y"].to_numpy() / height] + ) + x, y = _to_cm(arena, norm) + finite = np.isfinite(x) & np.isfinite(y) + on = np.zeros(len(x), dtype=bool) + on[finite] = ~np.asarray(arena.outside_cm(x[finite], y[finite]), dtype=bool) + cw, ch = _cell_size(arena, n) + col = np.clip(np.floor(np.where(on, x, 0) / cw).astype(int), 0, n - 1) + row = np.clip(np.floor(np.where(on, y, 0) / ch).astype(int), 0, n - 1) + on &= floor[row, col] + + np.add.at(seconds, (row[on], col[on]), row_s) + + full = None + hits = np.flatnonzero(on) + cells, first = np.unique(row[hits] * n + col[hits], return_index=True) + if floor.any() and len(cells) == int(floor.sum()) and view.fps: + last = hits[first.max()] + full = (int(frames[last]) - int(frames[0])) / view.fps + return Exploration(n, floor, seconds, full) diff --git a/src/glider/analysis/epochs.py b/src/glider/analysis/epochs.py index 13748c0d..d1e73ddc 100644 --- a/src/glider/analysis/epochs.py +++ b/src/glider/analysis/epochs.py @@ -69,6 +69,13 @@ class Metric: Metric("dam_in_nest_frac", "Dam in nest (fraction)", 2), Metric("nose_nest_mean_cm", "Nose to nest (cm)", 1), ) +# Grid coverage over the drawn arena (AnalysisWindow._exploration_columns). +# "_percent", not "_pct": metric_value would read a `_pct` key as seconds. +_EXPLORATION = ( + Metric("cells_entered", "Grid cells entered", 0), + Metric("explored_percent", "Explored %", 1), + Metric("full_coverage_s", "Time to enter every cell (s)", 1), +) _THRESHOLDS = ( Metric("freeze_threshold", "Freeze < (cm/s)", 2), Metric("dart_threshold", "Dart > (cm/s)", 2), @@ -84,7 +91,8 @@ def metric_catalog(rows: Iterable[dict]) -> list[Metric]: Movement; each behaviour as seconds, as a share of the range and as bouts begun in it; each - zone; each device; pup retrieval where a session was scored for it; + zone; grid exploration where an arena was drawn; each device; pup + retrieval where a session was scored for it; then the thresholds each session was scored with. Freezing and darting are always offered -- every row carries them. """ @@ -106,6 +114,8 @@ def metric_catalog(rows: Iterable[dict]) -> list[Metric]: Metric(f"zone_{zone}_entries", f"{zone} entries", 0), Metric(f"zone_{zone}_latency_s", f"{zone} latency (s)", 2), ] + if any("explored_percent" in row for row in rows): + out += list(_EXPLORATION) out += [Metric(f"hw_{device}_on_s", f"{device} on (s)", 2) for device in devices] if any("pups_retrieved" in row for row in rows): out += list(_RETRIEVAL) @@ -181,6 +191,12 @@ def missing_reason(row: dict, key: str) -> str: return "No zone data for this session" if key.startswith("hw_"): return "This session has no such device" + if key in ("cells_entered", "explored_percent"): + return "No arena drawn for this session" + if key == "full_coverage_s": + if key in row: + return "Some cell was never entered in this range" + return "No arena drawn for this session" if key in ("freeze_threshold", "dart_threshold"): if row.get(f"{key}_px_frame") is not None: return "Uncalibrated: shown in px/frame" diff --git a/src/glider/gui/behavior/analysis_window.py b/src/glider/gui/behavior/analysis_window.py index 991c138f..bd0f0c9b 100644 --- a/src/glider/gui/behavior/analysis_window.py +++ b/src/glider/gui/behavior/analysis_window.py @@ -367,6 +367,9 @@ def __init__(self, parent=None): # there). A recording's CSVs are megabytes parsed on the GUI thread. self._recordings: dict[Path, object] = {} self._zones = None + # Each session's drawn arena, by cohort index: ``(arena, source)``, or + # None when the session has none. Looked up once, on first need. + self._arenas: dict[int, tuple | None] = {} self._tour = None self._cohort_root: Path | None = None # Marker files by the folder they sit in, kept for the cohort's @@ -452,6 +455,8 @@ def __init__(self, parent=None): self._build_bout_stepper() self._build_fixes() self._export_btn.clicked.connect(self._export_window) + self._inspector.grid_n.setValue(int(_settings().value(self._GRID_KEY, 3, type=int) or 3)) + self._inspector.grid_n.valueChanged.connect(self._on_grid_changed) self._trail_s.valueChanged.connect(self._apply_trail) self._bar.marker_clicked.connect(self._go_to_marker) self._bar.marker_activated.connect(self._edit_marker) @@ -480,14 +485,14 @@ def __init__(self, parent=None): # construction def _even_out_menus(self) -> None: - """Open, Zones and Export at one width: the widest's, in this platform's font. + """Open, Arena, Zones and Export at one width: the widest's, in this platform's font. A fixed min-width in the stylesheet held only while every label fit - inside it -- true of the macOS font, not of Windows', where the same - three came out 110, 118 and 130 px. Measured after the theme is + inside it -- true of the macOS font, not of Windows', where three of + them came out 110, 118 and 130 px. Measured after the theme is applied, so the padding that makes room for the arrow is counted. """ - buttons = (self._open_btn, self._zones_btn, self._export_menu_btn) + buttons = (self._open_btn, self._arena_btn, self._zones_btn, self._export_menu_btn) for button in buttons: button.ensurePolished() width = max(button.sizeHint().width() for button in buttons) @@ -594,6 +599,13 @@ def _build_top_bar(self) -> QFrame: ], role="primary", ) + self._arena_btn, _ = self._menu_button( + "Arena", + [ + ("Draw arena…", self._draw_arena), + ("Load arena…", self._load_arena), + ], + ) self._zones_btn, _ = self._menu_button( "Zones", [ @@ -628,7 +640,13 @@ def _build_top_bar(self) -> QFrame: ) set_button_role(self._tour_btn, "ghost") self._tour_btn.setObjectName("ReviewHelp") - for button in (self._open_btn, self._zones_btn, self._export_menu_btn, self._tour_btn): + for button in ( + self._open_btn, + self._arena_btn, + self._zones_btn, + self._export_menu_btn, + self._tour_btn, + ): row.addWidget(button) return bar @@ -658,6 +676,7 @@ def _wire_transport(self) -> None: "video for this session can be found." ) t.video_on.toggled.connect(self._canvas.set_show_video) + t.arena_on.toggled.connect(self._canvas.set_show_arena) t.poses_on.toggled.connect(self._canvas.set_show_poses) t.distances_on.toggled.connect(self._canvas.set_show_distances) t.distances_on.customContextMenuRequested.connect( @@ -1027,6 +1046,7 @@ def _set_cohort(self, loaded, *, ids=None, groups=None) -> None: self._invalidate_cohort_cache() self._marker_stores = {} self._retrieval_state, self._retrievals = {}, {} + self._arenas = {} self._cohort_store = cohort_store self._timelines = [timeline for timeline, _ in built] self._recordings_of = [recording for _, recording in built] @@ -1125,6 +1145,7 @@ def _reload_shown(self, *, pose_csv: Path | None = None) -> None: # New poses may move the nose; the scale is re-read with the view. self._retrieval_state.pop(index, None) self._retrievals.pop(index, None) + self._arenas.pop(index, None) self._invalidate_cohort_cache() self._fill_pool(index) self._show_session(index) @@ -1213,6 +1234,9 @@ def _adopt(self, path: Path, view: SessionView, timeline) -> None: self._canvas.set_hud_names(self._bar.behavior_order()) self._bar.set_hidden(_settings().value(self._hidden_key(), [], type=list) or []) self._canvas.set_view(view) + found = self._arena_for(self._shown) + self._canvas.set_arena(found[0] if found else None) + self._canvas.set_grid(self._grid_n()) self._restore_pairs() # The overlay belongs to the session that just left. self._canvas.set_heatmap(None) @@ -2568,6 +2592,7 @@ def _on_selection(self, start: int, end: int) -> None: self._summary.setText(self._describe(stats)) self._fill_epochs() self._fill_zones(start, end) + self._fill_exploration(start, end) self._fill_hardware(start, end) self._apply_heatmap() self._export_btn.setEnabled(True) @@ -2588,6 +2613,7 @@ def _on_selection_cleared(self) -> None: self._fill_epochs() # the cohort's ranges stay, less the current one self._export_btn.setEnabled(False) self._export_window_action.setEnabled(False) + self._canvas.set_visited(None) self._apply_heatmap() def _fill_movement(self, stats) -> None: @@ -2663,6 +2689,156 @@ def _adopt_zones(self) -> None: self._on_selection(*selection) self._fill_epochs() + # ------------------------------------------------------------------ + # the arena and its exploration grid + + #: QSettings key for the grid size: a lab scores every cohort on one grid. + _GRID_KEY = "review/grid_n" + + def _grid_n(self) -> int: + """Cells a side of the exploration grid; below 2 there is none.""" + n = self._inspector.grid_n.value() + return n if n >= 2 else 0 + + def _on_grid_changed(self, n: int) -> None: + _settings().setValue(self._GRID_KEY, int(n)) + self._canvas.set_grid(self._grid_n()) + self._invalidate_cohort_cache() + selection = self._bar.selection() + if selection is not None: + self._fill_exploration(*selection) + self._fill_epochs() + + def _arena_for(self, index: int): + """``(arena, source)`` for a cohort session, or None; looked up once.""" + if not 0 <= index < len(self._cohort): + return None + if index not in self._arenas: + from glider.analysis.behavior.exploration import find_session_arena + + path, view = self._cohort[index] + try: + self._arenas[index] = find_session_arena(path, view) + except (OSError, ValueError): + logger.debug("no arena for %s", path, exc_info=True) + self._arenas[index] = None + return self._arenas[index] + + def _exploration(self, index: int, start: int, end: int): + """``(Exploration, None)`` over frames ``[start, end]``, or ``(None, why not)``.""" + from glider.analysis.behavior import exploration, spatial + + n = self._grid_n() + if not n: + return None, "Pick a grid size to count the cells entered." + found = self._arena_for(index) + if found is None: + return None, "No arena drawn for this session: Arena ▸ Draw arena…" + view = self._cohort[index][1] + try: + result = exploration.explore(view, found[0], n, start_frame=start, end_frame=end) + except (spatial.SpatialError, ValueError) as e: + return None, f"No exploration: {e}." + return result, None + + def _fill_exploration(self, start: int, end: int) -> None: + result, why = self._exploration(self._shown, start, end) + if result is None: + self._inspector.set_exploration("—", "—", "—", why) + self._canvas.set_visited(None) + return + source = self._arena_for(self._shown)[1] + self._inspector.set_exploration( + f"{result.cells_visited} / {result.cells}", + f"{100.0 * result.fraction:.0f}", + "—" if result.full_coverage_s is None else f"{result.full_coverage_s:.1f}", + f"Arena from {source.name}. A cell counts once the body centre enters it.", + ) + self._canvas.set_visited(result.visited) + + def _exploration_columns(self, index: int, start: int, end: int) -> dict: + """Cells entered and the share explored, for a session with an arena and a grid.""" + result, _why = self._exploration(index, start, end) + if result is None: + return {} + return { + "grid_n": result.n, + "cells_entered": result.cells_visited, + "explored_percent": 100.0 * result.fraction, + "full_coverage_s": ( + math.nan if result.full_coverage_s is None else result.full_coverage_s + ), + } + + def _adopt_arena(self, arena, *, title: str) -> None: + """Save *arena* beside the shown session and show it.""" + import json + + from glider.analysis.behavior.exploration import arena_file_for + + path = arena_file_for(self._cohort[self._shown][0]) + try: + path.write_text(json.dumps(arena.to_dict(), indent=2), encoding="utf-8") + except OSError as e: + QMessageBox.critical(self, title, f"Could not save the arena to {path}: {e}") + return + self._arenas[self._shown] = (arena, path) + self._canvas.set_arena(arena) + self._canvas.set_grid(self._grid_n()) + self._invalidate_cohort_cache() + self.statusBar().showMessage(f"Arena saved to {path}", 8000) + selection = self._bar.selection() + if selection is not None: + self._fill_exploration(*selection) + self._fill_epochs() + + def _draw_arena(self) -> None: + """Outline the shown session's floor on the frame on screen, and save it.""" + if self._view is None: + return + frame = self._canvas.current_frame() + if frame is None: + QMessageBox.warning( + self, + "Draw arena", + "The arena is drawn on a video frame, and no video was found for " + "this session.\n\nUse “Load arena…” to bring in one drawn elsewhere.", + ) + return + from glider.gui.dialogs.arena_dialog import ArenaDialog + + dialog = ArenaDialog(frame, title=self._ids[self._shown], parent=self) + try: + found = self._arena_for(self._shown) + if found is not None: + dialog.set_arena(found[0]) + if dialog.exec() != QDialog.DialogCode.Accepted: + return + arena = dialog.calibration() + finally: + dialog.deleteLater() + if arena is not None: + self._adopt_arena(arena, title="Draw arena") + + def _load_arena(self) -> None: + """Use an arena file drawn elsewhere for the shown session.""" + if self._view is None: + return + from glider.analysis.behavior.exploration import read_arena + + path, _ = QFileDialog.getOpenFileName( + self, "Load arena", str(self._session_folder(self._shown)), "Arena files (*.json)" + ) + if not path: + return + arena = read_arena(path) + if arena is None: + QMessageBox.critical( + self, "Load arena", f"{Path(path).name} is not a readable arena file." + ) + return + self._adopt_arena(arena, title="Load arena") + def _clear_zones(self) -> None: self._zones = None self._adopt_zones() @@ -2851,6 +3027,7 @@ def total(state, table=by_state): }, **{f"{state}_bouts": n for state, n in view.bout_onsets(start, end).items()}, **self._zone_columns(start, end, view), + **self._exploration_columns(index, start, end), **self._hardware_columns(timeline, view.fps, start, end), **self._retrieval_columns(index, start, end), } diff --git a/src/glider/gui/review/inspector.py b/src/glider/gui/review/inspector.py index bb6a235f..4e5e940c 100644 --- a/src/glider/gui/review/inspector.py +++ b/src/glider/gui/review/inspector.py @@ -23,6 +23,7 @@ QListWidgetItem, QPushButton, QSizePolicy, + QSpinBox, QTableWidget, QTableWidgetItem, QTabWidget, @@ -113,6 +114,30 @@ def __init__(self, caption: str, parent=None): column.addWidget(label) +#: The grid sizes on offer; 1 means no grid. +GRID_MAX = 20 + + +class GridSpin(QSpinBox): + """Cells a side of the exploration grid, shown as "3 × 3", or "Off".""" + + def __init__(self, parent=None): + super().__init__(parent) + self.setRange(1, GRID_MAX) + self.setSpecialValueText("Off") + # Stepped with the arrows or the wheel: "3 × 3" is not text a plain + # spin box can parse back as it is typed. + self.lineEdit().setReadOnly(True) + self.setToolTip("Lay an N × N grid over the drawn arena and count the cells entered") + + def textFromValue(self, value: int) -> str: # noqa: N802 - Qt override + return f"{value} × {value}" + + def valueFromText(self, text: str) -> int: # noqa: N802 - Qt override + digits = text.split("×")[0].strip() + return int(digits) if digits.isdigit() else self.minimum() + + class Inspector(QTabWidget): """Range, Session, Markers and Retrieval tabs. The window fills them.""" @@ -177,6 +202,19 @@ def _build_range(self) -> QWidget: self.range_text = hint("") column.addWidget(self.range_text) + column.addWidget(_section("Exploration")) + self.grid_n = GridSpin() + column.addLayout(labelled_row("Grid", self.grid_n)) + explore = QHBoxLayout() + explore.setSpacing(6) + tiles = [_Kpi("cells entered"), _Kpi("% explored"), _Kpi("s to cover all")] + for tile in tiles: + explore.addWidget(tile) + column.addLayout(explore) + self.cells_visited, self.explored_pct, self.coverage_s = (t.value for t in tiles) + self.exploration_text = hint("") + column.addWidget(self.exploration_text) + column.addWidget(_section("Behavior")) self.bouts = _grow_with_rows(QTableWidget(0, 6)) self.bouts.setHorizontalHeaderLabels( @@ -274,6 +312,7 @@ def set_range(self, title: str, detail: str) -> None: def clear_range(self) -> None: self.set_range(_NO_RANGE, _NO_RANGE_HINT) self.set_kpis("—", "—", "—") + self.set_exploration("—", "—", "—") self.range_text.setText("") self.set_hardware([]) for table in (self.bouts, self.zones): @@ -289,6 +328,12 @@ def set_kpis(self, distance: str, mean: str, peak: str) -> None: self.mean_speed.setText(mean) self.peak_speed.setText(peak) + def set_exploration(self, cells: str, pct: str, coverage: str, note: str = "") -> None: + self.cells_visited.setText(cells) + self.explored_pct.setText(pct) + self.coverage_s.setText(coverage) + self.exploration_text.setText(note) + def set_hardware(self, rows: list[tuple[str, str, QColor]]) -> None: self.hardware.setRowCount(len(rows)) for r, (name, text, colour) in enumerate(rows): diff --git a/src/glider/gui/review/viewer.py b/src/glider/gui/review/viewer.py index 8f70eec9..ed76bdf9 100644 --- a/src/glider/gui/review/viewer.py +++ b/src/glider/gui/review/viewer.py @@ -118,6 +118,12 @@ def __init__(self, parent=None): self._show_video = True self._zones = None self._heatmap = None + # The drawn floor, the grid laid over it (cells a side, 0 = none) and + # which of its cells the selected range visited (row = y, col = x). + self._arena = None + self._grid_n = 0 + self._visited = None + self._show_arena = True self._reader = None # VideoFileSource, opened lazily self._video_failed = False # this view's video would not open; retried on set_view self._cached: tuple[int, QImage] | None = None @@ -157,6 +163,27 @@ def set_zones(self, zones) -> None: self._zones = zones self.update() + def set_arena(self, arena) -> None: + """Outline a drawn floor (an ``ArenaCalibration``), or clear it with None.""" + self._arena = arena + self._visited = None + self.update() + + def set_grid(self, n: int) -> None: + """Lay an ``n`` x ``n`` grid over the arena; 0 or 1 draws none.""" + self._grid_n = max(0, int(n)) + self._visited = None + self.update() + + def set_visited(self, visited) -> None: + """Shade the grid cells the range entered: ``(n, n)`` bool, or None.""" + self._visited = visited + self.update() + + def set_show_arena(self, enabled: bool) -> None: + self._show_arena = bool(enabled) + self.update() + def current_frame(self): """The decoded BGR frame on screen, or None without a video. @@ -308,6 +335,8 @@ def paintEvent(self, _event): painter.drawImage(arena, self._heatmap) painter.setPen(QPen(QColor(colors.BORDER), 1)) painter.drawRect(arena) + if self._show_arena and self._arena is not None: + self._paint_arena(painter, arena) if self._show_zones: self._paint_zones(painter, arena) @@ -371,6 +400,43 @@ def _paint_zones(self, painter, arena: QRectF) -> None: b, g, r = zone.color self._paint_zone_shape(painter, arena, zone, QPen(QColor(r, g, b), 2)) + def _paint_arena(self, painter, arena: QRectF) -> None: + """The drawn floor, its grid, and the grid cells the range visited. + + The arena is stored normalised, like the zones, so it maps onto the + on-screen rectangle directly. The grid lives on the floor, in + centimetres, so under a tilted camera its far cells are drawn smaller. + """ + from glider.analysis.behavior import exploration + + def to_widget(points): + return [ + QPointF(arena.x() + x * arena.width(), arena.y() + y * arena.height()) + for x, y in points + ] + + colour = QColor(colors.SUCCESS) + try: + n = self._grid_n + if n >= 1 and self._visited is not None and self._visited.shape == (n, n): + painter.setPen(Qt.PenStyle.NoPen) + painter.setBrush(colors.qcolor_with_alpha(colour, 0.18)) + for row, col in zip(*np.nonzero(self._visited), strict=True): + painter.drawPolygon( + *to_widget(exploration.cell_polygon(self._arena, n, int(col), int(row))) + ) + if n >= 2: + painter.setPen(QPen(colors.qcolor_with_alpha(colour, 0.9), 1.5)) + for line in exploration.grid_segments(self._arena, n): + painter.drawPolyline(*to_widget(line)) + width, height = self._view.resolution + outline = [(x / width, y / height) for x, y in self._arena.outline(width, height)] + except ValueError: # a degenerate arena: nothing sensible to draw + return + painter.setPen(QPen(colour, 2)) + painter.setBrush(Qt.BrushStyle.NoBrush) + painter.drawPolygon(*to_widget(outline)) + @staticmethod def _paint_zone_shape(painter, arena: QRectF, zone, pen: QPen) -> None: """One zone's outline, its normalised vertices mapped onto *arena*.""" @@ -753,6 +819,9 @@ def __init__(self, parent=None): "Heatmap", "Where the animal spent the selected range", checked=False ) self.zones_on = _toggle("Zones", "Outline the loaded zones", checked=True) + self.arena_on = _toggle( + "Arena", "Outline the drawn arena floor and its exploration grid", checked=True + ) self.hud_on = _toggle( "HUD", "Name the behaviour and active outputs on each frame", checked=True ) @@ -764,6 +833,7 @@ def __init__(self, parent=None): self.trail_on, self.heatmap_on, self.zones_on, + self.arena_on, self.hud_on, ): row.addWidget(toggle) diff --git a/tests/unit/analysis/behavior/test_exploration.py b/tests/unit/analysis/behavior/test_exploration.py new file mode 100644 index 00000000..7df2093b --- /dev/null +++ b/tests/unit/analysis/behavior/test_exploration.py @@ -0,0 +1,174 @@ +"""Grid exploration over the drawn arena: how many floor cells the animal entered.""" + +from __future__ import annotations + +import json + +import numpy as np +import pandas as pd +import pytest + +from glider.analysis.behavior.exploration import ( + arena_file_for, + cell_polygon, + explore, + find_session_arena, + floor_cells, + grid_segments, + read_arena, +) +from glider.analysis.behavior.session_view import SessionView +from glider.vision.arena import ArenaCalibration + +NAMES = ["nose", "l_ear", "r_ear", "tail_base"] + + +def _view(tmp_path, xs, ys, resolution=(640, 480), fps=30.0): + """A session whose animal walks a path we choose.""" + from glider.vision.pose.core import PoseData + from glider.vision.pose.dlc import to_dlc_csv + + folder = tmp_path / "v" + folder.mkdir(parents=True, exist_ok=True) + n = len(xs) + pd.DataFrame({"frame": range(n), "behavior": ["groom"] * n}).to_csv( + folder / "ethogram_raw.csv", index=False + ) + xy = np.zeros((n, len(NAMES), 2)) + xy[:, :, 0] = np.asarray(xs, dtype=float)[:, None] + xy[:, :, 1] = np.asarray(ys, dtype=float)[:, None] + to_dlc_csv( + PoseData( + xy=xy, + confidence=np.ones((n, len(NAMES))), + keypoint_names=NAMES, + fps=fps, + metadata={"resolution": resolution}, + ), + folder / "vDLC_exp-7.csv", + ) + return SessionView.load(folder / "ethogram_raw.csv") + + +def _full_frame(shape="rectangle"): + """An arena filling the whole 640x480 frame, 40 x 30 cm.""" + return ArenaCalibration( + corners=[(0, 0), (1, 0), (1, 1), (0, 1)], + width_cm=40.0, + height_cm=30.0, + frame_size=(640, 480), + shape=shape, + ) + + +class TestGeometry: + def test_a_rectangle_is_all_floor(self): + assert floor_cells(_full_frame(), 3).all() + + def test_a_round_arena_drops_the_box_corners(self): + floor = floor_cells(_full_frame("ellipse"), 4) + assert not floor[0, 0] and not floor[3, 3] + assert floor[1:3, 1:3].all() + + def test_grid_lines_split_the_floor(self): + lines = grid_segments(_full_frame(), 3) + assert len(lines) == 4 # two vertical, two horizontal + xs = sorted(round(line[0][0], 6) for line in lines if line[0][0] == line[-1][0]) + assert xs == pytest.approx([1 / 3, 2 / 3]) + + def test_no_lines_without_a_grid(self): + assert grid_segments(_full_frame(), 1) == [] + + def test_grid_follows_perspective(self): + """A far wall drawn narrower gives far cells drawn smaller.""" + tilted = ArenaCalibration( + corners=[(0.3, 0.1), (0.7, 0.1), (0.9, 0.9), (0.1, 0.9)], + width_cm=30.0, + height_cm=30.0, + ) + top = cell_polygon(tilted, 3, 1, 0) + bottom = cell_polygon(tilted, 3, 1, 2) + assert abs(top[1][0] - top[0][0]) < abs(bottom[1][0] - bottom[0][0]) + + +class TestExplore: + def test_counts_the_cells_entered(self, tmp_path): + # Along the top row of a 3x3 grid only: three of nine cells. + xs = np.linspace(10, 630, 90) + view = _view(tmp_path, xs, np.full(90, 40.0)) + result = explore(view, _full_frame(), 3) + assert result.cells == 9 + assert result.cells_visited == 3 + assert result.fraction == pytest.approx(1 / 3) + assert result.full_coverage_s is None + assert result.visited[0].all() and not result.visited[1:].any() + + def test_time_in_cells_adds_up_to_the_window(self, tmp_path): + view = _view(tmp_path, np.linspace(10, 630, 90), np.full(90, 40.0)) + result = explore(view, _full_frame(), 3) + assert np.nansum(result.seconds) == pytest.approx(90 / 30.0) + + def test_full_coverage_time(self, tmp_path): + # Visit the four cells of a 2x2 grid, 30 frames each, in turn. + xs = np.repeat([100, 500, 500, 100], 30) + ys = np.repeat([100, 100, 400, 400], 30) + result = explore(_view(tmp_path, xs, ys), _full_frame(), 2) + assert result.fraction == 1.0 + assert result.full_coverage_s == pytest.approx(90 / 30.0) + + def test_the_window_limits_it(self, tmp_path): + xs = np.repeat([100, 500, 500, 100], 30) + ys = np.repeat([100, 100, 400, 400], 30) + result = explore(_view(tmp_path, xs, ys), _full_frame(), 2, start_frame=0, end_frame=59) + assert result.cells_visited == 2 + + def test_positions_off_the_floor_do_not_count(self, tmp_path): + """An arena in the middle of the frame: the animal outside it explores nothing.""" + arena = ArenaCalibration(corners=[(0.25, 0.25), (0.75, 0.25), (0.75, 0.75), (0.25, 0.75)]) + result = explore(_view(tmp_path, np.full(30, 20.0), np.full(30, 20.0)), arena, 3) + assert result.cells_visited == 0 + + def test_a_grid_needs_a_cell(self, tmp_path): + with pytest.raises(ValueError): + explore(_view(tmp_path, [1, 2], [1, 2]), _full_frame(), 0) + + +class TestFindingTheArena: + def test_reads_a_bare_and_a_wrapped_arena(self, tmp_path): + arena = _full_frame() + bare, wrapped = tmp_path / "a.json", tmp_path / "b.json" + bare.write_text(json.dumps(arena.to_dict())) + wrapped.write_text(json.dumps({"arena": arena.to_dict(), "px_per_cm_centre": 16.0})) + assert read_arena(bare) == arena + assert read_arena(wrapped) == arena + + def test_an_unreadable_file_is_none(self, tmp_path): + junk = tmp_path / "x.json" + junk.write_text("not json") + assert read_arena(junk) is None + + def test_a_saved_arena_beside_the_session_is_found(self, tmp_path): + view = _view(tmp_path, [1, 2], [1, 2]) + path = arena_file_for(tmp_path / "v" / "ethogram_raw.csv") + assert path == tmp_path / "v" / "v_arena.json" + path.write_text(json.dumps(_full_frame().to_dict())) + found = find_session_arena(tmp_path / "v" / "ethogram_raw.csv", view) + assert found == (_full_frame(), path) + + def test_the_batch_calibration_is_found_by_video_stem(self, tmp_path): + """The master file keys arenas by the video the batch saw, often elsewhere now.""" + from glider.vision.calibration import CameraCalibration + from glider.vision.calibration_set import CalibrationSet + + view = _view(tmp_path, [1, 2], [1, 2]) + calibrations = CalibrationSet() + elsewhere = tmp_path / "gone" / "v.mp4" + calibrations.set(elsewhere, CameraCalibration()) + calibrations.set_arena(elsewhere, _full_frame()) + calibrations.save(tmp_path / "pose_calibration.json") + found = find_session_arena(tmp_path / "v" / "ethogram_raw.csv", view) + assert found is not None and found[0] == _full_frame() + + def test_none_when_nothing_was_drawn(self, tmp_path): + view = _view(tmp_path, [1, 2], [1, 2]) + assert find_session_arena(tmp_path / "v" / "ethogram_raw.csv", view) is None diff --git a/tests/unit/gui/behavior/test_analysis_window.py b/tests/unit/gui/behavior/test_analysis_window.py index c8fe05c0..e5bca444 100644 --- a/tests/unit/gui/behavior/test_analysis_window.py +++ b/tests/unit/gui/behavior/test_analysis_window.py @@ -3644,3 +3644,99 @@ def test_the_retrieval_keys_come_and_go_with_the_preset(self, qtbot, tmp_path): assert "pups_retrieved" not in win.range_rows(0.0, 3.0)[0] win.set_experiment_type("Pup Retrieval") assert "pups_retrieved" in win.range_rows(0.0, 3.0)[0] + + +class TestArenaGrid: + """The drawn arena, its N x N grid and the share of cells entered.""" + + def _win(self, qtbot, tmp_path, *, with_arena=True): + import json + + from glider.vision.arena import ArenaCalibration + + csv = _session(tmp_path / "v") + if with_arena: + arena = ArenaCalibration(corners=[(0, 0), (1, 0), (1, 1), (0, 1)]) + (tmp_path / "v" / "v_arena.json").write_text(json.dumps(arena.to_dict())) + win = AnalysisWindow() + qtbot.addWidget(win) + win.load(csv) + return win + + def test_a_saved_arena_is_shown(self, qtbot, tmp_path): + win = self._win(qtbot, tmp_path) + assert win._canvas._arena is not None + assert win._canvas._grid_n == 3 # the default grid + + def test_the_range_reports_the_cells_entered(self, qtbot, tmp_path): + # The fixture walks x = 0..299 at y = 100 on a 640x480 frame: on a 3x3 + # grid that is the top row's first two cells. + win = self._win(qtbot, tmp_path) + win._bar.set_selection(0, 299) + assert win._inspector.cells_visited.text() == "2 / 9" + assert win._inspector.explored_pct.text() == "22" + assert win._canvas._visited.sum() == 2 + + def test_changing_the_grid_recounts(self, qtbot, tmp_path): + win = self._win(qtbot, tmp_path) + win._bar.set_selection(0, 299) + win._inspector.grid_n.setValue(2) + assert win._inspector.cells_visited.text() == "1 / 4" + assert win._canvas._grid_n == 2 + + def test_grid_off_draws_no_grid(self, qtbot, tmp_path): + win = self._win(qtbot, tmp_path) + win._bar.set_selection(0, 299) + win._inspector.grid_n.setValue(1) + assert win._inspector.grid_n.text() == "Off" + assert win._canvas._grid_n == 0 + assert win._inspector.cells_visited.text() == "—" + + def test_without_an_arena_it_says_how_to_get_one(self, qtbot, tmp_path): + win = self._win(qtbot, tmp_path, with_arena=False) + win._bar.set_selection(0, 299) + assert win._inspector.cells_visited.text() == "—" + assert "Draw arena" in win._inspector.exploration_text.text() + + def test_exploration_reaches_the_cohort_rows_and_epoch_metrics(self, qtbot, tmp_path): + from glider.analysis.epochs import metric_catalog + + win = self._win(qtbot, tmp_path) + rows = win.cohort_rows(0, 299) + assert rows[0]["cells_entered"] == 2 + assert rows[0]["explored_percent"] == pytest.approx(200 / 9) + assert "explored_percent" in [m.key for m in metric_catalog(rows)] + + def test_an_arena_set_here_is_saved_beside_the_session(self, qtbot, tmp_path): + from glider.analysis.behavior.exploration import read_arena + from glider.vision.arena import ArenaCalibration + + win = self._win(qtbot, tmp_path, with_arena=False) + arena = ArenaCalibration(corners=[(0.1, 0.1), (0.9, 0.1), (0.9, 0.9), (0.1, 0.9)]) + win._adopt_arena(arena, title="Draw arena") + assert read_arena(tmp_path / "v" / "v_arena.json") == arena + assert win._canvas._arena == arena + + def test_the_canvas_draws_the_grid_on_the_floor(self, qtbot, tmp_path): + from glider.vision.arena import ArenaCalibration + + canvas = KeypointCanvas() + qtbot.addWidget(canvas) + canvas.resize(400, 300) + canvas.set_view(SessionView.load(_session(tmp_path / "v"))) + canvas.set_arena(ArenaCalibration(corners=[(0, 0), (1, 0), (1, 1), (0, 1)])) + canvas.set_show_video(False) + + def green_on_first_line(): + # 640x480 fills 400x300 exactly, so a 3x3 grid's first vertical + # line sits at x = 133; the poses walk y ~ 62, well above. + image = canvas.grab().toImage() + column = [image.pixelColor(x, y) for x in (132, 133, 134) for y in range(150, 290)] + return any(c.green() > c.red() + 60 for c in column) + + canvas.set_grid(0) + assert not green_on_first_line() + canvas.set_grid(3) + assert green_on_first_line() + canvas.set_show_arena(False) + assert not green_on_first_line()