File size: 2,752 Bytes
3972fea | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 | from __future__ import annotations
import sys
import threading
import time
import cv2
import numpy as np
class CameraStream:
"""Threaded camera frame grabber."""
def __init__(self, index: int = 0, width: int = 1280, height: int = 720,
fps: int = 60) -> None:
backend = cv2.CAP_AVFOUNDATION if sys.platform == "darwin" else cv2.CAP_ANY
cap = cv2.VideoCapture(index, backend)
if not cap.isOpened():
cap = cv2.VideoCapture(index)
if not cap.isOpened():
raise RuntimeError(
f"Could not open camera {index}. "
"On macOS grant camera access in System Settings, Privacy and "
"Security, Camera, then restart."
)
cap.set(cv2.CAP_PROP_FRAME_WIDTH, width)
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, height)
cap.set(cv2.CAP_PROP_FPS, fps)
cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
self.cap = cap
ok, frame = cap.read()
if not ok or frame is None:
cap.release()
raise RuntimeError("Camera opened but returns no frames.")
self._frame: np.ndarray = frame
self._seq = 0
self._lock = threading.Lock()
self._stop = threading.Event()
self.failed = False
self._thread = threading.Thread(target=self._loop, daemon=True)
self._thread.start()
@property
def shape(self) -> tuple[int, int]:
"""Frame height and width."""
return self._frame.shape[0], self._frame.shape[1]
@property
def source_fps(self) -> float:
"""Camera reported FPS."""
v = self.cap.get(cv2.CAP_PROP_FPS)
return float(v) if v and v > 0 else 0.0
def _loop(self) -> None:
"""Background capture loop."""
while not self._stop.is_set():
ok, frame = self.cap.read()
if not ok or frame is None:
self.failed = True
break
with self._lock:
self._frame = frame
self._seq += 1
def read(self) -> tuple[np.ndarray, int]:
"""Latest frame and sequence."""
with self._lock:
return self._frame, self._seq
def wait_next(self, last_seq: int, timeout: float = 1.0) -> tuple[np.ndarray, int]:
"""Wait for fresh frame."""
deadline = time.perf_counter() + timeout
while True:
frame, seq = self.read()
if seq != last_seq or self.failed or time.perf_counter() > deadline:
return frame, seq
time.sleep(0.001)
def release(self) -> None:
"""Stop thread and release."""
self._stop.set()
self._thread.join(timeout=1.0)
self.cap.release()
|