feat: add motion-stop countdown capture strategy

bookmark/capture_strategies/motion_stop.py

@@ -0,0 +1,35 @@
+import numpy as np
+
+from .base import CaptureStrategy
+
+
+class MotionStopStrategy(CaptureStrategy):
+    """
+    Fires after motion in the frame stops (frame-to-frame diff stays at or
+    below `motion_threshold`) for `stable_frames` consecutive observe()
+    calls.
+    """
+
+    def __init__(self, motion_threshold: float = 10.0, stable_frames: int = 5) -> None:
+        self.motion_threshold = motion_threshold
+        self.stable_frames = stable_frames
+        self._previous_frame: np.ndarray | None = None
+        self._stable_count = 0
+        self._fired = False
+
+    def reset(self) -> None:
+        self._previous_frame = None
+        self._stable_count = 0
+        self._fired = False
+
+    def observe(self, frame: np.ndarray) -> bool:
+        if self._fired:
+            return False
+        if self._previous_frame is not None and self._previous_frame.shape == frame.shape:
+            diff = np.abs(frame.astype(int) - self._previous_frame.astype(int)).mean()
+            self._stable_count = self._stable_count + 1 if diff <= self.motion_threshold else 0
+        self._previous_frame = frame.copy()
+        if self._stable_count >= self.stable_frames:
+            self._fired = True
+            return True
+        return False

tests/test_motion_stop_strategy.py

@@ -0,0 +1,38 @@
+import numpy as np
+
+from bookmark.capture_strategies.motion_stop import MotionStopStrategy
+
+
+def make_frame(value: int) -> np.ndarray:
+    return np.full((10, 10, 3), value, dtype=np.uint8)
+
+
+def test_does_not_fire_while_frames_keep_changing():
+    strategy = MotionStopStrategy(motion_threshold=5.0, stable_frames=3)
+    for value in range(0, 50, 10):
+        assert strategy.observe(make_frame(value)) is False
+
+
+def test_fires_after_stable_frames():
+    strategy = MotionStopStrategy(motion_threshold=5.0, stable_frames=3)
+    strategy.observe(make_frame(100))
+    strategy.observe(make_frame(100))
+    strategy.observe(make_frame(100))
+    assert strategy.observe(make_frame(100)) is True
+
+
+def test_fires_only_once():
+    strategy = MotionStopStrategy(motion_threshold=5.0, stable_frames=2)
+    strategy.observe(make_frame(50))
+    strategy.observe(make_frame(50))
+    assert strategy.observe(make_frame(50)) is True
+    assert strategy.observe(make_frame(50)) is False
+
+
+def test_reset_clears_state():
+    strategy = MotionStopStrategy(motion_threshold=5.0, stable_frames=2)
+    strategy.observe(make_frame(50))
+    strategy.observe(make_frame(50))
+    strategy.observe(make_frame(50))
+    strategy.reset()
+    assert strategy.observe(make_frame(50)) is False