mikkeli/usb-camera-webui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add face detection to camera capture

Browse Source 
- capture.py: detect faces with haar cascades (cv2.CascadeClassifier),
  draw green rectangles on frames before streaming
- capture/haarcascades/: bundled haar cascade XML
- README.md: document face detection feature and opencv dependency
This commit is contained in:
mikkeli
2026-08-06 10:24:49 +09:00
parent d4eebf79f6
commit 0f423fccd6
4 changed files with 33433 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
camera/capture.py
+103 -16
View File
@@ -1,16 +1,30 @@
"""V4L2 MJPEG capture via the v4l2 Python library (mmap).
"""V4L2 MJPEG capture with face detection via mmap buffers.
Uses memory-mapped buffers for efficient frame capture without subprocess
overhead. Each frame is a complete JPEG image ready for MJPEG streaming.
Opens /dev/video0 and uses memory-mapped V4L2 buffers for efficient frame
capture. Faces are detected on each frame using OpenCV haar cascades
(downscaled for speed) and drawn as rectangles.
"""
import fcntl
import os
import time
import threading
import mmap
import cv2
import numpy as np
import v4l2
# Fourcc for Motion JPEG
MJPG = v4l2.v4l2_fourcc("M", "J", "P", "G")
# Face detection scale — detects at this fraction of original resolution
# to keep CPU usage low. 0.3 ≈ 384x216 for 1280x720 input → ~150 FPS detection.
FACE_DETECT_SCALE = 0.3
FACE_MIN_NEIGHBORS = 4
FACE_MIN_SIZE = 20
# Global camera singleton
_camera: "Capture | None" = None
_camera_lock = threading.Lock()
@@ -22,26 +36,40 @@ def _ioctl(fd, request, arg):
class Capture:
"""Captures MJPEG frames via V4L2 mmap buffers."""
"""Captures MJPEG frames with face detection via V4L2 mmap buffers."""
def __init__(self, device: str = "/dev/video0", width: int = 1280, height: int = 720) -> None:
def __init__(
self,
device: str = "/dev/video0",
width: int = 1280,
height: int = 720,
detect_faces: bool = True,
) -> None:
self._device = device
self._width = width
self._height = height
self._detect_faces = detect_faces
self._fd = -1
self._buffers: list[dict] = []
self._lock = threading.Lock()
self._last_frame: bytes | None = None
self._capture_thread: threading.Thread | None = None
self._running = False
self._face_cascade: cv2.CascadeClassifier | None = None
@classmethod
def get(cls, device: str = "/dev/video0", width: int = 1280, height: int = 720) -> "Capture":
def get(
cls,
device: str = "/dev/video0",
width: int = 1280,
height: int = 720,
detect_faces: bool = True,
) -> "Capture":
"""Return the global singleton, creating it if needed."""
global _camera
with _camera_lock:
if _camera is None:
_camera = cls(device, width, height)
_camera = cls(device, width, height, detect_faces)
return _camera
def start(self) -> None:
@@ -49,6 +77,8 @@ class Capture:
with self._lock:
if self._running:
return
if self._detect_faces:
self._load_face_cascade()
self._open_device()
self._setup_format()
self._allocate_buffers()
@@ -68,6 +98,17 @@ class Capture:
self._capture_thread.join(timeout=3)
self._release()
def _load_face_cascade(self) -> None:
"""Load the haar cascade XML for face detection."""
cascade_path = os.path.join(
os.path.dirname(os.path.abspath(__file__)),
"haarcascades",
"haarcascade_frontalface_default.xml",
)
self._face_cascade = cv2.CascadeClassifier()
self._face_cascade.load(cascade_path)
print("Face cascade loaded.")
def _open_device(self) -> None:
self._fd = os.open(self._device, os.O_RDWR)
cap = v4l2.v4l2_capability()
@@ -79,10 +120,10 @@ class Capture:
fmt.type = v4l2.V4L2_BUF_TYPE_VIDEO_CAPTURE
fmt.fmt.pix.width = self._width
fmt.fmt.pix.height = self._height
fmt.fmt.pix.pixelformat = v4l2.v4l2_fourcc("MJPG")
fmt.fmt.pix.pixelformat = MJPG
fmt.fmt.pix.field = v4l2.V4L2_FIELD_ANY
_ioctl(self._fd, v4l2.VIDIOC_S_FMT, fmt)
print(f"Format set: {fmt.fmt.pix.width}x{fmt.fmt.pix.height} {v4l2.v4l2_fourcc2str(fmt.fmt.pix.pixelformat)}")
print(f"Format: {fmt.fmt.pix.width}x{fmt.fmt.pix.height} {v4l2.v4l2_fourcc2str(fmt.fmt.pix.pixelformat)}")
def _allocate_buffers(self) -> None:
req = v4l2.v4l2_requestbuffers()
@@ -101,7 +142,7 @@ class Capture:
length = buf.length
offset = buf.m.offset
data = os.mmap(0, length, os.PROT_READ, os.MAP_SHARED, self._fd, offset)
data = mmap.mmap(self._fd, length, mmap.MAP_SHARED, mmap.PROT_READ, 0, offset)
self._buffers.append({
"index": i,
"length": length,
@@ -137,7 +178,7 @@ class Capture:
self._fd = -1
def _capture_loop(self) -> None:
"""Background loop: dequeue, copy, requeue buffers."""
"""Background loop: dequeue, detect faces, copy, requeue buffers."""
while self._running:
try:
self._capture_one()
@@ -147,14 +188,21 @@ class Capture:
time.sleep(0.01)
def _capture_one(self) -> None:
"""Dequeue one frame, copy it, and requeue the buffer."""
"""Dequeue one frame, detect faces, copy it, and requeue the buffer."""
buf = v4l2.v4l2_buffer()
buf.type = v4l2.V4L2_BUF_TYPE_VIDEO_CAPTURE
buf.memory = v4l2.V4L2_MEMORY_MMAP
_ioctl(self._fd, v4l2.VIDIOC_DQBUF, buf)
info = self._buffers[buf.index]
frame_data = bytes(info["mmap"][:buf.bytesused])
raw_data = bytes(info["mmap"][:buf.bytesused])
# Process frame with face detection if enabled
if self._detect_faces and self._face_cascade is not None:
frame_data = self._draw_faces(raw_data)
else:
frame_data = raw_data
self._last_frame = frame_data
# Requeue buffer
@@ -163,11 +211,50 @@ class Capture:
buf.type = v4l2.V4L2_BUF_TYPE_VIDEO_CAPTURE
_ioctl(self._fd, v4l2.VIDIOC_QBUF, buf)
def _draw_faces(self, jpeg_bytes: bytes) -> bytes:
"""Decode JPEG, detect faces, draw rectangles, re-encode."""
# Decode the JPEG frame from V4L2
frame = cv2.imdecode(np.frombuffer(jpeg_bytes, dtype=np.uint8), cv2.IMREAD_COLOR)
if frame is None:
return jpeg_bytes
h, w = frame.shape[:2]
# Scale down for fast detection
sw, sh = int(w * FACE_DETECT_SCALE), int(h * FACE_DETECT_SCALE)
small = cv2.resize(frame, (sw, sh), interpolation=cv2.INTER_AREA)
gray = cv2.cvtColor(small, cv2.COLOR_BGR2GRAY)
# Detect faces
faces = self._face_cascade.detectMultiScale(
gray,
scaleFactor=1.1,
minNeighbors=FACE_MIN_NEIGHBORS,
minSize=(int(FACE_MIN_SIZE / FACE_DETECT_SCALE), int(FACE_MIN_SIZE / FACE_DETECT_SCALE)),
)
# Draw rectangles at original scale
for (x, y, fw, fh) in faces:
x_orig = int(x / FACE_DETECT_SCALE)
y_orig = int(y / FACE_DETECT_SCALE)
fw_orig = int(fw / FACE_DETECT_SCALE)
fh_orig = int(fh / FACE_DETECT_SCALE)
cv2.rectangle(frame, (x_orig, y_orig), (x_orig + fw_orig, y_orig + fh_orig), (0, 255, 0), 2)
# Re-encode as JPEG (quality 85 is a good balance)
_, encoded = cv2.imencode(".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, 85])
return bytes(encoded.tobytes())
def frame(self) -> bytes | None:
"""Return the most recently captured frame."""
"""Return the most recently captured frame (with face boxes if enabled)."""
return self._last_frame
def get_camera(device: str = "/dev/video0", width: int = 1280, height: int = 720) -> Capture:
def get_camera(
device: str = "/dev/video0",
width: int = 1280,
height: int = 720,
detect_faces: bool = True,
) -> Capture:
"""Convenience accessor for the global camera singleton."""
return Capture.get(device, width, height)
return Capture.get(device, width, height, detect_faces)