temporary commit

core/gphoto_adapter.py

@@ -0,0 +1,84 @@
+import numpy as np
+import cv2
+
+from PySide6.QtCore import QObject, QThread, Signal, QTimer
+from PySide6.QtGui import QImage, QPixmap
+
+from .base import  BaseImageSource, BaseControlSource
+
+import gphoto2 as gp
+
+# try:
+#     import gphoto2 as gp
+# except:
+#     from . import mock_gphoto as gp
+
+class GPhotoImageSource(BaseImageSource):
+
+    def __init__(self, camera: gp.Camera, fps=10, parent=None): # type: ignore
+        super().__init__(parent)
+        self.camera = camera
+        self.fps = fps
+        self.timer = None
+
+    def start(self):
+        self.timer = QTimer()
+        self.timer.timeout.connect(self._grab_frame)
+        self.timer.start(int(1000 / self.fps))
+
+    def stop(self):
+        if self.timer:
+            self.timer.stop()
+
+    def _grab_frame(self):
+        try:
+            file = self.camera.capture_preview()
+            data = file.get_data_and_size()
+            frame = np.frombuffer(data, dtype=np.uint8)
+            frame = cv2.imdecode(frame, cv2.IMREAD_COLOR)
+            if frame is None:
+                return
+
+            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.new_frame.emit(pixmap)
+        except gp.GPhoto2Error as e:
+            self.errorOccurred.emit(f"GPhoto2 error: {e}")
+
+
+class GPhotoControlSource(BaseControlSource):
+
+    def __init__(self, camera: gp.Camera, parent=None): # type: ignore
+        super().__init__(parent)
+        self.camera = camera
+
+    def set_parameter(self, name, value):
+        try:
+            config = self.camera.get_config()
+            child = config.get_child_by_name(name)
+            child.set_value(value)
+            self.camera.set_config(config)
+            self.parameterChanged.emit(name, value)
+        except gp.GPhoto2Error as e:
+            self.errorOccurred.emit(str(e))
+
+    def get_parameter(self, name):
+        try:
+            config = self.camera.get_config()
+            child = config.get_child_by_name(name)
+            return child.get_value()
+        except gp.GPhoto2Error as e:
+            self.errorOccurred.emit(str(e))
+            return None
+
+    def list_parameters(self):
+        params = {}
+        try:
+            config = self.camera.get_config()
+            for child in config.get_children():
+                params[child.get_name()] = child.get_value()
+        except gp.GPhoto2Error as e:
+            self.errorOccurred.emit(str(e))
+        return params