usb-camera-webui/camera/capture.py

"""V4L2 MJPEG capture via the v4l2 Python library (mmap).

Uses memory-mapped buffers for efficient frame capture without subprocess
overhead. Each frame is a complete JPEG image ready for MJPEG streaming.
"""

import fcntl
import os
import time
import threading

import v4l2

# Global camera singleton
_camera: "Capture | None" = None
_camera_lock = threading.Lock()


def _ioctl(fd, request, arg):
    """Wrapper for fcntl.ioctl with v4l2 structs."""
    return fcntl.ioctl(fd, request, arg)


class Capture:
    """Captures MJPEG frames via V4L2 mmap buffers."""

    def __init__(self, device: str = "/dev/video0", width: int = 1280, height: int = 720) -> None:
        self._device = device
        self._width = width
        self._height = height
        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

    @classmethod
    def get(cls, device: str = "/dev/video0", width: int = 1280, height: int = 720) -> "Capture":
        """Return the global singleton, creating it if needed."""
        global _camera
        with _camera_lock:
            if _camera is None:
                _camera = cls(device, width, height)
            return _camera

    def start(self) -> None:
        """Open the device, set up buffers, and begin capturing."""
        with self._lock:
            if self._running:
                return
            self._open_device()
            self._setup_format()
            self._allocate_buffers()
            self._queue_buffers()
            self._start_streaming()
            self._running = True
            self._capture_thread = threading.Thread(
                target=self._capture_loop, daemon=True, name="camera-capture"
            )
            self._capture_thread.start()

    def stop(self) -> None:
        """Stop streaming and release resources."""
        with self._lock:
            self._running = False
            if self._capture_thread:
                self._capture_thread.join(timeout=3)
            self._release()

    def _open_device(self) -> None:
        self._fd = os.open(self._device, os.O_RDWR)
        cap = v4l2.v4l2_capability()
        _ioctl(self._fd, v4l2.VIDIOC_QUERYCAP, cap)
        print(f"Camera: {cap.card.decode('utf-8', errors='replace')[:40]}")

    def _setup_format(self) -> None:
        fmt = v4l2.v4l2_format()
        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.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)}")

    def _allocate_buffers(self) -> None:
        req = v4l2.v4l2_requestbuffers()
        req.count = 4
        req.type = v4l2.V4L2_BUF_TYPE_VIDEO_CAPTURE
        req.memory = v4l2.V4L2_MEMORY_MMAP
        _ioctl(self._fd, v4l2.VIDIOC_REQBUFS, req)

        self._buffers = []
        for i in range(req.count):
            buf = v4l2.v4l2_buffer()
            buf.type = v4l2.V4L2_BUF_TYPE_VIDEO_CAPTURE
            buf.memory = v4l2.V4L2_MEMORY_MMAP
            buf.index = i
            _ioctl(self._fd, v4l2.VIDIOC_QUERYBUF, buf)

            length = buf.length
            offset = buf.m.offset
            data = os.mmap(0, length, os.PROT_READ, os.MAP_SHARED, self._fd, offset)
            self._buffers.append({
                "index": i,
                "length": length,
                "mmap": data,
            })

    def _queue_buffers(self) -> None:
        for info in self._buffers:
            buf = v4l2.v4l2_buffer()
            buf.type = v4l2.V4L2_BUF_TYPE_VIDEO_CAPTURE
            buf.memory = v4l2.V4L2_MEMORY_MMAP
            buf.index = info["index"]
            _ioctl(self._fd, v4l2.VIDIOC_QBUF, buf)

    def _start_streaming(self) -> None:
        typ = v4l2.v4l2_buf_type(v4l2.V4L2_BUF_TYPE_VIDEO_CAPTURE)
        _ioctl(self._fd, v4l2.VIDIOC_STREAMON, typ)

    def _stop_streaming(self) -> None:
        typ = v4l2.v4l2_buf_type(v4l2.V4L2_BUF_TYPE_VIDEO_CAPTURE)
        try:
            _ioctl(self._fd, v4l2.VIDIOC_STREAMOFF, typ)
        except Exception:
            pass

    def _release(self) -> None:
        self._stop_streaming()
        for info in self._buffers:
            info["mmap"].close()
        self._buffers = []
        if self._fd >= 0:
            os.close(self._fd)
            self._fd = -1

    def _capture_loop(self) -> None:
        """Background loop: dequeue, copy, requeue buffers."""
        while self._running:
            try:
                self._capture_one()
            except BlockingIOError:
                time.sleep(0.001)
            except Exception:
                time.sleep(0.01)

    def _capture_one(self) -> None:
        """Dequeue one frame, 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])
        self._last_frame = frame_data

        # Requeue buffer
        buf.index = info["index"]
        buf.memory = v4l2.V4L2_MEMORY_MMAP
        buf.type = v4l2.V4L2_BUF_TYPE_VIDEO_CAPTURE
        _ioctl(self._fd, v4l2.VIDIOC_QBUF, buf)

    def frame(self) -> bytes | None:
        """Return the most recently captured frame."""
        return self._last_frame


def get_camera(device: str = "/dev/video0", width: int = 1280, height: int = 720) -> Capture:
    """Convenience accessor for the global camera singleton."""
    return Capture.get(device, width, height)