Spaces:
Sleeping
Sleeping
| """ | |
| camera_service.py | |
| Manages OpenCV camera capture. Start/Stop on demand (not continuous). | |
| """ | |
| import cv2 | |
| import threading | |
| import time | |
| class CameraService: | |
| def __init__(self): | |
| self._cap = None | |
| self._running = False | |
| self._lock = threading.Lock() | |
| self._frame = None | |
| def is_running(self) -> bool: | |
| return self._running | |
| def start(self, source: int = 0): | |
| with self._lock: | |
| if self._running: | |
| return | |
| self._cap = cv2.VideoCapture(source) | |
| if not self._cap.isOpened(): | |
| raise RuntimeError("Cannot open camera") | |
| self._running = True | |
| threading.Thread(target=self._capture_loop, daemon=True).start() | |
| def stop(self): | |
| with self._lock: | |
| self._running = False | |
| if self._cap: | |
| self._cap.release() | |
| self._cap = None | |
| self._frame = None | |
| def _capture_loop(self): | |
| while self._running: | |
| if self._cap and self._cap.isOpened(): | |
| ok, frame = self._cap.read() | |
| if ok: | |
| self._frame = frame | |
| time.sleep(0.033) # ~30 fps | |
| def get_frame(self): | |
| """Return the latest frame as JPEG bytes, or None.""" | |
| if self._frame is None: | |
| return None | |
| ok, buf = cv2.imencode(".jpg", self._frame) | |
| return buf.tobytes() if ok else None | |
| def get_frame_raw(self): | |
| """Return raw numpy frame for processing.""" | |
| return self._frame.copy() if self._frame is not None else None | |
| def generate_mjpeg(self): | |
| """Generator for MJPEG streaming.""" | |
| while self._running: | |
| jpeg = self.get_frame() | |
| if jpeg: | |
| yield ( | |
| b"--frame\r\n" | |
| b"Content-Type: image/jpeg\r\n\r\n" + jpeg + b"\r\n" | |
| ) | |
| time.sleep(0.066) # ~15 fps for streaming | |
| # Singleton | |
| camera = CameraService() | |