129 lines
3.4 KiB
Python
129 lines
3.4 KiB
Python
import cv2
|
|
from cv2_enumerate_cameras import enumerate_cameras
|
|
from typing import List
|
|
|
|
from .base_camera import BaseCamera
|
|
|
|
|
|
class OpenCvCamera(BaseCamera):
|
|
"""Kamera oparta na cv2.VideoCapture"""
|
|
|
|
config_map = {
|
|
0: {"name": "frame_width", "cv_prop": cv2.CAP_PROP_FRAME_WIDTH, "default": 640},
|
|
1: {"name": "frame_height", "cv_prop": cv2.CAP_PROP_FRAME_HEIGHT, "default": 480},
|
|
2: {"name": "fps", "cv_prop": cv2.CAP_PROP_FPS, "default": 30},
|
|
3: {"name": "brightness", "cv_prop": cv2.CAP_PROP_BRIGHTNESS, "default": 0.5},
|
|
4: {"name": "contrast", "cv_prop": cv2.CAP_PROP_CONTRAST, "default": 0.5},
|
|
5: {"name": "saturation", "cv_prop": cv2.CAP_PROP_SATURATION, "default": 0.5},
|
|
}
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.cap = None
|
|
self.configs: List[dict] = []
|
|
self.camera_list = []
|
|
self.camera_index = 0
|
|
|
|
@staticmethod
|
|
def detect():
|
|
camera_list = enumerate_cameras(cv2.CAP_ANY)
|
|
result = {}
|
|
seen_ports = set()
|
|
|
|
for camera in camera_list:
|
|
# unikamy duplikatów tego samego /dev/videoX albo tej samej ścieżki na Windows/macOS
|
|
if camera.path in seen_ports:
|
|
continue
|
|
|
|
cap = cv2.VideoCapture(camera.index, camera.backend) # próbujemy otworzyć
|
|
ret, frame = cap.read()
|
|
cap.release()
|
|
|
|
if ret and frame is not None and frame.size > 0:
|
|
result[camera.index] = {
|
|
"name": camera.name,
|
|
"port": camera.path,
|
|
"backend": camera.backend,
|
|
}
|
|
seen_ports.add(camera.path)
|
|
|
|
return result
|
|
|
|
def connect(self, index: int | None = None) -> bool:
|
|
self.error_msg = None
|
|
try:
|
|
if index:
|
|
self.camera_index = index
|
|
|
|
self.cap = cv2.VideoCapture(self.camera_index)
|
|
|
|
if not self.cap.isOpened():
|
|
self.error_msg = f"[CV2] Could not open camera {self.camera_index}"
|
|
return False
|
|
|
|
self.configs.clear()
|
|
for id, conf in self.config_map.items():
|
|
value = self.cap.get(conf["cv_prop"])
|
|
self.configs.append(
|
|
{
|
|
"id": id,
|
|
"name": conf["name"],
|
|
"label": conf["name"].capitalize(),
|
|
"value": value,
|
|
"choices": None, # brak predefiniowanych wyborów
|
|
"cv_prop": conf["cv_prop"],
|
|
}
|
|
)
|
|
return True
|
|
|
|
except Exception as e:
|
|
self.error_msg = f"[CV2] {e}"
|
|
self.cap = None
|
|
return False
|
|
|
|
def disconnect(self) -> None:
|
|
if self.cap:
|
|
self.cap.release()
|
|
self.cap = None
|
|
self.configs.clear()
|
|
|
|
def get_frame(self):
|
|
self.error_msg = None
|
|
if self.cap is None or not self.cap.isOpened():
|
|
self.error_msg = "[CV2] Camera is not initialized."
|
|
return (False, None)
|
|
|
|
try:
|
|
ret, frame = self.cap.read()
|
|
if not ret:
|
|
self.error_msg = "[CV2] Failed to read frame."
|
|
return (False, None)
|
|
return (True, frame)
|
|
except Exception as e:
|
|
self.error_msg = f"[CV2] {e}"
|
|
return (False, None)
|
|
|
|
def get_config_by_id(self, id: int):
|
|
return next(w for w in self.configs if w["id"] == id)
|
|
|
|
def get_config_by_name(self, name: str):
|
|
return next(w for w in self.configs if w["name"] == name)
|
|
|
|
def set_config(self, config, value: float):
|
|
if not self.cap:
|
|
return
|
|
try:
|
|
self.cap.set(config["cv_prop"], value)
|
|
config["value"] = self.cap.get(
|
|
config["cv_prop"]) # sprawdz co ustawiło
|
|
except Exception as e:
|
|
self.error_msg = f"[CV2] {e}"
|
|
|
|
def set_config_by_id(self, id: int, value: float):
|
|
config = self.get_config_by_id(id)
|
|
self.set_config(config, value)
|
|
|
|
def set_config_by_name(self, name: str, value: float):
|
|
config = self.get_config_by_name(name)
|
|
self.set_config(config, value)
|