Compare commits

...

5 Commits

6 changed files with 458 additions and 342 deletions

View File

@@ -5,89 +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 core.camera.gphoto_camera import GPhotoCamera
# 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()
camera = GPhotoCamera()
self.manager = CameraManager(camera)
# camera = GPhotoCamera()
# self.manager = CameraController(camera)
manager = CameraManager()
manager.detect_gphoto()
manager.detect_opencv()
# self.camera_controller = CameraController()
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.photo_button: QPushButton = view.photo_button
self.photo_button.clicked.connect(self.take_photo)
self.record_button: QPushButton = view.record_button
# self.record_button.clicked.connect(self.fun_test)
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.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)
# self.camera_controller = CameraController()
def start_camera(self):
pass
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.photo_button: QPushButton = view.photo_button
self.photo_button.clicked.connect(self.take_photo)
self.record_button: QPushButton = view.record_button
# self.record_button.clicked.connect(self.fun_test)
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.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 load_colors(self) -> None:
colors = self.db.get_all_colors()
print("Loaded colors:", colors)
self.color_list.set_colors(colors)
def start_camera(self):
pass
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 load_colors(self) -> None:
colors = self.db.get_all_colors()
print("Loaded colors:", colors)
self.color_list.set_colors(colors)
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_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 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}")
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 take_photo(self):
print("Robienie zdjęcia...")
self.split_view.toglle_live_view()
def on_edit_color(self, color_name: str):
print(f"Edycja koloru: {color_name}")
def start_liveview(self):
pass
# self.manager.start_camera()
# self.manager.start_stream()
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):
self.manager.start_camera()
# self.manager.start_stream()
def shutdown(self):
self.manager.stop()
def shutdown(self):
pass
# self.manager.stop()

View File

@@ -2,36 +2,41 @@ from abc import ABC, abstractmethod
class BaseCamera(ABC):
def __init__(self) -> None:
self.error_msg = None
def __init__(self) -> None:
self.error_msg = None
@abstractmethod
def connect(self) -> bool:
raise NotImplementedError
@staticmethod
@abstractmethod
def detect() -> dict:
raise NotImplementedError
@abstractmethod
def disconnect(self) -> None:
raise NotImplementedError
@abstractmethod
def connect(self, index: int | None = None) -> bool:
raise NotImplementedError
@abstractmethod
def get_frame(self):
raise NotImplementedError
@abstractmethod
def disconnect(self) -> None:
raise NotImplementedError
@abstractmethod
def get_config_by_id(self, id: int) -> dict:
raise NotImplementedError
@abstractmethod
def get_frame(self):
raise NotImplementedError
@abstractmethod
def get_config_by_name(self, name: str) -> dict:
raise NotImplementedError
@abstractmethod
def get_config_by_id(self, id: int) -> dict:
raise NotImplementedError
@abstractmethod
def set_config_by_id(self, id: int, value) -> None:
raise NotImplementedError
@abstractmethod
def get_config_by_name(self, name: str) -> dict:
raise NotImplementedError
@abstractmethod
def set_config_by_name(self, name: str, value) -> None:
raise NotImplementedError
@abstractmethod
def set_config_by_id(self, id: int, value) -> None:
raise NotImplementedError
def get_error_msg(self):
return str(self.error_msg)
@abstractmethod
def set_config_by_name(self, name: str, value) -> None:
raise NotImplementedError
def get_error_msg(self):
return str(self.error_msg)

View 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()

View File

@@ -1,88 +1,20 @@
from PySide6.QtCore import QObject, QThread, QTimer, Signal, Slot
from PySide6.QtGui import QImage, QPixmap
import cv2
from .base_camera import BaseCamera
class CameraManager(QThread):
frame_ready = Signal(QPixmap)
photo_ready = Signal(QPixmap)
error_occurred = Signal(str)
from .gphoto_camera import GPhotoCamera
from .opencv_camera import OpenCvCamera
from .camera_controller import CameraController
def __init__(self, camera: BaseCamera, fps: int = 15, parent: QObject | None = None) -> None:
super().__init__(parent)
self.camera = camera
self.fps = fps
self.timer = None
self.is_streaming = False
self.is_connected = False
class CameraManager:
def __init__(self) -> None:
pass
self.timer = QTimer()
self.timer.setInterval(int(1000 / self.fps))
self.timer.timeout.connect(self._update_frame)
self.start()
def detect_gphoto(self):
camera_list = GPhotoCamera.detect()
print(camera_list)
return camera_list
def run(self) -> None:
# self.timer = QTimer()
# self.timer.setInterval(int(1000 / self.fps))
# self.timer.timeout.connect(self._update_frame)
self.exec()
def start_camera(self) -> None:
if 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:
self.is_streaming = False
self.is_connected = False
if self.timer:
self.timer.stop()
self.camera.disconnect()
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()
def stop_stream(self) -> None:
if self.is_streaming:
self.is_streaming = False
if self.timer:
self.timer.stop()
def _update_frame(self) -> None:
if not self.is_streaming or not self.is_connected:
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 stop(self):
self.stop_camera()
self.quit()
self.wait()
def detect_opencv(self):
camera_list = OpenCvCamera.detect()
print(camera_list)
return camera_list

View File

@@ -7,116 +7,143 @@ import numpy as np
from .base_camera import BaseCamera
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
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
}
class GPhotoCamera(BaseCamera):
def __init__(self) -> None:
super().__init__()
self.camera = None
self.configs: List[dict] = []
def __init__(self) -> None:
super().__init__()
self.camera = None
self.configs: List[dict] = []
self.camera_index = 0
def connect(self) -> bool:
self.error_msg = None
try:
self.camera = gp.Camera() # type: ignore
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
@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 {}
camera_list = {}
for i in range(cameras.count()): # type: ignore
name = cameras.get_name(i) # type: ignore
port = cameras.get_value(i) # type: ignore
camera_list[i] = {"name": name, "port": port}
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()
def disconnect(self) -> None:
if self.camera:
self.camera.exit()
self.camera = None
self.configs.clear()
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 get_frame(self):
self.error_msg = None
def disconnect(self) -> None:
if self.camera:
self.camera.exit()
self.camera = None
self.configs.clear()
if self.camera is None:
self.error_msg = "[GPHOTO2] Camera is not initialized."
return (False, None)
def get_frame(self):
self.error_msg = 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)
if self.camera is None:
self.error_msg = "[GPHOTO2] Camera is not initialized."
return (False, None)
return (True, frame)
except Exception as e:
self.error_msg = f"[GPHOTO2] {e}"
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)
def get_config_by_id(self, id: int):
return next(w for w in self.configs if w['id'] == id)
return (True, frame)
except Exception as e:
self.error_msg = f"[GPHOTO2] {e}"
return (False, None)
def get_config_by_name(self, name: str):
return next(w for w in self.configs if w['name'] == name)
def get_config_by_id(self, id: int):
return next(w for w in self.configs if w['id'] == id)
def set_config(self, config, value):
if value not in config['choices']:
return
def get_config_by_name(self, name: str):
return next(w for w in self.configs if w['name'] == name)
config['config'].set_value(value) # type: ignore
if self._save_config(config):
config['value'] = value
def set_config(self, config, value):
if value not in config['choices']:
return
def set_config_by_id(self, id: int, value: str):
config = self.get_config_by_id(id)
config['widget'].set_value(value) # type: ignore
if self._save_config(config):
config['value'] = value
self.set_config(config, value)
def set_config_by_id(self, id: int, value: str):
config = self.get_config_by_id(id)
def set_config_by_name(self, name: str, value: str):
config = self.get_config_by_name(name)
self.set_config(config, value)
self.set_config(config, value)
def set_config_by_name(self, name: str, value: str):
config = self.get_config_by_name(name)
def _save_config(self, config):
if not self.camera:
return False
self.set_config(config, value)
self.camera.set_single.config(config['name'], config['widget'])
return True
def _save_config(self, config):
if not self.camera:
return False
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
}
self.camera.set_single.config(config['name'], config['widget'])
return True
try:
new_config["value"] = config.get_value()
except gp.GPhoto2Error:
pass
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["choices"] = list(config.get_choices())
except gp.GPhoto2Error:
pass
try:
new_config["value"] = config.get_value()
except gp.GPhoto2Error:
pass
return new_config
try:
new_config["choices"] = list(config.get_choices())
except gp.GPhoto2Error:
pass
def read_config(self, config):
self.configs.append(self.parse_config(config))
return new_config
for i in range(config.count_children()):
child = config.get_child(i)
self.read_config(child)
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)

View File

@@ -1,98 +1,128 @@
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},
}
class CvCamera(BaseCamera):
"""Kamera oparta na cv2.VideoCapture"""
def __init__(self) -> None:
super().__init__()
self.cap = None
self.configs: List[dict] = []
self.camera_list = []
self.camera_index = 0
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},
}
@staticmethod
def detect():
camera_list = enumerate_cameras(cv2.CAP_ANY)
result = {}
seen_ports = set()
def __init__(self, device_index: int = 0) -> None:
super().__init__()
self.device_index = device_index
self.cap = None
self.configs: List[dict] = []
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
def connect(self) -> bool:
self.error_msg = None
try:
self.cap = cv2.VideoCapture(self.device_index)
if not self.cap.isOpened():
self.error_msg = f"[CV2] Could not open camera {self.device_index}"
return False
cap = cv2.VideoCapture(camera.index, camera.backend) # próbujemy otworzyć
ret, frame = cap.read()
cap.release()
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
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)
except Exception as e:
self.error_msg = f"[CV2] {e}"
self.cap = None
return False
return result
def disconnect(self) -> None:
if self.cap:
self.cap.release()
self.cap = None
self.configs.clear()
def connect(self, index: int | None = None) -> bool:
self.error_msg = None
try:
if index:
self.camera_index = index
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)
self.cap = cv2.VideoCapture(self.camera_index)
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)
if not self.cap.isOpened():
self.error_msg = f"[CV2] Could not open camera {self.camera_index}"
return False
def get_config_by_id(self, id: int):
return next(w for w in self.configs if w["id"] == id)
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
def get_config_by_name(self, name: str):
return next(w for w in self.configs if w["name"] == name)
except Exception as e:
self.error_msg = f"[CV2] {e}"
self.cap = None
return False
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 disconnect(self) -> None:
if self.cap:
self.cap.release()
self.cap = None
self.configs.clear()
def set_config_by_id(self, id: int, value: float):
config = self.get_config_by_id(id)
self.set_config(config, value)
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)
def set_config_by_name(self, name: str, value: float):
config = self.get_config_by_name(name)
self.set_config(config, value)
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)