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>
86 lines
2.6 KiB
Python
86 lines
2.6 KiB
Python
"""Thread-safe ring buffer for pre-motion frame storage.
|
|
|
|
Keeps a rolling window of frames so that when motion is detected,
|
|
we can include the seconds leading up to the trigger in the recording.
|
|
"""
|
|
|
|
import threading
|
|
import time
|
|
from collections import deque
|
|
from dataclasses import dataclass, field
|
|
|
|
import numpy as np
|
|
|
|
|
|
@dataclass
|
|
class TimestampedFrame:
|
|
"""A frame with its capture timestamp."""
|
|
frame: np.ndarray
|
|
timestamp: float # time.monotonic()
|
|
frame_number: int = 0
|
|
|
|
|
|
class RingBuffer:
|
|
"""Fixed-duration ring buffer for video frames.
|
|
|
|
Stores the most recent `duration_s * max_fps` frames, automatically
|
|
discarding the oldest when full. Thread-safe for single-producer,
|
|
single-consumer use (camera worker writes, recorder reads).
|
|
"""
|
|
|
|
def __init__(self, duration_s: int = 5, max_fps: int = 30):
|
|
self._max_frames = duration_s * max_fps
|
|
self._buffer: deque[TimestampedFrame] = deque(maxlen=self._max_frames)
|
|
self._lock = threading.Lock()
|
|
self._frame_count = 0
|
|
|
|
def push(self, frame: np.ndarray) -> None:
|
|
"""Add a frame to the buffer."""
|
|
with self._lock:
|
|
self._frame_count += 1
|
|
self._buffer.append(TimestampedFrame(
|
|
frame=frame,
|
|
timestamp=time.monotonic(),
|
|
frame_number=self._frame_count,
|
|
))
|
|
|
|
def flush(self) -> list[TimestampedFrame]:
|
|
"""Drain all frames from the buffer and return them in order.
|
|
|
|
Used when motion is detected to get the pre-motion context.
|
|
The buffer is cleared after flush.
|
|
"""
|
|
with self._lock:
|
|
frames = list(self._buffer)
|
|
self._buffer.clear()
|
|
return frames
|
|
|
|
def peek_latest(self) -> TimestampedFrame | None:
|
|
"""Get the most recent frame without removing it."""
|
|
with self._lock:
|
|
return self._buffer[-1] if self._buffer else None
|
|
|
|
@property
|
|
def count(self) -> int:
|
|
"""Current number of frames in the buffer."""
|
|
with self._lock:
|
|
return len(self._buffer)
|
|
|
|
@property
|
|
def capacity(self) -> int:
|
|
"""Maximum number of frames the buffer can hold."""
|
|
return self._max_frames
|
|
|
|
@property
|
|
def duration_s(self) -> float:
|
|
"""Actual duration of content in the buffer, in seconds."""
|
|
with self._lock:
|
|
if len(self._buffer) < 2:
|
|
return 0.0
|
|
return self._buffer[-1].timestamp - self._buffer[0].timestamp
|
|
|
|
def clear(self) -> None:
|
|
"""Discard all frames."""
|
|
with self._lock:
|
|
self._buffer.clear()
|