feat(describe): scene-cut + stride frame winnowing; deterministic extraction · solpbc.org/solstone@642d98a

+106 -19

solstone/observe/describe.py

··· 65 65 66 66 logger = logging.getLogger(__name__) 67 67 68 + # Perceptual-distance at/above which a qualified frame is a scene cut and 69 + # bypasses the stride floor unconditionally (out of 64 dHash bits). 70 + SCENE_CUT_THRESHOLD = 25 71 + # Minimum wall-clock gap (seconds) between kept frames for non-scene-cut, 72 + # dHash-qualified frames; closer arrivals are stride-dropped. 73 + MIN_STRIDE_SECONDS = 5.0 74 + 75 + 76 + def _winnow_decision( 77 + current_hash: int, 78 + current_timestamp: float, 79 + last_kept_hash: int, 80 + last_kept_timestamp: float, 81 + dhash_threshold: int, 82 + scene_cut_threshold: int, 83 + min_stride_seconds: float, 84 + ) -> tuple[bool, bool, str]: 85 + """Decide whether a non-first frame is kept, measured against the last KEPT frame. 86 + 87 + All three gates (dHash change, scene-cut, stride floor) compare against the 88 + last kept frame's hash/timestamp. The reference advances only on a kept frame, 89 + so a frame dropped by either the dHash gate or the stride floor leaves the 90 + reference untouched. 91 + 92 + Returns (keep, scene_cut, reason); reason is one of: 93 + - "below_threshold": dHash change < dhash_threshold; not qualified, not kept. 94 + - "scene_cut": change >= scene_cut_threshold; kept, bypasses the stride floor. 95 + - "stride_dropped": qualified, non-scene-cut, arrived < min_stride_seconds 96 + after the last kept frame; not kept. 97 + - "kept": qualified, non-scene-cut, past the stride floor; kept. 98 + """ 99 + distance = bin(last_kept_hash ^ current_hash).count("1") 100 + if distance < dhash_threshold: 101 + return (False, False, "below_threshold") 102 + if distance >= scene_cut_threshold: 103 + return (True, True, "scene_cut") 104 + if (current_timestamp - last_kept_timestamp) < min_stride_seconds: 105 + return (False, False, "stride_dropped") 106 + return (True, False, "kept") 107 + 68 108 69 109 class RequestType(Enum): 70 110 """Type of vision analysis request.""" ··· 345 385 self.height: Optional[int] = None 346 386 # Store qualified frames as simple list 347 387 self.qualified_frames: List[dict] = [] 388 + describe_config = get_config().get("describe", {}) 389 + self.scene_cut_threshold: int = describe_config.get( 390 + "scene_cut_threshold", SCENE_CUT_THRESHOLD 391 + ) 392 + self.min_stride_seconds: float = describe_config.get( 393 + "min_stride_seconds", MIN_STRIDE_SECONDS 394 + ) 395 + self.winnow_metrics: dict = {} 348 396 349 397 def process(self) -> List[dict]: 350 398 """ 351 399 Process video and return qualified frames. 352 400 353 401 Uses dHash perceptual hashing to detect significant changes. Caches 354 - the dHash of the last qualified frame for comparison. 402 + the dHash of the last kept frame for comparison. 355 403 356 404 Returns: 357 405 List of qualified frames with timestamp and frame_bytes. 358 406 """ 359 - # Cache for the last qualified frame hash 360 - last_hash: Optional[int] = None 407 + # Reference = last KEPT frame; advances only on keep. 408 + last_kept_hash: Optional[int] = None 409 + last_kept_timestamp: float = 0.0 361 410 self.first_hash = None 362 411 self.last_hash = None 363 412 self.qualified_count = 0 364 413 self.decode_failed = False 414 + 415 + # Winnowing counters (see the metrics line after the loop for definitions). 416 + raw_frames = 0 417 + dhash_qualified = 0 418 + scene_cut_count = 0 419 + stride_dropped = 0 365 420 366 421 # Imports deferred: av (PyAV) and cv2 (via observe.aruco) bundle 367 422 # mismatched libavdevice majors. Keeping them out of module scope ··· 385 440 386 441 frame_count = 0 387 442 for frame in container.decode(video=0): 443 + raw_frames += 1 388 444 if frame.pts is None: 389 445 continue 390 446 ··· 437 493 "extrapolated" 438 494 ] 439 495 440 - # First frame: always qualify 441 - if last_hash is None: 496 + # First frame: always kept 497 + if last_kept_hash is None: 442 498 frame_data["frame_bytes"] = self._frame_to_bytes(pil_img) 443 - last_hash = self._dhash(pil_img) 444 - self.first_hash = last_hash 445 - self.last_hash = last_hash 499 + first_hash = self._dhash(pil_img) 500 + last_kept_hash = first_hash 501 + last_kept_timestamp = timestamp 502 + self.first_hash = first_hash 503 + self.last_hash = first_hash 446 504 pil_img.close() 447 505 448 506 self.qualified_frames.append(frame_data) 507 + dhash_qualified += 1 449 508 450 509 logger.debug(f"First frame at {timestamp:.2f}s") 451 510 continue 452 511 453 - # Compare current frame with last qualified using dHash 512 + # Decide against the last KEPT frame (single reference). 454 513 current_hash = self._dhash(pil_img) 455 - distance = bin(last_hash ^ current_hash).count("1") 514 + keep, scene_cut, reason = _winnow_decision( 515 + current_hash, 516 + timestamp, 517 + last_kept_hash, 518 + last_kept_timestamp, 519 + self.DHASH_THRESHOLD, 520 + self.scene_cut_threshold, 521 + self.min_stride_seconds, 522 + ) 523 + 524 + if reason == "below_threshold": 525 + pil_img.close() 526 + continue 456 527 457 - if distance < self.DHASH_THRESHOLD: 458 - # Not enough change - skip this frame 528 + # Passed the dHash gate. 529 + dhash_qualified += 1 530 + if not keep: 531 + stride_dropped += 1 459 532 pil_img.close() 460 533 continue 461 534 462 - # Qualified - convert full frame to bytes 535 + # Kept: convert full frame to bytes and advance the reference. 463 536 frame_data["frame_bytes"] = self._frame_to_bytes(pil_img) 464 537 pil_img.close() 465 538 466 539 self.qualified_frames.append(frame_data) 467 - 468 - # Update cached frame hash 469 - last_hash = current_hash 540 + last_kept_hash = current_hash 541 + last_kept_timestamp = timestamp 470 542 self.last_hash = current_hash 543 + if scene_cut: 544 + scene_cut_count += 1 471 545 472 546 logger.debug( 473 - f"Qualified frame at {timestamp:.2f}s (hamming: {distance})" 547 + f"Qualified frame at {timestamp:.2f}s (reason: {reason})" 474 548 ) 475 549 476 550 self.qualified_count = len(self.qualified_frames) 551 + self.winnow_metrics = { 552 + "raw": raw_frames, 553 + "dhash_qualified": dhash_qualified, 554 + "scene_cut": scene_cut_count, 555 + "stride_dropped": stride_dropped, 556 + "kept": self.qualified_count, 557 + } 477 558 logger.info( 478 - f"Processed {frame_count} frames from {self.video_path.name}, " 479 - f"{len(self.qualified_frames)} qualified" 559 + "winnowing %s raw=%d dhash_qualified=%d scene_cut=%d " 560 + "stride_dropped=%d kept=%d", 561 + self.video_path.name, 562 + raw_frames, 563 + dhash_qualified, 564 + scene_cut_count, 565 + stride_dropped, 566 + self.qualified_count, 480 567 ) 481 568 482 569 except av.error.InvalidDataError as e:

+66 -21

solstone/observe/extract.py

View file

··· 4 4 """Frame extraction selection for vision analysis pipeline. 5 5 6 6 Determines which categorized frames should receive detailed content extraction. 7 - Provides AI-based selection with random fallback. 7 + Provides AI-based selection with a deterministic max-temporal-spread fallback. 8 8 9 9 The first qualified frame is always included regardless of selection results. 10 10 """ ··· 13 13 14 14 import json 15 15 import logging 16 - import random 17 16 from pathlib import Path 18 17 from typing import TYPE_CHECKING 19 18 ··· 40 39 The first qualified frame is always included, even if it exceeds max_extractions 41 40 by one. This ensures we always have context from the start of the recording. 42 41 43 - Category importance settings from config (high/normal/low/ignore) are passed 44 - as advisory hints to the AI selection process but are not enforced programmatically. 42 + Category importance settings from config are a hard filter: ``ignore`` drops 43 + the category entirely; ``low`` keeps at most 2 frames per category; 44 + ``normal``/``high`` are uncapped. The first frame is always re-added even 45 + if its category is capped. 45 46 46 47 Parameters 47 48 ---------- ··· 65 66 if not categorized_frames: 66 67 return [] 67 68 68 - # Load config overrides for AI hints (importance is advisory, not a filter) 69 + # Load config overrides; importance is a hard per-category filter (see _apply_category_caps). 69 70 config_overrides = _get_category_config() 70 71 71 72 # Try AI selection if categories provided ··· 80 81 else: 81 82 selected = _fallback_select_frames(categorized_frames, max_extractions) 82 83 84 + # Enforce hard per-category caps before guaranteeing the first frame, 85 + # so the first frame is re-added even if its category is capped. 86 + selected = _apply_category_caps(selected, categorized_frames, config_overrides) 87 + 83 88 # Ensure first frame is always included 84 89 first_frame_id = categorized_frames[0]["frame_id"] 85 90 if first_frame_id not in selected: ··· 101 106 102 107 config = get_config() 103 108 return config.get("describe", {}).get("categories", {}) 109 + 110 + 111 + def _apply_category_caps( 112 + selected_ids: list[int], 113 + categorized_frames: list[dict[str, Any]], 114 + config_overrides: dict[str, dict], 115 + ) -> list[int]: 116 + """Enforce hard per-category caps on a selection. 117 + 118 + importance -> cap: ``ignore`` = 0 (dropped), ``low`` = 2 per category, 119 + ``normal``/``high`` = uncapped. Frames are considered lowest-frame_id first, 120 + so the surviving low-importance frames are deterministic. 121 + """ 122 + id_to_category = { 123 + f["frame_id"]: f.get("analysis", {}).get("primary") for f in categorized_frames 124 + } 125 + caps = {"ignore": 0, "low": 2} 126 + counts: dict[str, int] = {} 127 + kept: list[int] = [] 128 + for frame_id in sorted(selected_ids): 129 + category = id_to_category.get(frame_id) 130 + importance = config_overrides.get(category, {}).get("importance", "normal") 131 + cap = caps.get(importance) 132 + if cap == 0: 133 + continue 134 + if cap is not None: 135 + if counts.get(category, 0) >= cap: 136 + continue 137 + counts[category] = counts.get(category, 0) + 1 138 + kept.append(frame_id) 139 + return kept 104 140 105 141 106 142 def _build_extraction_guidance( ··· 286 322 ) -> list[int]: 287 323 """Fallback frame selection when AI selection is unavailable. 288 324 289 - If total frames <= max_extractions: returns all frames. 290 - Otherwise: returns random sample of max_extractions frames. 291 - 292 - Parameters 293 - ---------- 294 - categorized_frames : list[dict] 295 - List of categorized frame data. 296 - max_extractions : int 297 - Maximum number of frames to select. 298 - 299 - Returns 300 - ------- 301 - list[int] 302 - Selected frame IDs. 325 + If total frames <= max_extractions: returns all frame IDs (input order). 326 + Otherwise: deterministically selects max_extractions frames spread across the 327 + segment's timeline via greedy farthest-point sampling on the timestamp axis, 328 + seeded with the lowest-frame_id frame; ties are broken by lowest frame_id. 303 329 """ 304 330 if not categorized_frames: 305 331 return [] 306 332 307 333 all_ids = [f["frame_id"] for f in categorized_frames] 308 - 309 334 if len(all_ids) <= max_extractions: 310 335 return all_ids 311 336 312 - return random.sample(all_ids, max_extractions) 337 + frames = [(f["frame_id"], float(f["timestamp"])) for f in categorized_frames] 338 + seed = min(frames, key=lambda p: p[0]) 339 + selected = [seed] 340 + selected_ts = [seed[1]] 341 + remaining = [p for p in frames if p != seed] 342 + 343 + while len(selected) < max_extractions and remaining: 344 + best = None 345 + best_key: tuple[float, int] | None = None 346 + for frame_id, timestamp in remaining: 347 + min_dist = min(abs(timestamp - ts) for ts in selected_ts) 348 + # Maximize distance to the nearest selected frame; tie -> lowest frame_id. 349 + key = (min_dist, -frame_id) 350 + if best_key is None or key > best_key: 351 + best_key = key 352 + best = (frame_id, timestamp) 353 + selected.append(best) 354 + selected_ts.append(best[1]) 355 + remaining.remove(best) 356 + 357 + return [frame_id for frame_id, _ in selected] 313 358 314 359 315 360 __all__ = [

+59

tests/test_describe_scene_cut.py

View file

··· 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 + # Copyright (c) 2026 sol pbc 3 + 4 + from PIL import Image 5 + 6 + from solstone.observe import describe as describe_module 7 + from solstone.observe.describe import _winnow_decision 8 + 9 + 10 + def test_winnow_scene_cut_bypasses_stride(): 11 + last_kept_hash = 0 12 + current_hash = (1 << describe_module.SCENE_CUT_THRESHOLD) - 1 13 + 14 + assert _winnow_decision( 15 + current_hash, 16 + 0.1, 17 + last_kept_hash, 18 + 0.0, 19 + describe_module.VideoProcessor.DHASH_THRESHOLD, 20 + describe_module.SCENE_CUT_THRESHOLD, 21 + describe_module.MIN_STRIDE_SECONDS, 22 + ) == (True, True, "scene_cut") 23 + 24 + 25 + def test_winnow_below_threshold(): 26 + last_kept_hash = 0 27 + current_hash = (1 << (describe_module.VideoProcessor.DHASH_THRESHOLD - 1)) - 1 28 + 29 + assert _winnow_decision( 30 + current_hash, 31 + describe_module.MIN_STRIDE_SECONDS, 32 + last_kept_hash, 33 + 0.0, 34 + describe_module.VideoProcessor.DHASH_THRESHOLD, 35 + describe_module.SCENE_CUT_THRESHOLD, 36 + describe_module.MIN_STRIDE_SECONDS, 37 + ) == (False, False, "below_threshold") 38 + 39 + 40 + def test_dhash_identical_images_have_zero_distance(): 41 + processor = describe_module.VideoProcessor.__new__(describe_module.VideoProcessor) 42 + image = Image.new("RGB", (9, 8)) 43 + 44 + assert bin(processor._dhash(image) ^ processor._dhash(image.copy())).count("1") == 0 45 + 46 + 47 + def test_dhash_reversed_horizontal_ramps_have_full_distance(): 48 + processor = describe_module.VideoProcessor.__new__(describe_module.VideoProcessor) 49 + ramp = [col * 28 for col in range(9)] 50 + reversed_ramp = [col * 28 for col in reversed(range(9))] 51 + 52 + image = Image.new("RGB", (9, 8)) 53 + image.putdata([(v, v, v) for _row in range(8) for v in ramp]) 54 + reversed_image = Image.new("RGB", (9, 8)) 55 + reversed_image.putdata([(v, v, v) for _row in range(8) for v in reversed_ramp]) 56 + 57 + assert ( 58 + bin(processor._dhash(image) ^ processor._dhash(reversed_image)).count("1") == 64 59 + )

+214

tests/test_describe_stride.py

View file

··· 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 + # Copyright (c) 2026 sol pbc 3 + 4 + import types 5 + from pathlib import Path 6 + 7 + import pytest 8 + 9 + from solstone.observe import aruco as aruco_module 10 + from solstone.observe import describe as describe_module 11 + from solstone.observe.describe import _winnow_decision 12 + 13 + av = pytest.importorskip("av") 14 + np = pytest.importorskip("numpy") 15 + 16 + 17 + def _assert_metrics_reconcile(metrics: dict) -> None: 18 + assert metrics["raw"] >= metrics["dhash_qualified"] >= metrics["kept"] 19 + assert metrics["kept"] == metrics["dhash_qualified"] - metrics["stride_dropped"] 20 + assert metrics["scene_cut"] <= metrics["kept"] 21 + 22 + 23 + def _fake_av(monkeypatch, frames: list[object]) -> None: 24 + class FakeStream: 25 + def __init__(self): 26 + self.width = 8 27 + self.height = 8 28 + self.thread_type = None 29 + self.codec_context = types.SimpleNamespace(thread_count=0) 30 + 31 + class FakeContainer: 32 + def __init__(self): 33 + self.streams = types.SimpleNamespace(video=[FakeStream()]) 34 + 35 + def __enter__(self): 36 + return self 37 + 38 + def __exit__(self, *exc): 39 + return False 40 + 41 + def decode(self, video=0): 42 + yield from frames 43 + 44 + monkeypatch.setattr(av, "open", lambda _path: FakeContainer()) 45 + monkeypatch.setattr(aruco_module, "detect_markers", lambda _img: None) 46 + 47 + 48 + class _FakeFrame: 49 + def __init__(self, pts: int | None, time: float | None, arr): 50 + self.pts = pts 51 + self.time = time 52 + self._arr = arr 53 + 54 + def to_ndarray(self, format="rgb24"): 55 + return self._arr 56 + 57 + 58 + def _arr(): 59 + return np.zeros((8, 8, 3), dtype=np.uint8) 60 + 61 + 62 + def _run( 63 + monkeypatch, 64 + tmp_path: Path, 65 + frame_specs: list[tuple[int | None, float | None]], 66 + injected_hashes: list[int], 67 + **config, 68 + ): 69 + # frame_specs: list of (pts, timestamp). injected_hashes: one per non-pts-None frame. 70 + frames = [_FakeFrame(pts, ts, _arr()) for pts, ts in frame_specs] 71 + _fake_av(monkeypatch, frames) 72 + video_path = tmp_path / "screen.webm" 73 + video_path.write_bytes(b"x") 74 + monkeypatch.setattr( 75 + describe_module, "get_config", lambda: {"describe": config} if config else {} 76 + ) 77 + processor = describe_module.VideoProcessor(video_path) 78 + hashes = iter(injected_hashes) 79 + monkeypatch.setattr(processor, "_dhash", lambda _img: next(hashes)) 80 + kept = processor.process() 81 + return processor, kept 82 + 83 + 84 + def test_winnow_stride_drop_vs_keep(): 85 + last_kept_hash = 0 86 + current_hash = (1 << describe_module.VideoProcessor.DHASH_THRESHOLD) - 1 87 + 88 + assert _winnow_decision( 89 + current_hash, 90 + describe_module.MIN_STRIDE_SECONDS - 0.1, 91 + last_kept_hash, 92 + 0.0, 93 + describe_module.VideoProcessor.DHASH_THRESHOLD, 94 + describe_module.SCENE_CUT_THRESHOLD, 95 + describe_module.MIN_STRIDE_SECONDS, 96 + ) == (False, False, "stride_dropped") 97 + 98 + assert _winnow_decision( 99 + current_hash, 100 + describe_module.MIN_STRIDE_SECONDS, 101 + last_kept_hash, 102 + 0.0, 103 + describe_module.VideoProcessor.DHASH_THRESHOLD, 104 + describe_module.SCENE_CUT_THRESHOLD, 105 + describe_module.MIN_STRIDE_SECONDS, 106 + ) == (True, False, "kept") 107 + 108 + 109 + def test_video_processor_uses_config_overrides(monkeypatch, tmp_path): 110 + monkeypatch.setattr( 111 + describe_module, 112 + "get_config", 113 + lambda: {"describe": {"scene_cut_threshold": 30, "min_stride_seconds": 2.0}}, 114 + ) 115 + processor = describe_module.VideoProcessor(tmp_path / "x.webm") 116 + 117 + assert processor.scene_cut_threshold == 30 118 + assert processor.min_stride_seconds == 2.0 119 + 120 + 121 + def test_video_processor_defaults_when_config_absent(monkeypatch, tmp_path): 122 + monkeypatch.setattr(describe_module, "get_config", lambda: {}) 123 + processor = describe_module.VideoProcessor(tmp_path / "x.webm") 124 + 125 + assert processor.scene_cut_threshold == describe_module.SCENE_CUT_THRESHOLD 126 + assert processor.min_stride_seconds == describe_module.MIN_STRIDE_SECONDS 127 + 128 + 129 + def test_process_reference_stays_last_kept_for_stride_drop(monkeypatch, tmp_path): 130 + processor, kept = _run( 131 + monkeypatch, 132 + tmp_path, 133 + [(1, 0.0), (2, 1.0), (3, 6.0)], 134 + [0, 0x3FF, 0x3FF], 135 + ) 136 + 137 + assert [frame["frame_id"] for frame in kept] == [1, 3] 138 + assert processor.winnow_metrics == { 139 + "raw": 3, 140 + "dhash_qualified": 3, 141 + "scene_cut": 0, 142 + "stride_dropped": 1, 143 + "kept": 2, 144 + } 145 + _assert_metrics_reconcile(processor.winnow_metrics) 146 + 147 + 148 + def test_process_all_scene_cut_keeps_all_and_bypasses_stride(monkeypatch, tmp_path): 149 + # Timestamps far under min_stride (0.1s apart); every jump is a scene cut, 150 + # so the stride floor never fires and every frame is kept. 151 + processor, kept = _run( 152 + monkeypatch, 153 + tmp_path, 154 + [(1, 0.0), (2, 0.1), (3, 0.2)], 155 + [0, (1 << describe_module.SCENE_CUT_THRESHOLD) - 1, 0], 156 + ) 157 + assert [frame["frame_id"] for frame in kept] == [1, 2, 3] 158 + assert processor.winnow_metrics == { 159 + "raw": 3, 160 + "dhash_qualified": 3, 161 + "scene_cut": 2, 162 + "stride_dropped": 0, 163 + "kept": 3, 164 + } 165 + _assert_metrics_reconcile(processor.winnow_metrics) 166 + 167 + 168 + def test_process_quiet_single_frame_keeps_only_first(monkeypatch, tmp_path): 169 + processor, kept = _run(monkeypatch, tmp_path, [(1, 0.0)], [0]) 170 + assert [frame["frame_id"] for frame in kept] == [1] 171 + assert processor.winnow_metrics == { 172 + "raw": 1, 173 + "dhash_qualified": 1, 174 + "scene_cut": 0, 175 + "stride_dropped": 0, 176 + "kept": 1, 177 + } 178 + _assert_metrics_reconcile(processor.winnow_metrics) 179 + 180 + 181 + def test_process_no_decodable_frames_emits_zeroed_metrics(monkeypatch, tmp_path): 182 + processor, kept = _run(monkeypatch, tmp_path, [(None, None), (None, None)], []) 183 + assert kept == [] 184 + assert processor.winnow_metrics == { 185 + "raw": 2, 186 + "dhash_qualified": 0, 187 + "scene_cut": 0, 188 + "stride_dropped": 0, 189 + "kept": 0, 190 + } 191 + _assert_metrics_reconcile(processor.winnow_metrics) 192 + 193 + 194 + def test_process_honors_min_stride_override(monkeypatch, tmp_path): 195 + # Same sequence as the reference test, but min_stride lowered to 0.5 so the 196 + # 1.0s-later frame is no longer stride-dropped. Result diverges from the 197 + # default (which keeps [1, 3]); here f2 is kept and advances the reference, 198 + # making f3 a dHash-gate drop. 199 + processor, kept = _run( 200 + monkeypatch, 201 + tmp_path, 202 + [(1, 0.0), (2, 1.0), (3, 6.0)], 203 + [0, 0x3FF, 0x3FF], 204 + min_stride_seconds=0.5, 205 + ) 206 + assert [frame["frame_id"] for frame in kept] == [1, 2] 207 + assert processor.winnow_metrics == { 208 + "raw": 3, 209 + "dhash_qualified": 2, 210 + "scene_cut": 0, 211 + "stride_dropped": 0, 212 + "kept": 2, 213 + } 214 + _assert_metrics_reconcile(processor.winnow_metrics)

+9 -11

tests/test_extract.py

View file

··· 53 53 assert result == list(range(1, 11)) 54 54 55 55 56 - def test_more_than_max_returns_around_max(): 57 - """Test that more than max frames returns approximately max count.""" 56 + def test_more_than_max_returns_exactly_max(): 57 + """Test that more than max frames returns max count.""" 58 58 frames = _make_frames(30) 59 59 result = select_frames_for_extraction(frames, max_extractions=5) 60 - # May be max or max+1 if first frame wasn't in random selection 61 - assert 5 <= len(result) <= 6 60 + assert len(result) == 5 61 + assert 1 in result 62 + assert 30 in result 62 63 63 64 64 65 def test_first_frame_always_included(): 65 66 """Test that first frame is always in selection.""" 66 67 frames = _make_frames(100) 67 - # Run multiple times to account for randomness 68 - for _ in range(10): 69 - result = select_frames_for_extraction(frames, max_extractions=10) 70 - assert 1 in result, "First frame must always be included" 68 + result = select_frames_for_extraction(frames, max_extractions=10) 69 + assert 1 in result, "First frame must always be included" 70 + assert len(result) == 10 71 71 72 72 73 73 def test_results_sorted(): ··· 89 89 """Test edge case of max_extractions=1.""" 90 90 frames = _make_frames(10) 91 91 result = select_frames_for_extraction(frames, max_extractions=1) 92 - # First frame always included, plus possibly one random 93 - assert 1 in result 94 - assert 1 <= len(result) <= 2 92 + assert result == [1] 95 93 96 94 97 95 def test_non_sequential_frame_ids():

+125

tests/test_extract_winnowing.py

View file

··· 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 + # Copyright (c) 2026 sol pbc 3 + 4 + from solstone.observe import extract as extract_module 5 + from solstone.observe.extract import ( 6 + _apply_category_caps, 7 + _fallback_select_frames, 8 + select_frames_for_extraction, 9 + ) 10 + 11 + 12 + def _frame(frame_id: int, category: str | None, timestamp: float | None = None) -> dict: 13 + frame = { 14 + "frame_id": frame_id, 15 + "timestamp": float(frame_id) if timestamp is None else timestamp, 16 + "analysis": {}, 17 + } 18 + if category is not None: 19 + frame["analysis"]["primary"] = category 20 + return frame 21 + 22 + 23 + def _frames(count: int) -> list[dict]: 24 + return [_frame(frame_id, "code") for frame_id in range(1, count + 1)] 25 + 26 + 27 + def test_apply_category_caps_semantics(): 28 + categorized_frames = [ 29 + _frame(1, "ignored"), 30 + _frame(2, "ignored"), 31 + _frame(3, "low_priority"), 32 + _frame(4, "low_priority"), 33 + _frame(5, "low_priority"), 34 + _frame(6, "normal_priority"), 35 + _frame(7, "high_priority"), 36 + _frame(8, "unknown"), 37 + _frame(9, None), 38 + ] 39 + selected_ids = [9, 8, 7, 6, 5, 4, 3, 2, 1] 40 + config_overrides = { 41 + "ignored": {"importance": "ignore"}, 42 + "low_priority": {"importance": "low"}, 43 + "normal_priority": {"importance": "normal"}, 44 + "high_priority": {"importance": "high"}, 45 + } 46 + 47 + assert _apply_category_caps(selected_ids, categorized_frames, config_overrides) == [ 48 + 3, 49 + 4, 50 + 6, 51 + 7, 52 + 8, 53 + 9, 54 + ] 55 + 56 + 57 + def test_fallback_select_frames_is_deterministic_and_spread(): 58 + categorized_frames = _frames(30) 59 + 60 + result1 = _fallback_select_frames(categorized_frames, max_extractions=5) 61 + result2 = _fallback_select_frames(categorized_frames, max_extractions=5) 62 + 63 + assert result1 == result2 64 + assert len(result1) == 5 65 + assert 1 in result1 66 + assert 30 in result1 67 + 68 + timestamps = {frame["frame_id"]: frame["timestamp"] for frame in categorized_frames} 69 + selected_timestamps = sorted(timestamps[frame_id] for frame_id in result1) 70 + adjacent_gaps = [ 71 + right - left 72 + for left, right in zip(selected_timestamps, selected_timestamps[1:]) 73 + ] 74 + assert min(adjacent_gaps) >= 5 75 + 76 + 77 + def test_fallback_select_frames_returns_all_when_under_max_and_empty(): 78 + categorized_frames = _frames(3) 79 + 80 + assert _fallback_select_frames(categorized_frames, max_extractions=5) == [1, 2, 3] 81 + assert _fallback_select_frames([], max_extractions=5) == [] 82 + 83 + 84 + def test_select_frames_readds_first_frame_when_ignore_capped(monkeypatch): 85 + monkeypatch.setattr( 86 + extract_module, 87 + "_get_category_config", 88 + lambda: {"private": {"importance": "ignore"}}, 89 + ) 90 + categorized_frames = [ 91 + _frame(1, "private"), 92 + _frame(2, "private"), 93 + ] 94 + 95 + result = select_frames_for_extraction( 96 + categorized_frames, max_extractions=5, categories=None 97 + ) 98 + 99 + assert result == [1] 100 + 101 + 102 + def test_select_frames_applies_caps_with_fallback_and_sorts(monkeypatch): 103 + monkeypatch.setattr( 104 + extract_module, 105 + "_get_category_config", 106 + lambda: { 107 + "private": {"importance": "ignore"}, 108 + "low_priority": {"importance": "low"}, 109 + }, 110 + ) 111 + categorized_frames = [ 112 + _frame(1, "private"), 113 + _frame(2, "low_priority"), 114 + _frame(3, "low_priority"), 115 + _frame(4, "low_priority"), 116 + _frame(5, "normal_priority"), 117 + _frame(6, "high_priority"), 118 + _frame(7, "private"), 119 + ] 120 + 121 + result = select_frames_for_extraction( 122 + categorized_frames, max_extractions=10, categories=None 123 + ) 124 + 125 + assert result == [1, 2, 3, 5, 6]

Configure Feed

Configure Feed