| import threading |
| import time |
| from typing import Optional, Tuple |
| import cv2 |
|
|
| class CameraManager: |
| def __init__(self, device_index: int = 0, width: int = 1280, height: int = 720): |
| self.device_index = device_index |
| self.target_width = width |
| self.target_height = height |
| |
| self.cap: Optional[cv2.VideoCapture] = None |
| self.thread: Optional[threading.Thread] = None |
| self.running = False |
| self.lock = threading.Lock() |
| |
| self.latest_raw_frame = None |
| self.latest_jpeg: Optional[bytes] = None |
| self.actual_width = 0 |
| self.actual_height = 0 |
| self.fps = 0.0 |
| self.client_count = 0 |
| self.started_at: Optional[float] = None |
| |
| self.new_frame_event = threading.Event() |
|
|
| def start(self) -> Tuple[bool, str]: |
| with self.lock: |
| if self.running: |
| return True, "Camera already running" |
|
|
| try: |
| cap = cv2.VideoCapture(self.device_index, cv2.CAP_V4L2) |
| if not cap.isOpened(): |
| cap = cv2.VideoCapture(self.device_index) |
| |
| if not cap.isOpened(): |
| return False, f"Could not open video device /dev/video{self.device_index}" |
|
|
| |
| cap.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter_fourcc(*'MJPG')) |
| cap.set(cv2.CAP_PROP_FRAME_WIDTH, self.target_width) |
| cap.set(cv2.CAP_PROP_FRAME_HEIGHT, self.target_height) |
| cap.set(cv2.CAP_PROP_FPS, 30) |
|
|
| ret, frame = cap.read() |
| if not ret or frame is None: |
| cap.release() |
| return False, "Failed to read initial frame from camera sensor" |
|
|
| h, w = frame.shape[:2] |
| self.actual_width = w |
| self.actual_height = h |
| self.cap = cap |
| self.running = True |
| self.started_at = time.time() |
|
|
| self.thread = threading.Thread(target=self._capture_loop, daemon=True) |
| self.thread.start() |
| return True, f"Camera started successfully ({w}x{h})" |
| except Exception as e: |
| if self.cap: |
| self.cap.release() |
| self.cap = None |
| return False, f"Exception while opening camera: {str(e)}" |
|
|
| def stop(self) -> Tuple[bool, str]: |
| with self.lock: |
| if not self.running: |
| return True, "Camera is already in standby" |
|
|
| self.running = False |
| self.new_frame_event.set() |
|
|
| if self.thread: |
| self.thread.join(timeout=2.0) |
| self.thread = None |
|
|
| with self.lock: |
| if self.cap: |
| self.cap.release() |
| self.cap = None |
| self.latest_jpeg = None |
| self.latest_raw_frame = None |
| self.fps = 0.0 |
| self.started_at = None |
|
|
| return True, "Camera put into standby, device released" |
|
|
| def _capture_loop(self): |
| frame_counter = 0 |
| fps_timer = time.time() |
|
|
| while self.running and self.cap and self.cap.isOpened(): |
| ret, frame = self.cap.read() |
| if not ret or frame is None: |
| time.sleep(0.01) |
| continue |
|
|
| |
| ret_enc, jpeg = cv2.imencode('.jpg', frame, [int(cv2.IMWRITE_JPEG_QUALITY), 85]) |
| if ret_enc: |
| jpeg_bytes = jpeg.tobytes() |
| with self.lock: |
| self.latest_raw_frame = frame |
| self.latest_jpeg = jpeg_bytes |
|
|
| |
| self.new_frame_event.set() |
| self.new_frame_event.clear() |
|
|
| frame_counter += 1 |
| now = time.time() |
| if now - fps_timer >= 1.0: |
| self.fps = round(frame_counter / (now - fps_timer), 1) |
| frame_counter = 0 |
| fps_timer = now |
|
|
| |
| with self.lock: |
| if self.cap: |
| self.cap.release() |
| self.cap = None |
|
|
| def get_latest_jpeg(self) -> Optional[bytes]: |
| with self.lock: |
| return self.latest_jpeg |
|
|
| def get_status(self) -> dict: |
| uptime = int(time.time() - self.started_at) if self.running and self.started_at else 0 |
| return { |
| "running": self.running, |
| "width": self.actual_width, |
| "height": self.actual_height, |
| "fps": self.fps, |
| "uptime_seconds": uptime, |
| "client_count": self.client_count, |
| "device": f"/dev/video{self.device_index}", |
| } |
|
|
| def generate_mjpeg(self): |
| self.client_count += 1 |
| try: |
| while self.running: |
| |
| self.new_frame_event.wait(timeout=0.2) |
| jpeg = self.get_latest_jpeg() |
| if jpeg: |
| yield ( |
| b"--frame\r\n" |
| b"Content-Type: image/jpeg\r\n" |
| b"Content-Length: " + str(len(jpeg)).encode() + b"\r\n\r\n" |
| + jpeg + b"\r\n" |
| ) |
| else: |
| time.sleep(0.05) |
| finally: |
| self.client_count = max(0, self.client_count - 1) |
|
|
