89 lines
2.6 KiB
Python
89 lines
2.6 KiB
Python
"""Camera backends: RealSense SDK and OpenCV V4L2."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Optional
|
|
|
|
import cv2
|
|
import numpy as np
|
|
|
|
from saqr.utils.logger import get_logger
|
|
|
|
log = get_logger("Inference", "camera")
|
|
|
|
try:
|
|
import pyrealsense2 as rs
|
|
HAS_REALSENSE = True
|
|
except ImportError:
|
|
HAS_REALSENSE = False
|
|
|
|
|
|
class RealSenseCapture:
|
|
"""pyrealsense2 pipeline with an OpenCV-like read() interface."""
|
|
|
|
def __init__(self, width: int = 640, height: int = 480, fps: int = 30,
|
|
serial: Optional[str] = None):
|
|
if not HAS_REALSENSE:
|
|
raise RuntimeError("pyrealsense2 not installed")
|
|
self.pipeline = rs.pipeline()
|
|
cfg = rs.config()
|
|
if serial:
|
|
cfg.enable_device(serial)
|
|
cfg.enable_stream(rs.stream.color, width, height, rs.format.bgr8, fps)
|
|
self.profile = self.pipeline.start(cfg)
|
|
self._open = True
|
|
dev = self.profile.get_device()
|
|
log.info(f"RealSense opened | {dev.get_info(rs.camera_info.name)} "
|
|
f"serial={dev.get_info(rs.camera_info.serial_number)} "
|
|
f"{width}x{height}@{fps}")
|
|
|
|
def isOpened(self) -> bool:
|
|
return self._open
|
|
|
|
def read(self):
|
|
if not self._open:
|
|
return False, None
|
|
try:
|
|
frames = self.pipeline.wait_for_frames(timeout_ms=3000)
|
|
color = frames.get_color_frame()
|
|
if not color:
|
|
return False, None
|
|
return True, np.asanyarray(color.get_data())
|
|
except Exception:
|
|
return False, None
|
|
|
|
def release(self):
|
|
if self._open:
|
|
self.pipeline.stop()
|
|
self._open = False
|
|
|
|
|
|
def open_capture(source: str):
|
|
if source.lower().startswith("realsense"):
|
|
serial = None
|
|
if ":" in source:
|
|
serial = source.split(":", 1)[1]
|
|
return RealSenseCapture(width=640, height=480, fps=30, serial=serial)
|
|
|
|
if str(source).isdigit():
|
|
idx = int(source)
|
|
cap = cv2.VideoCapture(idx)
|
|
if cap.isOpened():
|
|
return cap
|
|
cap = cv2.VideoCapture(idx, cv2.CAP_ANY)
|
|
if cap.isOpened():
|
|
return cap
|
|
cap = cv2.VideoCapture(idx, cv2.CAP_V4L2)
|
|
return cap
|
|
|
|
if source.startswith("/dev/video"):
|
|
cap = cv2.VideoCapture(source, cv2.CAP_V4L2)
|
|
if cap.isOpened():
|
|
cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
|
|
cap.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter_fourcc(*"MJPG"))
|
|
cap.set(cv2.CAP_PROP_FRAME_WIDTH, 640)
|
|
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 480)
|
|
cap.set(cv2.CAP_PROP_FPS, 30)
|
|
return cap
|
|
|
|
return cv2.VideoCapture(source)
|