From 642d98a2ddbb83b0de3d5679ac94ed652cfb6cba Mon Sep 17 00:00:00 2001 From: Jer Miller Date: Wed, 1 Jul 2026 16:37:48 -0600 Subject: [PATCH] feat(describe): scene-cut + stride frame winnowing; deterministic extraction Add a content-aware stage on top of the existing dHash frame qualification in VideoProcessor.process(): qualified frames at Hamming distance >= 25 are tagged scene cuts and always kept; non-scene-cut qualified frames arriving < 5s after the last KEPT frame are stride-dropped. All three gates (dHash change, scene cut, stride floor) measure against the single last-kept reference, which advances only on a kept frame. Thresholds are config-backed under `describe` (scene_cut_threshold=25, min_stride_seconds=5.0) and honored at both process() call sites via VideoProcessor.__init__. Per-segment winnowing metrics (raw/dhash_qualified/scene_cut/stride_dropped/kept) emit as one INFO line; no behavior is gated on them. The dHash body, metadata header, JSONL schema, and per-sensor change-detection are unchanged. extract.py: importance is now a hard per-category filter (ignore->0, low-><=2 per category, normal/high uncapped) via _apply_category_caps, applied before the first-frame guarantee. The AI-fallback selector is now deterministic greedy max-temporal-spread (seeded lowest-frame_id, lowest-frame_id tie-break) instead of random.sample; the `random` import and stale "advisory" wording are gone. Reimplementation on top of main (not a merge of the ~198-commit-old PR branch). The MobileViT embedding stage is deliberately NOT ported: three local evals proved it inert on screencasts, and it does not earn its complexity or 21 MB wheel weight. No new runtime dependency, no new assets. Design decisions: - Config keys `describe.scene_cut_threshold` (25) / `describe.min_stride_seconds` (5.0); module constants SCENE_CUT_THRESHOLD / MIN_STRIDE_SECONDS are the defaults and the values tests import. No config for DHASH_THRESHOLD. - Per-frame keep decision is a pure module-level _winnow_decision(); process() owns reference and counter state. - Counters: raw = every decoder-yielded frame (incl. pts=None and masked skips); dhash_qualified = first frame + dHash-8 gate passers (== kept + stride_dropped); scene_cut subset of kept. Identities raw >= dhash_qualified >= kept, kept == dhash_qualified - stride_dropped, scene_cut <= kept hold in all cases. Co-Authored-By: Claude Opus 4.8 (1M context) --- solstone/observe/describe.py | 125 +++++++++++++++--- solstone/observe/extract.py | 87 ++++++++++--- tests/test_describe_scene_cut.py | 59 +++++++++ tests/test_describe_stride.py | 214 +++++++++++++++++++++++++++++++ tests/test_extract.py | 20 ++- tests/test_extract_winnowing.py | 125 ++++++++++++++++++ 6 files changed, 579 insertions(+), 51 deletions(-) create mode 100644 tests/test_describe_scene_cut.py create mode 100644 tests/test_describe_stride.py create mode 100644 tests/test_extract_winnowing.py diff --git a/solstone/observe/describe.py b/solstone/observe/describe.py index c5a08f435..633a93121 100644 --- a/solstone/observe/describe.py +++ b/solstone/observe/describe.py @@ -65,6 +65,46 @@ from solstone.think.utils import ( logger = logging.getLogger(__name__) +# Perceptual-distance at/above which a qualified frame is a scene cut and +# bypasses the stride floor unconditionally (out of 64 dHash bits). +SCENE_CUT_THRESHOLD = 25 +# Minimum wall-clock gap (seconds) between kept frames for non-scene-cut, +# dHash-qualified frames; closer arrivals are stride-dropped. +MIN_STRIDE_SECONDS = 5.0 + + +def _winnow_decision( + current_hash: int, + current_timestamp: float, + last_kept_hash: int, + last_kept_timestamp: float, + dhash_threshold: int, + scene_cut_threshold: int, + min_stride_seconds: float, +) -> tuple[bool, bool, str]: + """Decide whether a non-first frame is kept, measured against the last KEPT frame. + + All three gates (dHash change, scene-cut, stride floor) compare against the + last kept frame's hash/timestamp. The reference advances only on a kept frame, + so a frame dropped by either the dHash gate or the stride floor leaves the + reference untouched. + + Returns (keep, scene_cut, reason); reason is one of: + - "below_threshold": dHash change < dhash_threshold; not qualified, not kept. + - "scene_cut": change >= scene_cut_threshold; kept, bypasses the stride floor. + - "stride_dropped": qualified, non-scene-cut, arrived < min_stride_seconds + after the last kept frame; not kept. + - "kept": qualified, non-scene-cut, past the stride floor; kept. + """ + distance = bin(last_kept_hash ^ current_hash).count("1") + if distance < dhash_threshold: + return (False, False, "below_threshold") + if distance >= scene_cut_threshold: + return (True, True, "scene_cut") + if (current_timestamp - last_kept_timestamp) < min_stride_seconds: + return (False, False, "stride_dropped") + return (True, False, "kept") + class RequestType(Enum): """Type of vision analysis request.""" @@ -345,24 +385,39 @@ class VideoProcessor: self.height: Optional[int] = None # Store qualified frames as simple list self.qualified_frames: List[dict] = [] + describe_config = get_config().get("describe", {}) + self.scene_cut_threshold: int = describe_config.get( + "scene_cut_threshold", SCENE_CUT_THRESHOLD + ) + self.min_stride_seconds: float = describe_config.get( + "min_stride_seconds", MIN_STRIDE_SECONDS + ) + self.winnow_metrics: dict = {} def process(self) -> List[dict]: """ Process video and return qualified frames. Uses dHash perceptual hashing to detect significant changes. Caches - the dHash of the last qualified frame for comparison. + the dHash of the last kept frame for comparison. Returns: List of qualified frames with timestamp and frame_bytes. """ - # Cache for the last qualified frame hash - last_hash: Optional[int] = None + # Reference = last KEPT frame; advances only on keep. + last_kept_hash: Optional[int] = None + last_kept_timestamp: float = 0.0 self.first_hash = None self.last_hash = None self.qualified_count = 0 self.decode_failed = False + # Winnowing counters (see the metrics line after the loop for definitions). + raw_frames = 0 + dhash_qualified = 0 + scene_cut_count = 0 + stride_dropped = 0 + # Imports deferred: av (PyAV) and cv2 (via observe.aruco) bundle # mismatched libavdevice majors. Keeping them out of module scope # avoids the macOS ObjC duplicate-class warning on every caller that @@ -385,6 +440,7 @@ class VideoProcessor: frame_count = 0 for frame in container.decode(video=0): + raw_frames += 1 if frame.pts is None: continue @@ -437,46 +493,77 @@ class VideoProcessor: "extrapolated" ] - # First frame: always qualify - if last_hash is None: + # First frame: always kept + if last_kept_hash is None: frame_data["frame_bytes"] = self._frame_to_bytes(pil_img) - last_hash = self._dhash(pil_img) - self.first_hash = last_hash - self.last_hash = last_hash + first_hash = self._dhash(pil_img) + last_kept_hash = first_hash + last_kept_timestamp = timestamp + self.first_hash = first_hash + self.last_hash = first_hash pil_img.close() self.qualified_frames.append(frame_data) + dhash_qualified += 1 logger.debug(f"First frame at {timestamp:.2f}s") continue - # Compare current frame with last qualified using dHash + # Decide against the last KEPT frame (single reference). current_hash = self._dhash(pil_img) - distance = bin(last_hash ^ current_hash).count("1") + keep, scene_cut, reason = _winnow_decision( + current_hash, + timestamp, + last_kept_hash, + last_kept_timestamp, + self.DHASH_THRESHOLD, + self.scene_cut_threshold, + self.min_stride_seconds, + ) - if distance < self.DHASH_THRESHOLD: - # Not enough change - skip this frame + if reason == "below_threshold": pil_img.close() continue - # Qualified - convert full frame to bytes + # Passed the dHash gate. + dhash_qualified += 1 + if not keep: + stride_dropped += 1 + pil_img.close() + continue + + # Kept: convert full frame to bytes and advance the reference. frame_data["frame_bytes"] = self._frame_to_bytes(pil_img) pil_img.close() self.qualified_frames.append(frame_data) - - # Update cached frame hash - last_hash = current_hash + last_kept_hash = current_hash + last_kept_timestamp = timestamp self.last_hash = current_hash + if scene_cut: + scene_cut_count += 1 logger.debug( - f"Qualified frame at {timestamp:.2f}s (hamming: {distance})" + f"Qualified frame at {timestamp:.2f}s (reason: {reason})" ) self.qualified_count = len(self.qualified_frames) + self.winnow_metrics = { + "raw": raw_frames, + "dhash_qualified": dhash_qualified, + "scene_cut": scene_cut_count, + "stride_dropped": stride_dropped, + "kept": self.qualified_count, + } logger.info( - f"Processed {frame_count} frames from {self.video_path.name}, " - f"{len(self.qualified_frames)} qualified" + "winnowing %s raw=%d dhash_qualified=%d scene_cut=%d " + "stride_dropped=%d kept=%d", + self.video_path.name, + raw_frames, + dhash_qualified, + scene_cut_count, + stride_dropped, + self.qualified_count, ) except av.error.InvalidDataError as e: diff --git a/solstone/observe/extract.py b/solstone/observe/extract.py index 436f3ce5a..5bfb5dbf2 100644 --- a/solstone/observe/extract.py +++ b/solstone/observe/extract.py @@ -4,7 +4,7 @@ """Frame extraction selection for vision analysis pipeline. Determines which categorized frames should receive detailed content extraction. -Provides AI-based selection with random fallback. +Provides AI-based selection with a deterministic max-temporal-spread fallback. The first qualified frame is always included regardless of selection results. """ @@ -13,7 +13,6 @@ from __future__ import annotations import json import logging -import random from pathlib import Path from typing import TYPE_CHECKING @@ -40,8 +39,10 @@ def select_frames_for_extraction( The first qualified frame is always included, even if it exceeds max_extractions by one. This ensures we always have context from the start of the recording. - Category importance settings from config (high/normal/low/ignore) are passed - as advisory hints to the AI selection process but are not enforced programmatically. + Category importance settings from config are a hard filter: ``ignore`` drops + the category entirely; ``low`` keeps at most 2 frames per category; + ``normal``/``high`` are uncapped. The first frame is always re-added even + if its category is capped. Parameters ---------- @@ -65,7 +66,7 @@ def select_frames_for_extraction( if not categorized_frames: return [] - # Load config overrides for AI hints (importance is advisory, not a filter) + # Load config overrides; importance is a hard per-category filter (see _apply_category_caps). config_overrides = _get_category_config() # Try AI selection if categories provided @@ -80,6 +81,10 @@ def select_frames_for_extraction( else: selected = _fallback_select_frames(categorized_frames, max_extractions) + # Enforce hard per-category caps before guaranteeing the first frame, + # so the first frame is re-added even if its category is capped. + selected = _apply_category_caps(selected, categorized_frames, config_overrides) + # Ensure first frame is always included first_frame_id = categorized_frames[0]["frame_id"] if first_frame_id not in selected: @@ -103,6 +108,37 @@ def _get_category_config() -> dict[str, dict]: return config.get("describe", {}).get("categories", {}) +def _apply_category_caps( + selected_ids: list[int], + categorized_frames: list[dict[str, Any]], + config_overrides: dict[str, dict], +) -> list[int]: + """Enforce hard per-category caps on a selection. + + importance -> cap: ``ignore`` = 0 (dropped), ``low`` = 2 per category, + ``normal``/``high`` = uncapped. Frames are considered lowest-frame_id first, + so the surviving low-importance frames are deterministic. + """ + id_to_category = { + f["frame_id"]: f.get("analysis", {}).get("primary") for f in categorized_frames + } + caps = {"ignore": 0, "low": 2} + counts: dict[str, int] = {} + kept: list[int] = [] + for frame_id in sorted(selected_ids): + category = id_to_category.get(frame_id) + importance = config_overrides.get(category, {}).get("importance", "normal") + cap = caps.get(importance) + if cap == 0: + continue + if cap is not None: + if counts.get(category, 0) >= cap: + continue + counts[category] = counts.get(category, 0) + 1 + kept.append(frame_id) + return kept + + def _build_extraction_guidance( categories: dict[str, dict], config_overrides: dict[str, dict] | None = None, @@ -286,30 +322,39 @@ def _fallback_select_frames( ) -> list[int]: """Fallback frame selection when AI selection is unavailable. - If total frames <= max_extractions: returns all frames. - Otherwise: returns random sample of max_extractions frames. - - Parameters - ---------- - categorized_frames : list[dict] - List of categorized frame data. - max_extractions : int - Maximum number of frames to select. - - Returns - ------- - list[int] - Selected frame IDs. + If total frames <= max_extractions: returns all frame IDs (input order). + Otherwise: deterministically selects max_extractions frames spread across the + segment's timeline via greedy farthest-point sampling on the timestamp axis, + seeded with the lowest-frame_id frame; ties are broken by lowest frame_id. """ if not categorized_frames: return [] all_ids = [f["frame_id"] for f in categorized_frames] - if len(all_ids) <= max_extractions: return all_ids - return random.sample(all_ids, max_extractions) + frames = [(f["frame_id"], float(f["timestamp"])) for f in categorized_frames] + seed = min(frames, key=lambda p: p[0]) + selected = [seed] + selected_ts = [seed[1]] + remaining = [p for p in frames if p != seed] + + while len(selected) < max_extractions and remaining: + best = None + best_key: tuple[float, int] | None = None + for frame_id, timestamp in remaining: + min_dist = min(abs(timestamp - ts) for ts in selected_ts) + # Maximize distance to the nearest selected frame; tie -> lowest frame_id. + key = (min_dist, -frame_id) + if best_key is None or key > best_key: + best_key = key + best = (frame_id, timestamp) + selected.append(best) + selected_ts.append(best[1]) + remaining.remove(best) + + return [frame_id for frame_id, _ in selected] __all__ = [ diff --git a/tests/test_describe_scene_cut.py b/tests/test_describe_scene_cut.py new file mode 100644 index 000000000..51a0f97b1 --- /dev/null +++ b/tests/test_describe_scene_cut.py @@ -0,0 +1,59 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from PIL import Image + +from solstone.observe import describe as describe_module +from solstone.observe.describe import _winnow_decision + + +def test_winnow_scene_cut_bypasses_stride(): + last_kept_hash = 0 + current_hash = (1 << describe_module.SCENE_CUT_THRESHOLD) - 1 + + assert _winnow_decision( + current_hash, + 0.1, + last_kept_hash, + 0.0, + describe_module.VideoProcessor.DHASH_THRESHOLD, + describe_module.SCENE_CUT_THRESHOLD, + describe_module.MIN_STRIDE_SECONDS, + ) == (True, True, "scene_cut") + + +def test_winnow_below_threshold(): + last_kept_hash = 0 + current_hash = (1 << (describe_module.VideoProcessor.DHASH_THRESHOLD - 1)) - 1 + + assert _winnow_decision( + current_hash, + describe_module.MIN_STRIDE_SECONDS, + last_kept_hash, + 0.0, + describe_module.VideoProcessor.DHASH_THRESHOLD, + describe_module.SCENE_CUT_THRESHOLD, + describe_module.MIN_STRIDE_SECONDS, + ) == (False, False, "below_threshold") + + +def test_dhash_identical_images_have_zero_distance(): + processor = describe_module.VideoProcessor.__new__(describe_module.VideoProcessor) + image = Image.new("RGB", (9, 8)) + + assert bin(processor._dhash(image) ^ processor._dhash(image.copy())).count("1") == 0 + + +def test_dhash_reversed_horizontal_ramps_have_full_distance(): + processor = describe_module.VideoProcessor.__new__(describe_module.VideoProcessor) + ramp = [col * 28 for col in range(9)] + reversed_ramp = [col * 28 for col in reversed(range(9))] + + image = Image.new("RGB", (9, 8)) + image.putdata([(v, v, v) for _row in range(8) for v in ramp]) + reversed_image = Image.new("RGB", (9, 8)) + reversed_image.putdata([(v, v, v) for _row in range(8) for v in reversed_ramp]) + + assert ( + bin(processor._dhash(image) ^ processor._dhash(reversed_image)).count("1") == 64 + ) diff --git a/tests/test_describe_stride.py b/tests/test_describe_stride.py new file mode 100644 index 000000000..e605b021f --- /dev/null +++ b/tests/test_describe_stride.py @@ -0,0 +1,214 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +import types +from pathlib import Path + +import pytest + +from solstone.observe import aruco as aruco_module +from solstone.observe import describe as describe_module +from solstone.observe.describe import _winnow_decision + +av = pytest.importorskip("av") +np = pytest.importorskip("numpy") + + +def _assert_metrics_reconcile(metrics: dict) -> None: + assert metrics["raw"] >= metrics["dhash_qualified"] >= metrics["kept"] + assert metrics["kept"] == metrics["dhash_qualified"] - metrics["stride_dropped"] + assert metrics["scene_cut"] <= metrics["kept"] + + +def _fake_av(monkeypatch, frames: list[object]) -> None: + class FakeStream: + def __init__(self): + self.width = 8 + self.height = 8 + self.thread_type = None + self.codec_context = types.SimpleNamespace(thread_count=0) + + class FakeContainer: + def __init__(self): + self.streams = types.SimpleNamespace(video=[FakeStream()]) + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def decode(self, video=0): + yield from frames + + monkeypatch.setattr(av, "open", lambda _path: FakeContainer()) + monkeypatch.setattr(aruco_module, "detect_markers", lambda _img: None) + + +class _FakeFrame: + def __init__(self, pts: int | None, time: float | None, arr): + self.pts = pts + self.time = time + self._arr = arr + + def to_ndarray(self, format="rgb24"): + return self._arr + + +def _arr(): + return np.zeros((8, 8, 3), dtype=np.uint8) + + +def _run( + monkeypatch, + tmp_path: Path, + frame_specs: list[tuple[int | None, float | None]], + injected_hashes: list[int], + **config, +): + # frame_specs: list of (pts, timestamp). injected_hashes: one per non-pts-None frame. + frames = [_FakeFrame(pts, ts, _arr()) for pts, ts in frame_specs] + _fake_av(monkeypatch, frames) + video_path = tmp_path / "screen.webm" + video_path.write_bytes(b"x") + monkeypatch.setattr( + describe_module, "get_config", lambda: {"describe": config} if config else {} + ) + processor = describe_module.VideoProcessor(video_path) + hashes = iter(injected_hashes) + monkeypatch.setattr(processor, "_dhash", lambda _img: next(hashes)) + kept = processor.process() + return processor, kept + + +def test_winnow_stride_drop_vs_keep(): + last_kept_hash = 0 + current_hash = (1 << describe_module.VideoProcessor.DHASH_THRESHOLD) - 1 + + assert _winnow_decision( + current_hash, + describe_module.MIN_STRIDE_SECONDS - 0.1, + last_kept_hash, + 0.0, + describe_module.VideoProcessor.DHASH_THRESHOLD, + describe_module.SCENE_CUT_THRESHOLD, + describe_module.MIN_STRIDE_SECONDS, + ) == (False, False, "stride_dropped") + + assert _winnow_decision( + current_hash, + describe_module.MIN_STRIDE_SECONDS, + last_kept_hash, + 0.0, + describe_module.VideoProcessor.DHASH_THRESHOLD, + describe_module.SCENE_CUT_THRESHOLD, + describe_module.MIN_STRIDE_SECONDS, + ) == (True, False, "kept") + + +def test_video_processor_uses_config_overrides(monkeypatch, tmp_path): + monkeypatch.setattr( + describe_module, + "get_config", + lambda: {"describe": {"scene_cut_threshold": 30, "min_stride_seconds": 2.0}}, + ) + processor = describe_module.VideoProcessor(tmp_path / "x.webm") + + assert processor.scene_cut_threshold == 30 + assert processor.min_stride_seconds == 2.0 + + +def test_video_processor_defaults_when_config_absent(monkeypatch, tmp_path): + monkeypatch.setattr(describe_module, "get_config", lambda: {}) + processor = describe_module.VideoProcessor(tmp_path / "x.webm") + + assert processor.scene_cut_threshold == describe_module.SCENE_CUT_THRESHOLD + assert processor.min_stride_seconds == describe_module.MIN_STRIDE_SECONDS + + +def test_process_reference_stays_last_kept_for_stride_drop(monkeypatch, tmp_path): + processor, kept = _run( + monkeypatch, + tmp_path, + [(1, 0.0), (2, 1.0), (3, 6.0)], + [0, 0x3FF, 0x3FF], + ) + + assert [frame["frame_id"] for frame in kept] == [1, 3] + assert processor.winnow_metrics == { + "raw": 3, + "dhash_qualified": 3, + "scene_cut": 0, + "stride_dropped": 1, + "kept": 2, + } + _assert_metrics_reconcile(processor.winnow_metrics) + + +def test_process_all_scene_cut_keeps_all_and_bypasses_stride(monkeypatch, tmp_path): + # Timestamps far under min_stride (0.1s apart); every jump is a scene cut, + # so the stride floor never fires and every frame is kept. + processor, kept = _run( + monkeypatch, + tmp_path, + [(1, 0.0), (2, 0.1), (3, 0.2)], + [0, (1 << describe_module.SCENE_CUT_THRESHOLD) - 1, 0], + ) + assert [frame["frame_id"] for frame in kept] == [1, 2, 3] + assert processor.winnow_metrics == { + "raw": 3, + "dhash_qualified": 3, + "scene_cut": 2, + "stride_dropped": 0, + "kept": 3, + } + _assert_metrics_reconcile(processor.winnow_metrics) + + +def test_process_quiet_single_frame_keeps_only_first(monkeypatch, tmp_path): + processor, kept = _run(monkeypatch, tmp_path, [(1, 0.0)], [0]) + assert [frame["frame_id"] for frame in kept] == [1] + assert processor.winnow_metrics == { + "raw": 1, + "dhash_qualified": 1, + "scene_cut": 0, + "stride_dropped": 0, + "kept": 1, + } + _assert_metrics_reconcile(processor.winnow_metrics) + + +def test_process_no_decodable_frames_emits_zeroed_metrics(monkeypatch, tmp_path): + processor, kept = _run(monkeypatch, tmp_path, [(None, None), (None, None)], []) + assert kept == [] + assert processor.winnow_metrics == { + "raw": 2, + "dhash_qualified": 0, + "scene_cut": 0, + "stride_dropped": 0, + "kept": 0, + } + _assert_metrics_reconcile(processor.winnow_metrics) + + +def test_process_honors_min_stride_override(monkeypatch, tmp_path): + # Same sequence as the reference test, but min_stride lowered to 0.5 so the + # 1.0s-later frame is no longer stride-dropped. Result diverges from the + # default (which keeps [1, 3]); here f2 is kept and advances the reference, + # making f3 a dHash-gate drop. + processor, kept = _run( + monkeypatch, + tmp_path, + [(1, 0.0), (2, 1.0), (3, 6.0)], + [0, 0x3FF, 0x3FF], + min_stride_seconds=0.5, + ) + assert [frame["frame_id"] for frame in kept] == [1, 2] + assert processor.winnow_metrics == { + "raw": 3, + "dhash_qualified": 2, + "scene_cut": 0, + "stride_dropped": 0, + "kept": 2, + } + _assert_metrics_reconcile(processor.winnow_metrics) diff --git a/tests/test_extract.py b/tests/test_extract.py index 405c2d5c4..b60fa7639 100644 --- a/tests/test_extract.py +++ b/tests/test_extract.py @@ -53,21 +53,21 @@ def test_exactly_max_returns_all(): assert result == list(range(1, 11)) -def test_more_than_max_returns_around_max(): - """Test that more than max frames returns approximately max count.""" +def test_more_than_max_returns_exactly_max(): + """Test that more than max frames returns max count.""" frames = _make_frames(30) result = select_frames_for_extraction(frames, max_extractions=5) - # May be max or max+1 if first frame wasn't in random selection - assert 5 <= len(result) <= 6 + assert len(result) == 5 + assert 1 in result + assert 30 in result def test_first_frame_always_included(): """Test that first frame is always in selection.""" frames = _make_frames(100) - # Run multiple times to account for randomness - for _ in range(10): - result = select_frames_for_extraction(frames, max_extractions=10) - assert 1 in result, "First frame must always be included" + result = select_frames_for_extraction(frames, max_extractions=10) + assert 1 in result, "First frame must always be included" + assert len(result) == 10 def test_results_sorted(): @@ -89,9 +89,7 @@ def test_max_extractions_of_one(): """Test edge case of max_extractions=1.""" frames = _make_frames(10) result = select_frames_for_extraction(frames, max_extractions=1) - # First frame always included, plus possibly one random - assert 1 in result - assert 1 <= len(result) <= 2 + assert result == [1] def test_non_sequential_frame_ids(): diff --git a/tests/test_extract_winnowing.py b/tests/test_extract_winnowing.py new file mode 100644 index 000000000..52706dae4 --- /dev/null +++ b/tests/test_extract_winnowing.py @@ -0,0 +1,125 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from solstone.observe import extract as extract_module +from solstone.observe.extract import ( + _apply_category_caps, + _fallback_select_frames, + select_frames_for_extraction, +) + + +def _frame(frame_id: int, category: str | None, timestamp: float | None = None) -> dict: + frame = { + "frame_id": frame_id, + "timestamp": float(frame_id) if timestamp is None else timestamp, + "analysis": {}, + } + if category is not None: + frame["analysis"]["primary"] = category + return frame + + +def _frames(count: int) -> list[dict]: + return [_frame(frame_id, "code") for frame_id in range(1, count + 1)] + + +def test_apply_category_caps_semantics(): + categorized_frames = [ + _frame(1, "ignored"), + _frame(2, "ignored"), + _frame(3, "low_priority"), + _frame(4, "low_priority"), + _frame(5, "low_priority"), + _frame(6, "normal_priority"), + _frame(7, "high_priority"), + _frame(8, "unknown"), + _frame(9, None), + ] + selected_ids = [9, 8, 7, 6, 5, 4, 3, 2, 1] + config_overrides = { + "ignored": {"importance": "ignore"}, + "low_priority": {"importance": "low"}, + "normal_priority": {"importance": "normal"}, + "high_priority": {"importance": "high"}, + } + + assert _apply_category_caps(selected_ids, categorized_frames, config_overrides) == [ + 3, + 4, + 6, + 7, + 8, + 9, + ] + + +def test_fallback_select_frames_is_deterministic_and_spread(): + categorized_frames = _frames(30) + + result1 = _fallback_select_frames(categorized_frames, max_extractions=5) + result2 = _fallback_select_frames(categorized_frames, max_extractions=5) + + assert result1 == result2 + assert len(result1) == 5 + assert 1 in result1 + assert 30 in result1 + + timestamps = {frame["frame_id"]: frame["timestamp"] for frame in categorized_frames} + selected_timestamps = sorted(timestamps[frame_id] for frame_id in result1) + adjacent_gaps = [ + right - left + for left, right in zip(selected_timestamps, selected_timestamps[1:]) + ] + assert min(adjacent_gaps) >= 5 + + +def test_fallback_select_frames_returns_all_when_under_max_and_empty(): + categorized_frames = _frames(3) + + assert _fallback_select_frames(categorized_frames, max_extractions=5) == [1, 2, 3] + assert _fallback_select_frames([], max_extractions=5) == [] + + +def test_select_frames_readds_first_frame_when_ignore_capped(monkeypatch): + monkeypatch.setattr( + extract_module, + "_get_category_config", + lambda: {"private": {"importance": "ignore"}}, + ) + categorized_frames = [ + _frame(1, "private"), + _frame(2, "private"), + ] + + result = select_frames_for_extraction( + categorized_frames, max_extractions=5, categories=None + ) + + assert result == [1] + + +def test_select_frames_applies_caps_with_fallback_and_sorts(monkeypatch): + monkeypatch.setattr( + extract_module, + "_get_category_config", + lambda: { + "private": {"importance": "ignore"}, + "low_priority": {"importance": "low"}, + }, + ) + categorized_frames = [ + _frame(1, "private"), + _frame(2, "low_priority"), + _frame(3, "low_priority"), + _frame(4, "low_priority"), + _frame(5, "normal_priority"), + _frame(6, "high_priority"), + _frame(7, "private"), + ] + + result = select_frames_for_extraction( + categorized_frames, max_extractions=10, categories=None + ) + + assert result == [1, 2, 3, 5, 6] -- 2.51.2