174 lines
4.9 KiB
Python
174 lines
4.9 KiB
Python
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 core.camera.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 operation, bit in operations:
|
|
if abilities.operations & bit: # type: ignore
|
|
abilities_name.append(operation)
|
|
|
|
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)
|