149 lines
5.0 KiB
Python
149 lines
5.0 KiB
Python
from PySide6.QtCore import QObject, Signal, QRunnable, QThreadPool
|
|
from PySide6.QtGui import QPixmap
|
|
|
|
from .camera_controller import CameraController
|
|
from .gphoto_camera import GPhotoCamera
|
|
from .opencv_camera import OpenCvCamera
|
|
from .base_camera import BaseCamera
|
|
|
|
|
|
class CameraDetectionWorker(QRunnable):
|
|
"""
|
|
Worker thread for detecting cameras to avoid blocking the GUI.
|
|
"""
|
|
class WorkerSignals(QObject):
|
|
finished = Signal(list)
|
|
error = Signal(str)
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.signals = self.WorkerSignals()
|
|
|
|
def run(self) -> None:
|
|
"""The main work of the worker."""
|
|
detected_cameras = []
|
|
|
|
try:
|
|
gphoto_cameras = GPhotoCamera.detect()
|
|
for index, info in gphoto_cameras.items():
|
|
detected_cameras.append({
|
|
"id": f"gphoto_{index}",
|
|
"name": f"{info['name']} ({info['port']})",
|
|
"type": "gphoto",
|
|
"index": index
|
|
})
|
|
except Exception as e:
|
|
self.signals.error.emit(f"Błąd podczas wykrywania kamer GPhoto: {e}")
|
|
|
|
try:
|
|
opencv_cameras = OpenCvCamera.detect()
|
|
for index, info in opencv_cameras.items():
|
|
detected_cameras.append({
|
|
"id": f"opencv_{index}",
|
|
"name": f"OpenCV: {info['name']}",
|
|
"type": "opencv",
|
|
"index": index
|
|
})
|
|
except Exception as e:
|
|
self.signals.error.emit(f"Błąd podczas wykrywania kamer OpenCV: {e}")
|
|
|
|
self.signals.finished.emit(detected_cameras)
|
|
|
|
|
|
class CameraManager(QObject):
|
|
"""
|
|
Zarządza wszystkimi operacjami związanymi z kamerami,
|
|
stanowiąc fasadę dla reszty aplikacji.
|
|
"""
|
|
frame_ready = Signal(QPixmap)
|
|
error_occurred = Signal(str)
|
|
|
|
detection_started = Signal()
|
|
cameras_detected = Signal(list)
|
|
|
|
camera_started = Signal()
|
|
camera_stopped = Signal()
|
|
|
|
def __init__(self, parent: QObject | None = None) -> None:
|
|
super().__init__(parent)
|
|
self._camera_controller = CameraController()
|
|
self._detected_cameras: list[dict] = []
|
|
self._active_camera: BaseCamera | None = None
|
|
self._active_camera_info: dict | None = None
|
|
self.thread_pool = QThreadPool.globalInstance()
|
|
|
|
self._camera_controller.frame_ready.connect(self.frame_ready)
|
|
self._camera_controller.error_occurred.connect(self.error_occurred)
|
|
|
|
def detect_cameras(self) -> None:
|
|
"""
|
|
Rozpoczyna asynchroniczne wykrywanie kamer w osobnym wątku.
|
|
"""
|
|
self.detection_started.emit()
|
|
worker = CameraDetectionWorker()
|
|
worker.signals.finished.connect(self._on_detection_finished)
|
|
worker.signals.error.connect(self.error_occurred)
|
|
self.thread_pool.start(worker)
|
|
|
|
def _on_detection_finished(self, detected_cameras: list):
|
|
"""
|
|
Slot wywoływany po zakończeniu pracy workera wykrywającego kamery.
|
|
"""
|
|
self._detected_cameras = detected_cameras
|
|
self.cameras_detected.emit(self._detected_cameras)
|
|
|
|
def get_detected_cameras(self) -> list[dict]:
|
|
return self._detected_cameras
|
|
|
|
def start_camera(self, camera_id: str, fps: int = 15) -> None:
|
|
"""Uruchamia wybraną kamerę."""
|
|
if self._active_camera:
|
|
self.stop_camera()
|
|
|
|
camera_info = next((c for c in self._detected_cameras if c['id'] == camera_id), None)
|
|
|
|
if not camera_info:
|
|
self.error_occurred.emit(f"Nie znaleziono kamery o ID: {camera_id}")
|
|
return
|
|
|
|
camera_type = camera_info['type']
|
|
camera_index = camera_info['index']
|
|
|
|
if camera_type == "gphoto":
|
|
self._active_camera = GPhotoCamera()
|
|
elif camera_type == "opencv":
|
|
self._active_camera = OpenCvCamera()
|
|
else:
|
|
self.error_occurred.emit(f"Nieznany typ kamery: {camera_type}")
|
|
return
|
|
|
|
self._active_camera_info = camera_info
|
|
|
|
self._camera_controller.set_camera(self._active_camera, fps)
|
|
self._camera_controller.start_camera()
|
|
|
|
# Trzeba sprawdzić, czy połączenie się udało
|
|
if self._camera_controller.is_connected:
|
|
self._camera_controller.start_stream()
|
|
self.camera_started.emit()
|
|
else:
|
|
# Błąd został już wyemitowany przez CameraController
|
|
self._active_camera = None
|
|
self._active_camera_info = None
|
|
|
|
|
|
def stop_camera(self) -> None:
|
|
"""Zatrzymuje aktywną kamerę."""
|
|
if self._active_camera:
|
|
self._camera_controller.stop_camera()
|
|
self._active_camera = None
|
|
self._active_camera_info = None
|
|
self.camera_stopped.emit()
|
|
|
|
def get_active_camera_info(self) -> dict | None:
|
|
return self._active_camera_info
|
|
|
|
def shutdown(self) -> None:
|
|
"""Zamyka kontroler kamery i jego wątek."""
|
|
self.stop_camera()
|
|
self._camera_controller.stop() |