Merge branch 'dev-camera'
This commit is contained in:
@@ -5,69 +5,96 @@ from core.media import MediaRepository
|
||||
from ui.widgets.color_list_widget import ColorListWidget
|
||||
from ui.widgets.thumbnail_list_widget import ThumbnailListWidget
|
||||
from ui.widgets.split_view_widget import SplitView
|
||||
from .camera_controller import CameraController
|
||||
# from .camera_controller import CameraController
|
||||
from core.camera.camera_controller import CameraController
|
||||
from core.camera.camera_manager import CameraManager
|
||||
|
||||
|
||||
from core.camera.gphoto_camera import GPhotoCamera
|
||||
from core.camera.camera_controller import CameraController
|
||||
|
||||
class MainController:
|
||||
def __init__(self, view):
|
||||
self.db = DatabaseManager()
|
||||
self.db.connect()
|
||||
self.media_repo = MediaRepository(self.db)
|
||||
self.media_repo.sync_media()
|
||||
def __init__(self, view):
|
||||
self.db = DatabaseManager()
|
||||
self.db.connect()
|
||||
self.media_repo = MediaRepository(self.db)
|
||||
self.media_repo.sync_media()
|
||||
|
||||
self.camera_controller = CameraController()
|
||||
# camera = GPhotoCamera()
|
||||
# self.manager = CameraController(camera)
|
||||
manager = CameraManager()
|
||||
manager.detect_gphoto()
|
||||
manager.detect_opencv()
|
||||
|
||||
self.view = view
|
||||
self.color_list: ColorListWidget = view.color_list_widget
|
||||
self.thumbnail_list: ThumbnailListWidget = view.thumbnail_widget
|
||||
self.split_view: SplitView = view.preview_widget
|
||||
# self.camera_controller = CameraController()
|
||||
|
||||
self.photo_button: QPushButton = view.photo_button
|
||||
self.photo_button.clicked.connect(self.take_photo)
|
||||
self.view = view
|
||||
self.color_list: ColorListWidget = view.color_list_widget
|
||||
self.thumbnail_list: ThumbnailListWidget = view.thumbnail_widget
|
||||
self.split_view: SplitView = view.preview_widget
|
||||
|
||||
self.color_list.colorSelected.connect(self.on_color_selected)
|
||||
self.color_list.editColor.connect(self.on_edit_color)
|
||||
self.thumbnail_list.selectedThumbnail.connect(self.on_thumbnail_selected)
|
||||
self.photo_button: QPushButton = view.photo_button
|
||||
self.photo_button.clicked.connect(self.take_photo)
|
||||
|
||||
self.camera_controller.errorOccurred.connect(self.split_view.widget_start.set_info_text)
|
||||
self.camera_controller.frameReady.connect(self.split_view.set_live_image)
|
||||
self.split_view.widget_start.camera_start_btn.clicked.connect(self.camera_controller.start)
|
||||
self.record_button: QPushButton = view.record_button
|
||||
# self.record_button.clicked.connect(self.fun_test)
|
||||
|
||||
def start_camera(self):
|
||||
pass
|
||||
self.color_list.colorSelected.connect(self.on_color_selected)
|
||||
self.color_list.editColor.connect(self.on_edit_color)
|
||||
self.thumbnail_list.selectedThumbnail.connect(self.on_thumbnail_selected)
|
||||
|
||||
def load_colors(self) -> None:
|
||||
colors = self.db.get_all_colors()
|
||||
print("Loaded colors:", colors)
|
||||
self.color_list.set_colors(colors)
|
||||
# self.camera_controller.errorOccurred.connect(self.split_view.widget_start.set_info_text)
|
||||
# self.manager.error_occurred.connect(self.split_view.widget_start.set_info_text)
|
||||
# self.camera_controller.frameReady.connect(self.split_view.set_live_image)
|
||||
# self.manager.frame_ready.connect(self.split_view.set_live_image)
|
||||
# self.split_view.widget_start.camera_start_btn.clicked.connect(self.camera_controller.start)
|
||||
self.split_view.widget_start.camera_start_btn.clicked.connect(self.start_liveview)
|
||||
|
||||
|
||||
def on_color_selected(self, color_name: str):
|
||||
print(f"Wybrano kolor: {color_name}")
|
||||
color_id = self.db.get_color_id(color_name)
|
||||
if color_id is not None:
|
||||
media_items = self.db.get_media_for_color(color_id)
|
||||
print(f"Media dla koloru {color_name} (ID: {color_id}):", media_items)
|
||||
def start_camera(self):
|
||||
pass
|
||||
|
||||
self.thumbnail_list.list_widget.clear()
|
||||
for media in media_items:
|
||||
if media['file_type'] == 'photo':
|
||||
file_name = Path(media['media_path']).name
|
||||
self.thumbnail_list.add_thumbnail(media['media_path'], file_name, media['id'])
|
||||
else:
|
||||
print(f"Nie znaleziono koloru o nazwie: {color_name}")
|
||||
def load_colors(self) -> None:
|
||||
colors = self.db.get_all_colors()
|
||||
print("Loaded colors:", colors)
|
||||
self.color_list.set_colors(colors)
|
||||
|
||||
def on_edit_color(self, color_name: str):
|
||||
print(f"Edycja koloru: {color_name}")
|
||||
|
||||
def on_thumbnail_selected(self, media_id: int):
|
||||
media = self.db.get_media(media_id)
|
||||
if media:
|
||||
print(f"Wybrano miniaturę o ID: {media_id}, ścieżka: {media['media_path']}")
|
||||
self.split_view.set_reference_image(media['media_path'])
|
||||
else:
|
||||
print(f"Nie znaleziono mediów o ID: {media_id}")
|
||||
def on_color_selected(self, color_name: str):
|
||||
print(f"Wybrano kolor: {color_name}")
|
||||
color_id = self.db.get_color_id(color_name)
|
||||
if color_id is not None:
|
||||
media_items = self.db.get_media_for_color(color_id)
|
||||
print(f"Media dla koloru {color_name} (ID: {color_id}):", media_items)
|
||||
|
||||
def take_photo(self):
|
||||
print("Robienie zdjęcia...")
|
||||
self.split_view.toglle_live_view()
|
||||
self.thumbnail_list.list_widget.clear()
|
||||
for media in media_items:
|
||||
if media['file_type'] == 'photo':
|
||||
file_name = Path(media['media_path']).name
|
||||
self.thumbnail_list.add_thumbnail(media['media_path'], file_name, media['id'])
|
||||
else:
|
||||
print(f"Nie znaleziono koloru o nazwie: {color_name}")
|
||||
|
||||
def on_edit_color(self, color_name: str):
|
||||
print(f"Edycja koloru: {color_name}")
|
||||
|
||||
def on_thumbnail_selected(self, media_id: int):
|
||||
media = self.db.get_media(media_id)
|
||||
if media:
|
||||
print(f"Wybrano miniaturę o ID: {media_id}, ścieżka: {media['media_path']}")
|
||||
self.split_view.set_reference_image(media['media_path'])
|
||||
else:
|
||||
print(f"Nie znaleziono mediów o ID: {media_id}")
|
||||
|
||||
def take_photo(self):
|
||||
print("Robienie zdjęcia...")
|
||||
self.split_view.toglle_live_view()
|
||||
|
||||
def start_liveview(self):
|
||||
pass
|
||||
# self.manager.start_camera()
|
||||
# self.manager.start_stream()
|
||||
|
||||
def shutdown(self):
|
||||
pass
|
||||
# self.manager.stop()
|
||||
@@ -1,6 +1,16 @@
|
||||
import cv2
|
||||
import numpy as np
|
||||
|
||||
GP_WIDGET_WINDOW = 0
|
||||
GP_WIDGET_SECTION = 0
|
||||
GP_WIDGET_TEXT = 0
|
||||
GP_WIDGET_RANGE = 0
|
||||
GP_WIDGET_TOGGLE = 0
|
||||
GP_WIDGET_RADIO = 0
|
||||
GP_WIDGET_MENU = 0
|
||||
GP_WIDGET_BUTTON = 0
|
||||
GP_WIDGET_DATE = 0
|
||||
|
||||
class GPhoto2Error(Exception):
|
||||
pass
|
||||
|
||||
@@ -19,6 +29,65 @@ class CameraFileMock:
|
||||
return self._data
|
||||
return self._data, len(self._data)
|
||||
|
||||
class CameraListMock:
|
||||
def count(self):
|
||||
return 1
|
||||
|
||||
def get_name(self, idx):
|
||||
return f"mock_name {idx}"
|
||||
|
||||
def get_value(self, idx):
|
||||
return f"mock_value {idx}"
|
||||
|
||||
class MockPortInfo:
|
||||
def __init__(self, address):
|
||||
self.address = address
|
||||
|
||||
class PortInfoList:
|
||||
def __init__(self):
|
||||
self._ports = []
|
||||
|
||||
def load(self):
|
||||
# Dodaj przykładowe porty
|
||||
self._ports = [MockPortInfo("usb:001,002"), MockPortInfo("usb:001,003")]
|
||||
|
||||
def lookup_path(self, port_address):
|
||||
for idx, port in enumerate(self._ports):
|
||||
if port.address == port_address:
|
||||
return idx
|
||||
raise ValueError("Port not found")
|
||||
|
||||
def __getitem__(self, idx):
|
||||
return self._ports[idx]
|
||||
|
||||
class ConfigMock:
|
||||
def get_id(self):
|
||||
return 0
|
||||
def get_name(self):
|
||||
return "name"
|
||||
def get_label(self):
|
||||
return "label"
|
||||
def get_type(self):
|
||||
return 0
|
||||
def get_value(self):
|
||||
return "value"
|
||||
def get_choices(self):
|
||||
return []
|
||||
def count_children(self):
|
||||
return 0
|
||||
def get_child(self):
|
||||
return ConfigMock()
|
||||
|
||||
|
||||
class CameraAbilitiesList:
|
||||
def __init__(self) -> None:
|
||||
self.abilities = []
|
||||
def load(self):
|
||||
return
|
||||
def lookup_model(self, name):
|
||||
return 1
|
||||
def get_abilities(self, abilities_index):
|
||||
return 0
|
||||
|
||||
class Camera:
|
||||
def __init__(self):
|
||||
@@ -62,3 +131,18 @@ class Camera:
|
||||
|
||||
self._frame_counter += 1
|
||||
return CameraFileMock(frame)
|
||||
|
||||
def set_port_info(self, obj):
|
||||
return False
|
||||
|
||||
def get_config(self):
|
||||
return ConfigMock()
|
||||
|
||||
def set_single_config(self, name, widget):
|
||||
return True
|
||||
|
||||
def gp_camera_autodetect():
|
||||
return CameraListMock()
|
||||
|
||||
def check_result(obj):
|
||||
return obj
|
||||
42
core/camera/base_camera.py
Normal file
42
core/camera/base_camera.py
Normal file
@@ -0,0 +1,42 @@
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
|
||||
class BaseCamera(ABC):
|
||||
def __init__(self) -> None:
|
||||
self.error_msg = None
|
||||
|
||||
@staticmethod
|
||||
@abstractmethod
|
||||
def detect() -> dict:
|
||||
raise NotImplementedError
|
||||
|
||||
@abstractmethod
|
||||
def connect(self, index: int | None = None) -> bool:
|
||||
raise NotImplementedError
|
||||
|
||||
@abstractmethod
|
||||
def disconnect(self) -> None:
|
||||
raise NotImplementedError
|
||||
|
||||
@abstractmethod
|
||||
def get_frame(self):
|
||||
raise NotImplementedError
|
||||
|
||||
@abstractmethod
|
||||
def get_config_by_id(self, id: int) -> dict:
|
||||
raise NotImplementedError
|
||||
|
||||
@abstractmethod
|
||||
def get_config_by_name(self, name: str) -> dict:
|
||||
raise NotImplementedError
|
||||
|
||||
@abstractmethod
|
||||
def set_config_by_id(self, id: int, value) -> None:
|
||||
raise NotImplementedError
|
||||
|
||||
@abstractmethod
|
||||
def set_config_by_name(self, name: str, value) -> None:
|
||||
raise NotImplementedError
|
||||
|
||||
def get_error_msg(self):
|
||||
return str(self.error_msg)
|
||||
115
core/camera/camera_controller.py
Normal file
115
core/camera/camera_controller.py
Normal file
@@ -0,0 +1,115 @@
|
||||
from PySide6.QtCore import QObject, QThread, QTimer, Signal, Slot, QMutex, QMutexLocker
|
||||
from PySide6.QtGui import QImage, QPixmap
|
||||
import cv2
|
||||
|
||||
from .base_camera import BaseCamera
|
||||
|
||||
|
||||
class CameraController(QThread):
|
||||
frame_ready = Signal(QPixmap)
|
||||
photo_ready = Signal(QPixmap)
|
||||
error_occurred = Signal(str)
|
||||
_enable_timer = Signal(bool)
|
||||
|
||||
|
||||
def __init__(self, parent: QObject | None = None) -> None:
|
||||
super().__init__(parent)
|
||||
self.camera = None
|
||||
self.timer = None
|
||||
self.fps = 15
|
||||
self.is_streaming = False
|
||||
self.is_connected = False
|
||||
|
||||
self._camera_mutex = QMutex()
|
||||
self.start()
|
||||
|
||||
|
||||
def run(self) -> None:
|
||||
self.timer = QTimer()
|
||||
self.timer.timeout.connect(self._update_frame)
|
||||
self._enable_timer.connect(self._set_timer)
|
||||
self.exec()
|
||||
|
||||
def stop(self):
|
||||
self.stop_camera()
|
||||
self.quit()
|
||||
self.wait()
|
||||
|
||||
def set_camera(self, camera: BaseCamera, fps: int = 15) -> None:
|
||||
with QMutexLocker(self._camera_mutex):
|
||||
self.stop_stream()
|
||||
self.stop_camera()
|
||||
|
||||
self.camera = camera
|
||||
self.fps = fps
|
||||
|
||||
def start_camera(self) -> None:
|
||||
if self.camera is None or self.is_connected:
|
||||
return
|
||||
|
||||
if self.camera.connect():
|
||||
self.is_connected = True
|
||||
else:
|
||||
self.is_connected = False
|
||||
self.error_occurred.emit(self.camera.get_error_msg())
|
||||
|
||||
def stop_camera(self) -> None:
|
||||
if self.is_streaming:
|
||||
self.stop_stream()
|
||||
|
||||
if self.camera is not None:
|
||||
self.camera.disconnect()
|
||||
|
||||
self.is_connected = False
|
||||
|
||||
def start_stream(self):
|
||||
if not self.is_connected:
|
||||
return
|
||||
|
||||
if self.is_streaming:
|
||||
return
|
||||
|
||||
if self.timer:
|
||||
self.is_streaming = True
|
||||
# self.timer.start()
|
||||
self._enable_timer.emit(True)
|
||||
|
||||
def stop_stream(self) -> None:
|
||||
if self.is_streaming:
|
||||
self.is_streaming = False
|
||||
if self.timer:
|
||||
# self.timer.stop()
|
||||
self._enable_timer.emit(False)
|
||||
|
||||
def _update_frame(self) -> None:
|
||||
with QMutexLocker(self._camera_mutex):
|
||||
if self.camera is None or not self.is_connected:
|
||||
return
|
||||
|
||||
if not self.is_streaming:
|
||||
return
|
||||
|
||||
ret, frame = self.camera.get_frame()
|
||||
|
||||
if not ret:
|
||||
self.error_occurred.emit(self.camera.get_error_msg())
|
||||
return
|
||||
|
||||
if frame is not None:
|
||||
rgb_image = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
|
||||
h, w, ch = rgb_image.shape
|
||||
qimg = QImage(rgb_image.data, w, h, ch * w, QImage.Format.Format_RGB888)
|
||||
pixmap = QPixmap.fromImage(qimg)
|
||||
|
||||
self.frame_ready.emit(pixmap)
|
||||
|
||||
def _set_timer(self, enable: bool):
|
||||
if not self.timer:
|
||||
return
|
||||
|
||||
if enable:
|
||||
self.timer.setInterval(int(1000 / self.fps))
|
||||
self.timer.start()
|
||||
else:
|
||||
self.timer.stop()
|
||||
|
||||
20
core/camera/camera_manager.py
Normal file
20
core/camera/camera_manager.py
Normal file
@@ -0,0 +1,20 @@
|
||||
|
||||
|
||||
from .gphoto_camera import GPhotoCamera
|
||||
from .opencv_camera import OpenCvCamera
|
||||
from .camera_controller import CameraController
|
||||
|
||||
|
||||
class CameraManager:
|
||||
def __init__(self) -> None:
|
||||
pass
|
||||
|
||||
def detect_gphoto(self):
|
||||
camera_list = GPhotoCamera.detect()
|
||||
print(camera_list)
|
||||
return camera_list
|
||||
|
||||
def detect_opencv(self):
|
||||
camera_list = OpenCvCamera.detect()
|
||||
print(camera_list)
|
||||
return camera_list
|
||||
173
core/camera/gphoto_camera.py
Normal file
173
core/camera/gphoto_camera.py
Normal file
@@ -0,0 +1,173 @@
|
||||
from typing import Optional, List
|
||||
from dataclasses import dataclass, field
|
||||
import cv2
|
||||
import numpy as np
|
||||
|
||||
from .base_camera import BaseCamera
|
||||
|
||||
try:
|
||||
import gphoto2 as gp # type: ignore
|
||||
except:
|
||||
import controllers.mock_gphoto as gp
|
||||
|
||||
camera_widget_types = {
|
||||
gp.GP_WIDGET_WINDOW: "GP_WIDGET_WINDOW", # type: ignore
|
||||
gp.GP_WIDGET_SECTION: "GP_WIDGET_SECTION", # type: ignore
|
||||
gp.GP_WIDGET_TEXT: "GP_WIDGET_TEXT", # type: ignore
|
||||
gp.GP_WIDGET_RANGE: "GP_WIDGET_RANGE", # type: ignore
|
||||
gp.GP_WIDGET_TOGGLE: "GP_WIDGET_TOGGLE", # type: ignore
|
||||
gp.GP_WIDGET_RADIO: "GP_WIDGET_RADIO", # type: ignore
|
||||
gp.GP_WIDGET_MENU: "GP_WIDGET_MENU", # type: ignore
|
||||
gp.GP_WIDGET_BUTTON: "GP_WIDGET_BUTTON", # type: ignore
|
||||
gp.GP_WIDGET_DATE: "GP_WIDGET_DATE", # type: ignore
|
||||
}
|
||||
|
||||
operations = [
|
||||
("GP_OPERATION_NONE", gp.GP_OPERATION_NONE), # type: ignore
|
||||
("GP_OPERATION_CAPTURE_IMAGE", gp.GP_OPERATION_CAPTURE_IMAGE), # type: ignore
|
||||
("GP_OPERATION_CAPTURE_VIDEO", gp.GP_OPERATION_CAPTURE_VIDEO), # type: ignore
|
||||
("GP_OPERATION_CAPTURE_AUDIO", gp.GP_OPERATION_CAPTURE_AUDIO), # type: ignore
|
||||
("GP_OPERATION_CAPTURE_PREVIEW", gp.GP_OPERATION_CAPTURE_PREVIEW), # type: ignore
|
||||
("GP_OPERATION_CONFIG", gp.GP_OPERATION_CONFIG), # type: ignore
|
||||
("GP_OPERATION_TRIGGER_CAPTURE", gp.GP_OPERATION_TRIGGER_CAPTURE), # type: ignore
|
||||
]
|
||||
|
||||
|
||||
class GPhotoCamera(BaseCamera):
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self.camera = None
|
||||
self.configs: List[dict] = []
|
||||
self.camera_index = 0
|
||||
|
||||
@staticmethod
|
||||
def detect() -> dict:
|
||||
cameras = gp.check_result(gp.gp_camera_autodetect()) # type: ignore
|
||||
# cameras = gp.Camera().autodetect()
|
||||
if not cameras or cameras.count() == 0: # type: ignore
|
||||
return {}
|
||||
|
||||
abilities_list = gp.CameraAbilitiesList() # type: ignore
|
||||
abilities_list.load()
|
||||
camera_list = {}
|
||||
for i in range(cameras.count()): # type: ignore
|
||||
name = cameras.get_name(i) # type: ignore
|
||||
port = cameras.get_value(i) # type: ignore
|
||||
|
||||
abilities_index = abilities_list.lookup_model(name)
|
||||
abilities = abilities_list.get_abilities(abilities_index)
|
||||
abilities_name = []
|
||||
for name, bit in operations:
|
||||
if abilities.operations & bit: # type: ignore
|
||||
abilities_name.append(name)
|
||||
|
||||
camera_list[i] = {"name": name, "port": port, "abilities": abilities_name}
|
||||
return camera_list
|
||||
|
||||
def connect(self, index: int | None = None) -> bool:
|
||||
self.error_msg = None
|
||||
self.camera = gp.Camera() # type: ignore
|
||||
|
||||
try:
|
||||
if index:
|
||||
self.camera_index = index
|
||||
camera_list = GPhotoCamera.detect()
|
||||
port_info_list = gp.PortInfoList()
|
||||
port_info_list.load()
|
||||
|
||||
port_address = camera_list[index]["port"]
|
||||
port_index = port_info_list.lookup_path(port_address)
|
||||
|
||||
self.camera.set_port_info(port_info_list[port_index])
|
||||
|
||||
self.camera.init()
|
||||
config = self.camera.get_config()
|
||||
self.read_config(config)
|
||||
return True
|
||||
except Exception as e:
|
||||
self.error_msg = f"[GPHOTO2] {e}"
|
||||
self.camera = None
|
||||
return False
|
||||
|
||||
def disconnect(self) -> None:
|
||||
if self.camera:
|
||||
self.camera.exit()
|
||||
self.camera = None
|
||||
self.configs.clear()
|
||||
|
||||
def get_frame(self):
|
||||
self.error_msg = None
|
||||
|
||||
if self.camera is None:
|
||||
self.error_msg = "[GPHOTO2] Camera is not initialized."
|
||||
return (False, None)
|
||||
|
||||
try:
|
||||
file = self.camera.capture_preview() # type: ignore
|
||||
data = file.get_data_and_size()
|
||||
frame = np.frombuffer(data, dtype=np.uint8)
|
||||
frame = cv2.imdecode(frame, cv2.IMREAD_COLOR)
|
||||
|
||||
return (True, frame)
|
||||
except Exception as e:
|
||||
self.error_msg = f"[GPHOTO2] {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):
|
||||
if value not in config['choices']:
|
||||
return
|
||||
|
||||
config['widget'].set_value(value) # type: ignore
|
||||
if self._save_config(config):
|
||||
config['value'] = value
|
||||
|
||||
def set_config_by_id(self, id: int, value: str):
|
||||
config = self.get_config_by_id(id)
|
||||
|
||||
self.set_config(config, value)
|
||||
|
||||
def set_config_by_name(self, name: str, value: str):
|
||||
config = self.get_config_by_name(name)
|
||||
|
||||
self.set_config(config, value)
|
||||
|
||||
def _save_config(self, config):
|
||||
if not self.camera:
|
||||
return False
|
||||
|
||||
self.camera.set_single_config(config['name'], config['widget'])
|
||||
return True
|
||||
|
||||
def parse_config(self, config):
|
||||
new_config = {
|
||||
"id": config.get_id(),
|
||||
"name": config.get_name(),
|
||||
"label": config.get_label(),
|
||||
"type": camera_widget_types[config.get_type()],
|
||||
"widget": config
|
||||
}
|
||||
|
||||
try:
|
||||
new_config["value"] = config.get_value()
|
||||
except gp.GPhoto2Error:
|
||||
pass
|
||||
|
||||
try:
|
||||
new_config["choices"] = list(config.get_choices())
|
||||
except gp.GPhoto2Error:
|
||||
pass
|
||||
|
||||
return new_config
|
||||
|
||||
def read_config(self, config):
|
||||
self.configs.append(self.parse_config(config))
|
||||
|
||||
for i in range(config.count_children()):
|
||||
child = config.get_child(i)
|
||||
self.read_config(child)
|
||||
122
core/camera/opencv_camera.py
Normal file
122
core/camera/opencv_camera.py
Normal file
@@ -0,0 +1,122 @@
|
||||
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 = {}
|
||||
|
||||
for camera in camera_list:
|
||||
cap = cv2.VideoCapture(camera.index, camera.backend)
|
||||
# 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,
|
||||
}
|
||||
|
||||
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)
|
||||
Reference in New Issue
Block a user