374 lines
12 KiB
Python
374 lines
12 KiB
Python
from dataclasses import dataclass, field, asdict
|
|
from typing import List, Optional, Dict, Any
|
|
|
|
|
|
@dataclass
|
|
class CameraSpec:
|
|
id: str
|
|
index: int
|
|
role: str = "generic" # rgb | nir | re | generic
|
|
connected: bool = False
|
|
enabled: bool = True
|
|
available: bool = False # detectada e pronta para uso
|
|
|
|
width: int = 640
|
|
height: int = 480
|
|
bayer_pattern: str = "GBRG"
|
|
bit_depth: int = 10
|
|
|
|
model: Optional[str] = None
|
|
serial: Optional[str] = None
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
@dataclass
|
|
class PayloadSpec:
|
|
frame_type: str = "RAW_BRUTO" # RAW_BRUTO | RGB | RGBNIR
|
|
dtype: str = "uint8" # uint8 | uint16 | float32
|
|
layout: str = "HW" # HW | CHW
|
|
channels: int = 1
|
|
channel_names: List[str] = field(default_factory=lambda: ["BAYER"])
|
|
width: int = 640
|
|
height: int = 480
|
|
sources: List[str] = field(default_factory=lambda: ["cam0"])
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
class ModuleState:
|
|
def __init__(self):
|
|
self.module_name = "multispectral"
|
|
self.version = "0.2.0"
|
|
self.payload_format_version = 2
|
|
|
|
self.status = "idle"
|
|
self.status_detail = None
|
|
self.last_error = None
|
|
self.last_command = None
|
|
|
|
self.initialized = False
|
|
self.streaming = False
|
|
|
|
# Estratégia do módulo
|
|
self.capture_mode = "AUTO" # AUTO | SINGLE | DUAL
|
|
self.detected_mode = "NONE" # NONE | SINGLE | DUAL
|
|
self.multi_camera_enabled = True
|
|
self.max_cameras = 2
|
|
|
|
# Configuração global base
|
|
self.fps = 10
|
|
self.jpeg_quality = 90
|
|
|
|
# Frame desejado de saída
|
|
self.frame_type = "RAW_BRUTO" # RAW_BRUTO | RGB | RGBNIR
|
|
self.output_dtype = "uint8" # uint8 | uint16 | float32
|
|
|
|
# Lista de câmeras conhecidas pelo módulo
|
|
self.cameras: List[CameraSpec] = [
|
|
CameraSpec(id="cam0", index=0, role="rgb"),
|
|
CameraSpec(id="cam1", index=1, role="nir"),
|
|
]
|
|
|
|
# Resumo dinâmico da topologia
|
|
self.camera_count_detected = 0
|
|
self.camera_count_active = 0
|
|
self.active_camera_ids: List[str] = []
|
|
|
|
# Spec do payload atual
|
|
self.payload = PayloadSpec()
|
|
|
|
# Trigger
|
|
self.trigger_enabled = True
|
|
self.trigger_pin = 18
|
|
self.trigger_active_high = True
|
|
self.trigger_pulse_ms = 5.0
|
|
self.trigger_settle_delay_ms = 0.0
|
|
|
|
# Streaming
|
|
self.stream_host = None
|
|
self.stream_port = None
|
|
self.stream_fps = None
|
|
self.stream_frame_id = 0
|
|
|
|
# Codec
|
|
self.codec_family = "none"
|
|
self.codec_name = "blosc"
|
|
self.codec_params = {
|
|
"cname": "zstd",
|
|
"clevel": 3,
|
|
"shuffle": "BITSHUFFLE",
|
|
}
|
|
|
|
# Controle da câmera
|
|
self.ae_enable = True
|
|
self.awb_enable = True
|
|
self.exposure_time_us = 15000
|
|
self.analogue_gain = 1.0
|
|
self.colour_gains = [1.0, 1.0]
|
|
self.frame_duration_limits = None
|
|
|
|
# Inicializa o payload conforme a configuração padrão
|
|
self.refresh_topology()
|
|
self.update_payload_spec()
|
|
|
|
# =========================================================
|
|
# Topologia / câmeras
|
|
# =========================================================
|
|
|
|
def get_camera(self, cam_id: str) -> Optional[CameraSpec]:
|
|
for cam in self.cameras:
|
|
if cam.id == cam_id:
|
|
return cam
|
|
return None
|
|
|
|
def get_camera_by_index(self, index: int) -> Optional[CameraSpec]:
|
|
for cam in self.cameras:
|
|
if cam.index == index:
|
|
return cam
|
|
return None
|
|
|
|
def set_camera_connected(
|
|
self,
|
|
index: int,
|
|
connected: bool,
|
|
*,
|
|
width: Optional[int] = None,
|
|
height: Optional[int] = None,
|
|
bayer_pattern: Optional[str] = None,
|
|
bit_depth: Optional[int] = None,
|
|
model: Optional[str] = None,
|
|
serial: Optional[str] = None,
|
|
) -> None:
|
|
cam = self.get_camera_by_index(index)
|
|
if cam is None:
|
|
raise ValueError(f"Câmera de índice {index} não existe.")
|
|
|
|
cam.connected = connected
|
|
cam.available = connected and cam.enabled
|
|
|
|
if width is not None:
|
|
cam.width = width
|
|
if height is not None:
|
|
cam.height = height
|
|
if bayer_pattern is not None:
|
|
cam.bayer_pattern = bayer_pattern
|
|
if bit_depth is not None:
|
|
cam.bit_depth = bit_depth
|
|
if model is not None:
|
|
cam.model = model
|
|
if serial is not None:
|
|
cam.serial = serial
|
|
|
|
self.refresh_topology()
|
|
self.update_payload_spec()
|
|
|
|
def set_camera_enabled(self, index: int, enabled: bool) -> None:
|
|
cam = self.get_camera_by_index(index)
|
|
if cam is None:
|
|
raise ValueError(f"Câmera de índice {index} não existe.")
|
|
|
|
cam.enabled = enabled
|
|
cam.available = cam.connected and cam.enabled
|
|
|
|
self.refresh_topology()
|
|
self.update_payload_spec()
|
|
|
|
def refresh_topology(self) -> None:
|
|
detected = [cam for cam in self.cameras if cam.connected]
|
|
active = [cam for cam in self.cameras if cam.connected and cam.enabled]
|
|
|
|
self.camera_count_detected = len(detected)
|
|
self.camera_count_active = len(active)
|
|
self.active_camera_ids = [cam.id for cam in active]
|
|
|
|
if self.camera_count_active >= 2 and self.multi_camera_enabled:
|
|
self.detected_mode = "DUAL"
|
|
elif self.camera_count_active == 1:
|
|
self.detected_mode = "SINGLE"
|
|
else:
|
|
self.detected_mode = "NONE"
|
|
|
|
def resolve_capture_mode(self) -> str:
|
|
if self.capture_mode == "AUTO":
|
|
return self.detected_mode
|
|
|
|
if self.capture_mode == "DUAL":
|
|
if self.camera_count_active >= 2 and self.multi_camera_enabled:
|
|
return "DUAL"
|
|
if self.camera_count_active == 1:
|
|
return "SINGLE"
|
|
return "NONE"
|
|
|
|
if self.capture_mode == "SINGLE":
|
|
if self.camera_count_active >= 1:
|
|
return "SINGLE"
|
|
return "NONE"
|
|
|
|
return "NONE"
|
|
|
|
# =========================================================
|
|
# Payload
|
|
# =========================================================
|
|
|
|
def update_payload_spec(self) -> None:
|
|
resolved_mode = self.resolve_capture_mode()
|
|
active_cams = [cam for cam in self.cameras if cam.connected and cam.enabled]
|
|
|
|
if resolved_mode == "NONE" or not active_cams:
|
|
self.payload = PayloadSpec(
|
|
frame_type=self.frame_type,
|
|
dtype=self.output_dtype,
|
|
layout="HW",
|
|
channels=0,
|
|
channel_names=[],
|
|
width=0,
|
|
height=0,
|
|
sources=[],
|
|
)
|
|
return
|
|
|
|
# Para SINGLE usamos a primeira ativa
|
|
cam0 = active_cams[0]
|
|
|
|
if self.frame_type == "RAW_BRUTO":
|
|
if resolved_mode == "SINGLE":
|
|
self.payload = PayloadSpec(
|
|
frame_type="RAW_BRUTO",
|
|
dtype=self.output_dtype,
|
|
layout="HW",
|
|
channels=1,
|
|
channel_names=["BAYER"],
|
|
width=cam0.width,
|
|
height=cam0.height,
|
|
sources=[cam0.id],
|
|
)
|
|
elif resolved_mode == "DUAL":
|
|
# Aqui estamos dizendo que o payload contém duas origens brutas.
|
|
# O frame_service / stream_sender depois decide como empacotar isso.
|
|
self.payload = PayloadSpec(
|
|
frame_type="RAW_BRUTO",
|
|
dtype=self.output_dtype,
|
|
layout="HW",
|
|
channels=2,
|
|
channel_names=["BAYER_CAM0", "BAYER_CAM1"],
|
|
width=cam0.width,
|
|
height=cam0.height,
|
|
sources=[cam.id for cam in active_cams[:2]],
|
|
)
|
|
|
|
elif self.frame_type == "RGB":
|
|
self.payload = PayloadSpec(
|
|
frame_type="RGB",
|
|
dtype=self.output_dtype,
|
|
layout="CHW",
|
|
channels=3,
|
|
channel_names=["R", "G", "B"],
|
|
width=cam0.width // 2,
|
|
height=cam0.height // 2,
|
|
sources=[cam0.id],
|
|
)
|
|
|
|
elif self.frame_type == "RGBNIR":
|
|
if resolved_mode == "DUAL":
|
|
cam1 = active_cams[1]
|
|
self.payload = PayloadSpec(
|
|
frame_type="RGBNIR",
|
|
dtype=self.output_dtype,
|
|
layout="CHW",
|
|
channels=5,
|
|
channel_names=["R", "G", "B", "NIR", "RE"],
|
|
width=min(cam0.width, cam1.width) // 2,
|
|
height=min(cam0.height, cam1.height) // 2,
|
|
sources=[cam0.id, cam1.id],
|
|
)
|
|
else:
|
|
# Ainda não dá para formar RGBNIR real com uma câmera só.
|
|
self.payload = PayloadSpec(
|
|
frame_type="RGBNIR",
|
|
dtype=self.output_dtype,
|
|
layout="CHW",
|
|
channels=0,
|
|
channel_names=[],
|
|
width=0,
|
|
height=0,
|
|
sources=[],
|
|
)
|
|
|
|
else:
|
|
raise ValueError(f"frame_type inválido: {self.frame_type}")
|
|
|
|
# =========================================================
|
|
# Estado / erro
|
|
# =========================================================
|
|
|
|
def clear_error(self) -> None:
|
|
self.last_error = None
|
|
if self.status == "error":
|
|
self.status = "idle"
|
|
self.status_detail = None
|
|
|
|
def set_error(self, error: str) -> None:
|
|
self.last_error = str(error)
|
|
self.status = "error"
|
|
self.status_detail = str(error)
|
|
|
|
# =========================================================
|
|
# Serialização
|
|
# =========================================================
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"module": self.module_name,
|
|
"version": self.version,
|
|
|
|
"status": self.status,
|
|
"status_detail": self.status_detail,
|
|
"last_error": self.last_error,
|
|
"last_command": self.last_command,
|
|
|
|
"initialized": self.initialized,
|
|
"streaming": self.streaming,
|
|
|
|
"capture_mode": self.capture_mode,
|
|
"detected_mode": self.detected_mode,
|
|
"multi_camera_enabled": self.multi_camera_enabled,
|
|
"max_cameras": self.max_cameras,
|
|
|
|
"camera_count_detected": self.camera_count_detected,
|
|
"camera_count_active": self.camera_count_active,
|
|
"active_camera_ids": self.active_camera_ids,
|
|
"cameras": [cam.to_dict() for cam in self.cameras],
|
|
|
|
"fps": self.fps,
|
|
"jpeg_quality": self.jpeg_quality,
|
|
|
|
"trigger_enabled": self.trigger_enabled,
|
|
"trigger_pin": self.trigger_pin,
|
|
"trigger_active_high": self.trigger_active_high,
|
|
"trigger_pulse_ms": self.trigger_pulse_ms,
|
|
"trigger_settle_delay_ms": self.trigger_settle_delay_ms,
|
|
|
|
"stream_host": self.stream_host,
|
|
"stream_port": self.stream_port,
|
|
"stream_fps": self.stream_fps,
|
|
"stream_frame_id": self.stream_frame_id,
|
|
|
|
"codec_family": self.codec_family,
|
|
"codec_name": self.codec_name,
|
|
"codec_params": self.codec_params,
|
|
|
|
"ae_enable": self.ae_enable,
|
|
"awb_enable": self.awb_enable,
|
|
"exposure_time_us": self.exposure_time_us,
|
|
"analogue_gain": self.analogue_gain,
|
|
"colour_gains": self.colour_gains,
|
|
"frame_duration_limits": self.frame_duration_limits,
|
|
|
|
"payload_format_version": self.payload_format_version,
|
|
"frame_type": self.frame_type,
|
|
"output_dtype": self.output_dtype,
|
|
"payload": self.payload.to_dict(),
|
|
} |