Suriyakumarvijayanayagam
2e60fbb57a
Measured rather than guessed, and the first guess was wrong. Wall clock said H.265 decode cost 58 ms a frame; cap.read() blocks until the next frame arrives, so that was the frame interval, not work. As CPU time: decode 3.7 ms, detection 31.0 ms - and detection ran on every frame whether or not anything was in front of the camera, 6,649 of 8,634 frames with faces_seen 0 and active_tracks 0 throughout. detect_threads: OpenCV spreads a small repeated job over eight threads, costing 31.0 ms of CPU for 8.9 ms of wall. One thread costs 15.3 ms for 15.3 ms, against a 66 ms budget at 15 fps. Half the CPU for latency nothing can notice. motion_gate: a 160x90 greyscale absdiff, 0.1 ms against detection's 15. Consulted only while no track is open; forced to look every motion_max_skip frames; compared against the last frame SEARCHED so a slow drift cannot creep under the threshold; and a threshold above this camera's measured noise and far below a person, so anything ambiguous detects. tests/test_motion_gate.py pins each of those rather than the saving, including asserting the longest run of skips rather than the total - counting the total would pass a gate that slept forty frames and then looked forty times. Together 80% -> 16% of a core, detection skipped on 92% of frames. faces_seen is still 0 and the gate is not why: run directly over the same frames the detector finds nothing at threshold 0.50 either. The placement is the limit, as recorded; the CPU was being spent to rediscover that fifteen times a second. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KGcjxF1cNLcuwc3DAPcnfj
98 lines
3.2 KiB
Python
98 lines
3.2 KiB
Python
"""The motion gate must save CPU without ever losing a face.
|
|
|
|
Cheapness is easy; the property that makes it acceptable is that every way it
|
|
could miss somebody is closed. These tests are that argument, written down.
|
|
"""
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from behavision.config import Config
|
|
from behavision.engine import CameraWorker
|
|
|
|
|
|
class _Worker:
|
|
"""CameraWorker's gate, without a camera, a model or a thread.
|
|
|
|
Built with object.__new__ and the two attributes the gate touches - the
|
|
pattern the other engine tests already use, so no test needs a 166 MB
|
|
model file or a live RTSP stream.
|
|
"""
|
|
|
|
def __new__(cls, **app):
|
|
w = object.__new__(CameraWorker)
|
|
w.cfg = Config()
|
|
for k, v in app.items():
|
|
setattr(w.cfg.app, k, v)
|
|
w._motion_prev = None
|
|
w._motion_skipped = 0
|
|
return w
|
|
|
|
|
|
def frame(value: int, size=(90, 160)) -> np.ndarray:
|
|
return np.full((*size, 3), value, dtype=np.uint8)
|
|
|
|
|
|
def test_the_first_frame_is_always_searched():
|
|
"""Nothing to compare against is not evidence that nothing moved."""
|
|
w = _Worker()
|
|
assert w._nothing_moved(frame(40)) is False
|
|
|
|
|
|
def test_a_still_room_is_skipped():
|
|
w = _Worker()
|
|
w._nothing_moved(frame(40)) # prime
|
|
assert w._nothing_moved(frame(40)) is True
|
|
|
|
|
|
def test_movement_is_never_skipped():
|
|
w = _Worker()
|
|
w._nothing_moved(frame(40))
|
|
# a person is an enormous change next to a 1.0 threshold
|
|
assert w._nothing_moved(frame(120)) is False
|
|
|
|
|
|
def test_it_gives_up_and_looks_anyway():
|
|
"""A change too small or too gradual for a thumbnail must still be found.
|
|
|
|
The invariant is about the longest RUN of skips, not the total: what
|
|
matters is the worst case a person could fall into, which is how long the
|
|
camera can go without actually looking. Counting the total instead would
|
|
pass a gate that skipped forty frames and then looked forty times.
|
|
"""
|
|
w = _Worker(motion_max_skip=5)
|
|
w._nothing_moved(frame(40))
|
|
run = longest = 0
|
|
for _ in range(40):
|
|
if w._nothing_moved(frame(40)):
|
|
run += 1
|
|
longest = max(longest, run)
|
|
else:
|
|
run = 0
|
|
assert longest <= 5, f"went {longest} frames without looking, cap is 5"
|
|
assert longest == 5, f"longest run was {longest} - the gate is not saving what it could"
|
|
|
|
|
|
def test_a_slow_drift_cannot_creep_past_the_threshold():
|
|
"""Each frame below the threshold, but the total far above it.
|
|
|
|
Compared against the last frame we SEARCHED rather than the last frame we
|
|
saw, so a gradual change accumulates and eventually trips the gate instead
|
|
of sliding under it one frame at a time. Without that, someone easing into
|
|
view slowly enough is invisible forever.
|
|
"""
|
|
w = _Worker(motion_max_skip=10_000) # the safety net must not rescue this
|
|
w._nothing_moved(frame(40))
|
|
tripped = None
|
|
for i in range(1, 30):
|
|
if w._nothing_moved(frame(40 + i)) is False:
|
|
tripped = i
|
|
break
|
|
assert tripped is not None, "a slow drift was never noticed"
|
|
assert tripped <= 5, f"took {tripped} frames of drift to notice"
|
|
|
|
|
|
@pytest.mark.parametrize("gate", [True, False])
|
|
def test_the_gate_is_switchable(gate):
|
|
w = _Worker(motion_gate=gate)
|
|
assert w.cfg.app.motion_gate is gate
|