refactor: update camera classes to improve initialization and connection handling

This commit is contained in:
2025-09-30 21:50:30 +02:00
parent 196eff7fd8
commit 324ab2e016
3 changed files with 229 additions and 190 deletions

View File

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

View File

@@ -7,116 +7,140 @@ import numpy as np
from .base_camera import BaseCamera from .base_camera import BaseCamera
camera_widget_types = { camera_widget_types = {
gp.GP_WIDGET_WINDOW: "GP_WIDGET_WINDOW", # type: ignore gp.GP_WIDGET_WINDOW: "GP_WIDGET_WINDOW", # type: ignore
gp.GP_WIDGET_SECTION: "GP_WIDGET_SECTION", # type: ignore gp.GP_WIDGET_SECTION: "GP_WIDGET_SECTION", # type: ignore
gp.GP_WIDGET_TEXT: "GP_WIDGET_TEXT", # type: ignore gp.GP_WIDGET_TEXT: "GP_WIDGET_TEXT", # type: ignore
gp.GP_WIDGET_RANGE: "GP_WIDGET_RANGE", # type: ignore gp.GP_WIDGET_RANGE: "GP_WIDGET_RANGE", # type: ignore
gp.GP_WIDGET_TOGGLE: "GP_WIDGET_TOGGLE", # type: ignore gp.GP_WIDGET_TOGGLE: "GP_WIDGET_TOGGLE", # type: ignore
gp.GP_WIDGET_RADIO: "GP_WIDGET_RADIO", # type: ignore gp.GP_WIDGET_RADIO: "GP_WIDGET_RADIO", # type: ignore
gp.GP_WIDGET_MENU: "GP_WIDGET_MENU", # type: ignore gp.GP_WIDGET_MENU: "GP_WIDGET_MENU", # type: ignore
gp.GP_WIDGET_BUTTON: "GP_WIDGET_BUTTON", # type: ignore gp.GP_WIDGET_BUTTON: "GP_WIDGET_BUTTON", # type: ignore
gp.GP_WIDGET_DATE: "GP_WIDGET_DATE", # type: ignore gp.GP_WIDGET_DATE: "GP_WIDGET_DATE", # type: ignore
} }
class GPhotoCamera(BaseCamera): class GPhotoCamera(BaseCamera):
def __init__(self) -> None: def __init__(self) -> None:
super().__init__() super().__init__()
self.camera = None self.camera = None
self.configs: List[dict] = [] self.configs: List[dict] = []
self.camera_list = []
def connect(self) -> bool: def detect(self) -> list:
self.error_msg = None self.camera_list.clear()
try: cameras = gp.check_result(gp.gp_camera_autodetect()) # type: ignore
self.camera = gp.Camera() # type: ignore # cameras = gp.Camera().autodetect()
self.camera.init() if not cameras or cameras.count() == 0: # type: ignore
config = self.camera.get_config() return []
self.read_config(config)
return True for i in range(cameras.count()): # type: ignore
except Exception as e: name = cameras.get_name(i) # type: ignore
self.error_msg = f"[GPHOTO2] {e}" port = cameras.get_value(i) # type: ignore
self.camera = None self.camera_list.append({"name": name, "port": port})
return False return self.camera_list
def connect(self, index: int | None = None) -> bool:
self.error_msg = None
self.camera = gp.Camera() # type: ignore
try:
if index:
port_info_list = gp.PortInfoList()
port_info_list.load()
def disconnect(self) -> None: port_address = self.camera_list[index]["port"]
if self.camera: port_index = port_info_list.lookup_path(port_address)
self.camera.exit()
self.camera = None self.camera.set_port_info(port_info_list[port_index])
self.configs.clear()
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): def disconnect(self) -> None:
self.error_msg = None if self.camera:
self.camera.exit()
self.camera = None
self.configs.clear()
if self.camera is None: def get_frame(self):
self.error_msg = "[GPHOTO2] Camera is not initialized." self.error_msg = None
return (False, None)
try: if self.camera is None:
file = self.camera.capture_preview() # type: ignore self.error_msg = "[GPHOTO2] Camera is not initialized."
data = file.get_data_and_size() return (False, None)
frame = np.frombuffer(data, dtype=np.uint8)
frame = cv2.imdecode(frame, cv2.IMREAD_COLOR)
return (True, frame) try:
except Exception as e: file = self.camera.capture_preview() # type: ignore
self.error_msg = f"[GPHOTO2] {e}" data = file.get_data_and_size()
return (False, None) frame = np.frombuffer(data, dtype=np.uint8)
frame = cv2.imdecode(frame, cv2.IMREAD_COLOR)
def get_config_by_id(self, id: int): return (True, frame)
return next(w for w in self.configs if w['id'] == id) except Exception as e:
self.error_msg = f"[GPHOTO2] {e}"
return (False, None)
def get_config_by_name(self, name: str): def get_config_by_id(self, id: int):
return next(w for w in self.configs if w['name'] == name) return next(w for w in self.configs if w['id'] == id)
def set_config(self, config, value): def get_config_by_name(self, name: str):
if value not in config['choices']: return next(w for w in self.configs if w['name'] == name)
return
config['widget'].set_value(value) # type: ignore def set_config(self, config, value):
if self._save_config(config): if value not in config['choices']:
config['value'] = value return
def set_config_by_id(self, id: int, value: str): config['widget'].set_value(value) # type: ignore
config = self.get_config_by_id(id) 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): self.set_config(config, value)
config = self.get_config_by_name(name)
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): self.set_config(config, value)
if not self.camera:
return False
self.camera.set_single.config(config['name'], config['widget']) def _save_config(self, config):
return True if not self.camera:
return False
def parse_config(self, config): self.camera.set_single.config(config['name'], config['widget'])
new_config = { return True
"id": config.get_id(),
"name": config.get_name(),
"label": config.get_label(),
"type": camera_widget_types[config.get_type()],
"widget": config
}
try: def parse_config(self, config):
new_config["value"] = config.get_value() new_config = {
except gp.GPhoto2Error: "id": config.get_id(),
pass "name": config.get_name(),
"label": config.get_label(),
"type": camera_widget_types[config.get_type()],
"widget": config
}
try: try:
new_config["choices"] = list(config.get_choices()) new_config["value"] = config.get_value()
except gp.GPhoto2Error: except gp.GPhoto2Error:
pass pass
return new_config try:
new_config["choices"] = list(config.get_choices())
except gp.GPhoto2Error:
pass
def read_config(self, config): return new_config
self.configs.append(self.parse_config(config))
for i in range(config.count_children()): def read_config(self, config):
child = config.get_child(i) self.configs.append(self.parse_config(config))
self.read_config(child)
for i in range(config.count_children()):
child = config.get_child(i)
self.read_config(child)

View File

@@ -1,98 +1,113 @@
import cv2 import cv2
from cv2_enumerate_cameras import enumerate_cameras
from typing import List from typing import List
from .base_camera import BaseCamera from .base_camera import BaseCamera
class CvCamera(BaseCamera): class CvCamera(BaseCamera):
"""Kamera oparta na cv2.VideoCapture""" """Kamera oparta na cv2.VideoCapture"""
config_map = { config_map = {
0: {"name": "frame_width", "cv_prop": cv2.CAP_PROP_FRAME_WIDTH, "default": 640}, 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}, 1: {"name": "frame_height", "cv_prop": cv2.CAP_PROP_FRAME_HEIGHT, "default": 480},
2: {"name": "fps", "cv_prop": cv2.CAP_PROP_FPS, "default": 30}, 2: {"name": "fps", "cv_prop": cv2.CAP_PROP_FPS, "default": 30},
3: {"name": "brightness", "cv_prop": cv2.CAP_PROP_BRIGHTNESS, "default": 0.5}, 3: {"name": "brightness", "cv_prop": cv2.CAP_PROP_BRIGHTNESS, "default": 0.5},
4: {"name": "contrast", "cv_prop": cv2.CAP_PROP_CONTRAST, "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}, 5: {"name": "saturation", "cv_prop": cv2.CAP_PROP_SATURATION, "default": 0.5},
} }
def __init__(self, device_index: int = 0) -> None: def __init__(self) -> None:
super().__init__() super().__init__()
self.device_index = device_index self.cap = None
self.cap = None self.configs: List[dict] = []
self.configs: List[dict] = [] self.camera_list = []
self.camera_index = 0
def connect(self) -> bool: def detect(self) -> list:
self.error_msg = None self.camera_list.clear()
try: self.camera_list = enumerate_cameras(cv2.CAP_ANY)
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
self.configs.clear() result = []
for id, conf in self.config_map.items(): for camera in self.camera_list:
value = self.cap.get(conf["cv_prop"]) result.append({"name": camera.name, "port": camera.path})
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: return result
self.error_msg = f"[CV2] {e}"
self.cap = None
return False
def disconnect(self) -> None: def connect(self, index: int | None = None) -> bool:
if self.cap: self.error_msg = None
self.cap.release() try:
self.cap = None if index:
self.configs.clear() self.camera_index = self.camera_list[index].index
def get_frame(self): self.cap = cv2.VideoCapture(self.camera_index)
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: if not self.cap.isOpened():
ret, frame = self.cap.read() self.error_msg = f"[CV2] Could not open camera {self.camera_index}"
if not ret: return False
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): self.configs.clear()
return next(w for w in self.configs if w["id"] == id) 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): except Exception as e:
return next(w for w in self.configs if w["name"] == name) self.error_msg = f"[CV2] {e}"
self.cap = None
return False
def set_config(self, config, value: float): def disconnect(self) -> None:
if not self.cap: if self.cap:
return self.cap.release()
try: self.cap = None
self.cap.set(config["cv_prop"], value) self.configs.clear()
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): def get_frame(self):
config = self.get_config_by_id(id) self.error_msg = None
self.set_config(config, value) 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): try:
config = self.get_config_by_name(name) ret, frame = self.cap.read()
self.set_config(config, value) 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)