move_detect/edge/adapters/uvc_source.py

"""USB UVC 소스 (실장비). v4l2 MJPEG. cv2.VideoCapture, 2개 USB3 컨트롤러 분산."""

import cv2


class UvcSource:
    def __init__(self, dev, mjpeg: bool = True, *, reopen_after: int = 30, _open=None):
        # _open lets tests inject a fake capture; production uses cv2.VideoCapture.
        self._dev = dev
        self._mjpeg = mjpeg
        self._reopen_after = reopen_after
        self._open = _open or (lambda d: cv2.VideoCapture(d))
        self._fail_streak = 0
        self.cap = self._open_cap()

    def _open_cap(self):
        cap = self._open(self._dev)
        # Fail LOUD: a device that never opens must not become a silently-dead capture that
        # returns (False, None) forever while the daemon looks healthy. build_runtimes catches
        # this and logs+skips the camera.
        if not cap.isOpened():
            raise RuntimeError(f"UVC device {self._dev!r} failed to open")
        if self._mjpeg:
            cap.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter_fourcc(*"MJPG"))
        return cap

    def read(self):
        ok, frame = self.cap.read()
        if ok:
            self._fail_streak = 0
            return ok, frame
        # A USB cam that browns out / is unplugged returns (False, None) indefinitely. After a
        # streak, release and reopen so a recovered device resumes on its own rather than the
        # camera staying dead until someone restarts the service.
        self._fail_streak += 1
        if self._fail_streak >= self._reopen_after:
            self._fail_streak = 0
            try:
                self.cap.release()
            except Exception:
                pass
            try:
                self.cap = self._open_cap()
            except Exception:
                pass  # still dead — the daemon's error/heartbeat path surfaces it
        return False, None