845a85d618
Phase 1 (Foundation): project skeleton, TOML config + Pydantic validation, MQTT bus wrapper, SQLite schema (9 tables), Click CLI, process supervisor. Phase 2 (Camera): RTSP capture via OpenCV, MOG2 motion detection with configurable sensitivity/zones, adaptive FPS recording (2fps idle/30fps motion) via FFmpeg subprocess, HLS live streaming, pre-motion ring buffer. Phase 3 (Web UI): Flask + Bootstrap 5 dark theme, 6 blueprints, Jinja2 templates (dashboard, kiosk 2x2 grid, events, sensors, recordings, settings), PWA with service worker + Web Push, full admin settings UI with config persistence. Remote Access: WireGuard tunnel configs, nginx reverse proxy with HLS caching + rate limiting, bandwidth-optimized remote HLS stream (426x240 @ 500kbps), DO droplet setup script, certbot TLS. 29 tests passing. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
75 lines
1.7 KiB
Python
75 lines
1.7 KiB
Python
"""Tests for the ring buffer."""
|
|
|
|
import numpy as np
|
|
|
|
from vigilar.camera.ring_buffer import RingBuffer
|
|
|
|
|
|
def test_push_and_count():
|
|
buf = RingBuffer(duration_s=1, max_fps=10)
|
|
assert buf.count == 0
|
|
assert buf.capacity == 10
|
|
|
|
frame = np.zeros((480, 640, 3), dtype=np.uint8)
|
|
for _ in range(5):
|
|
buf.push(frame)
|
|
assert buf.count == 5
|
|
|
|
|
|
def test_capacity_limit():
|
|
buf = RingBuffer(duration_s=1, max_fps=5)
|
|
frame = np.zeros((100, 100, 3), dtype=np.uint8)
|
|
|
|
for i in range(10):
|
|
buf.push(frame)
|
|
|
|
assert buf.count == 5 # maxlen enforced
|
|
|
|
|
|
def test_flush_returns_all_and_clears():
|
|
buf = RingBuffer(duration_s=1, max_fps=10)
|
|
frame = np.zeros((100, 100, 3), dtype=np.uint8)
|
|
|
|
for _ in range(7):
|
|
buf.push(frame)
|
|
|
|
frames = buf.flush()
|
|
assert len(frames) == 7
|
|
assert buf.count == 0
|
|
|
|
|
|
def test_flush_preserves_order():
|
|
buf = RingBuffer(duration_s=1, max_fps=10)
|
|
|
|
for i in range(5):
|
|
frame = np.full((10, 10, 3), i, dtype=np.uint8)
|
|
buf.push(frame)
|
|
|
|
frames = buf.flush()
|
|
for i, tsf in enumerate(frames):
|
|
assert tsf.frame[0, 0, 0] == i
|
|
|
|
|
|
def test_peek_latest():
|
|
buf = RingBuffer(duration_s=1, max_fps=10)
|
|
assert buf.peek_latest() is None
|
|
|
|
frame1 = np.full((10, 10, 3), 1, dtype=np.uint8)
|
|
frame2 = np.full((10, 10, 3), 2, dtype=np.uint8)
|
|
buf.push(frame1)
|
|
buf.push(frame2)
|
|
|
|
latest = buf.peek_latest()
|
|
assert latest is not None
|
|
assert latest.frame[0, 0, 0] == 2
|
|
assert buf.count == 2 # peek doesn't remove
|
|
|
|
|
|
def test_clear():
|
|
buf = RingBuffer(duration_s=1, max_fps=10)
|
|
frame = np.zeros((10, 10, 3), dtype=np.uint8)
|
|
buf.push(frame)
|
|
buf.push(frame)
|
|
buf.clear()
|
|
assert buf.count == 0
|