solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4import types
5from pathlib import Path
6
7import pytest
8
9from solstone.observe import aruco as aruco_module
10from solstone.observe import describe as describe_module
11from solstone.observe.describe import _winnow_decision
12
13av = pytest.importorskip("av")
14np = pytest.importorskip("numpy")
15
16
17def _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
23def _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
48class _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
58def _arr():
59 return np.zeros((8, 8, 3), dtype=np.uint8)
60
61
62def _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
84def 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
109def 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
121def 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
129def 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
148def 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
168def 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
181def 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
194def 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)