Zendion/agrobot_base
Zendion
/
agrobot_base
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

ajustado capture e hardware sync

This commit is contained in:
Diego Freitas 2026-09-10 15:54:28 -03:00
parent f48741f92d
commit f1e37c1dae
9 changed files with 2617 additions and 2158 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

565
AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/camera_worker/oak_fcc3_core/oak_fcc3_client.py
View File

@ -11,9 +11,6 @@ from .radiometric_controller import RadiometricController
PHYSICAL_CHANNEL_NAMES = ("R", "G", "B", "RE", "NIR")
PHYSICAL_CHANNEL_COUNT = len(PHYSICAL_CHANNEL_NAMES)
OAK_FCC3_CLIENT_VERSION = "production_v1_2026_08_24"
PRODUCT_SCHEMA = "multispec_module_params_v3"
ASSEMBLY_SCHEMA = "multispec_module_params_assembly_v1"
class OakFcc3Client:
@ -44,110 +41,83 @@ class OakFcc3Client:
capture_mode="AUTO",
raw_policy="allow_single",
module_calibration_json=None,
sync_mode="best",
sync_tolerance_ms=25.0,
mx_id=None,
imu_modo="rotation_vector",
imu_freq_hz=200,
evaluate_quality=True,
require_product_contract=False,
hardware_sync_enabled=None,
frame_sync_master=None,
sync_mode=None,
sync_tolerance_ms=None,
buffer_size=None,
**kwargs,
):
# width/height/bayer recebidos do caller ficam apenas como fallback legado.
# Em module_params de produção, hardware real + MP são a autoridade.
self.legacy_width = int(width)
self.legacy_height = int(height)
self.legacy_bayer = str(bayer or "BGGR").upper()
self.fps = float(fps)
self.frame_type = str(frame_type).upper()
self.output_dtype = str(output_dtype).lower()
self.capture_mode = str(capture_mode).upper()
self.raw_policy = str(raw_policy).lower()
self.width = width
self.height = height
self.bayer = bayer
self.fps = fps
self.frame_type = frame_type
self.output_dtype = output_dtype
self.capture_mode = capture_mode
self.raw_policy = raw_policy
self.module_calibration_json = module_calibration_json
self.require_product_contract = bool(require_product_contract)
self.module_params = self._load_module_params(
module_calibration_json,
required=self.require_product_contract,
)
self.product_contract = self._is_product_module_params(self.module_params)
if self.require_product_contract and not self.product_contract:
raise RuntimeError(
"Contrato de produção obrigatório, mas module_params não foi "
"gerado pelo assembler oficial."
)
self.module_params = self._load_module_params(module_calibration_json)
self.fusion_config = self.module_params.get("fusion_config", {}) or {}
self.camera_hardware = self._resolve_camera_hardware()
self.sensor_size_by_role = self._resolve_sensor_size_by_role()
rgb_size = self.sensor_size_by_role.get(
"rgb",
[self.legacy_width, self.legacy_height],
)
self.rgb_native_width = int(rgb_size[0])
self.rgb_native_height = int(rgb_size[1])
# Mantidos por compatibilidade, mas agora significam RGB de referência.
self.width = self.rgb_native_width
self.height = self.rgb_native_height
self.bayer = self._resolve_bayer_pattern()
self.default_target_size = self._resolve_default_target_size()
self.imu_modo = str(imu_modo).strip().lower()
self.imu_freq_hz = int(imu_freq_hz)
# Auditoria radiométrica completa do Raw5.
#
# True mantém o comportamento histórico e é útil para captura científica,
# normalize/auditoria e ferramentas offline.
#
# No runtime em tempo real deve ficar False: evaluate_frame_quality()
# calcula estatísticas/percentis pesados e não faz parte da montagem
# necessária para a inferência.
self.evaluate_quality = bool(evaluate_quality)
self.mx_id = str(mx_id) if mx_id else None
# Validação antecipada do modo produto. O Manager também repete estes
# checks antes de abrir o hardware, de propósito.
self._validate_static_product_contract()
self.svc = OakFcc3Service(
timeout=10,
fps=self.fps,
width=self.width,
height=self.height,
frame_type=self.frame_type,
output_dtype=self.output_dtype,
capture_mode=self.capture_mode,
raw_policy=self.raw_policy,
sync_mode=sync_mode,
sync_tolerance_ms=sync_tolerance_ms,
fps=fps,
width=width,
height=height,
frame_type=frame_type,
output_dtype=output_dtype,
capture_mode=capture_mode,
raw_policy=raw_policy,
mx_id=self.mx_id,
module_calibration_json=module_calibration_json,
module_params=self.module_params,
require_product_contract=self.require_product_contract,
imu_modo=self.imu_modo,
imu_freq_hz=self.imu_freq_hz,
hardware_sync_enabled=hardware_sync_enabled,
frame_sync_master=frame_sync_master,
sync_mode=sync_mode,
sync_tolerance_ms=sync_tolerance_ms,
buffer_size=buffer_size,
**kwargs,
)
self.applied_camera_controls = {}
self.radiometric_controller = None
self.radiometric_controller_enabled = False
self.core = RawProcessorCore(
sensor_width=self.rgb_native_width,
sensor_height=self.rgb_native_height,
bayer_pattern=self.bayer,
sensor_width=width,
sensor_height=height,
bayer_pattern=bayer,
calibration_json_path=module_calibration_json,
)
# Preview é apenas visual, porém também precisa usar o Bayer/raster
# reais do RGB para não mentir sobre a AR0234.
self.preview = RawProcessorPreview(
sensor_width=self.rgb_native_width,
sensor_height=self.rgb_native_height,
bayer_pattern=self.bayer,
sensor_width=width,
sensor_height=height,
bayer_pattern=bayer,
)
self._preview_cache = {
(self.rgb_native_width, self.rgb_native_height, self.bayer): self.preview,
}
def __enter__(self):
self.start()
@ -156,247 +126,19 @@ class OakFcc3Client:
def __exit__(self, exc_type, exc, tb):
self.stop()
def _load_module_params(self, path, required=False):
if not path:
if required:
raise FileNotFoundError(
"module_params obrigatório no contrato de produção."
)
return {}
if not os.path.isfile(path):
if required:
raise FileNotFoundError(
f"module_params não encontrado: {path}"
)
def _load_module_params(self, path):
if not path or not os.path.isfile(path):
return {}
with open(path, "r", encoding="utf-8") as f:
data = json.load(f)
if not isinstance(data, dict):
raise RuntimeError(
f"module_params root deve ser dict/object: {path}"
)
return data
@staticmethod
def _is_product_module_params(module_params):
mp = module_params or {}
assembly = mp.get("assembly_metadata", {}) or {}
return bool(
mp.get("schema") == PRODUCT_SCHEMA
and assembly.get("schema") == ASSEMBLY_SCHEMA
and isinstance(mp.get("camera_hardware"), dict)
and isinstance(mp.get("sensor_size_by_role"), dict)
and isinstance(mp.get("calibration_provenance"), dict)
)
@staticmethod
def _normalize_size(value, label):
if not (isinstance(value, (list, tuple)) and len(value) == 2):
raise RuntimeError(f"{label} deve ser [W,H], recebido={value!r}")
w = int(value[0])
h = int(value[1])
if w <= 0 or h <= 0:
raise RuntimeError(f"{label} inválido: {value!r}")
return [w, h]
def _resolve_camera_hardware(self):
raw = self.module_params.get("camera_hardware", {}) or {}
if not self.product_contract:
return {}
out = {}
expected = {
"rgb": ("CAM_A", ("OV9782", "AR0234")),
"re": ("CAM_B", ("OV9282",)),
"nir": ("CAM_C", ("OV9282",)),
}
for role, (expected_socket, allowed_sensors) in expected.items():
item = raw.get(role)
if not isinstance(item, dict):
raise RuntimeError(f"camera_hardware sem role={role}")
socket = str(item.get("socket") or item.get("socket_name") or "").upper()
sensor = str(item.get("sensor") or item.get("sensor_name") or "").upper()
size = self._normalize_size(item.get("size"), f"camera_hardware.{role}.size")
if socket != expected_socket:
raise RuntimeError(
f"camera_hardware.{role}.socket={socket!r}, esperado={expected_socket!r}"
)
if sensor not in allowed_sensors:
raise RuntimeError(
f"camera_hardware.{role}.sensor={sensor!r}, permitidos={allowed_sensors}"
)
out[role] = {
"role": role,
"socket": socket,
"sensor": sensor,
"size": size,
}
return out
def _resolve_sensor_size_by_role(self):
if not self.product_contract:
return {
"rgb": [self.legacy_width, self.legacy_height],
"re": [self.legacy_width, self.legacy_height],
"nir": [self.legacy_width, self.legacy_height],
}
raw = self.module_params.get("sensor_size_by_role", {}) or {}
out = {}
for role in ("rgb", "re", "nir"):
size = self._normalize_size(
raw.get(role),
f"sensor_size_by_role.{role}",
)
expected = self.camera_hardware[role]["size"]
if size != expected:
raise RuntimeError(
f"sensor_size_by_role.{role}={size} != camera_hardware={expected}"
)
out[role] = size
return out
def _resolve_bayer_pattern(self):
value = self.module_params.get("bayer_pattern") if self.product_contract else None
bayer = str(value or self.legacy_bayer).upper()
if bayer not in ("RGGB", "BGGR", "GRBG", "GBRG"):
raise RuntimeError(f"bayer_pattern inválido: {bayer!r}")
return bayer
def _resolve_default_target_size(self):
value = (self.fusion_config or {}).get("target_size")
if value is None:
return None
return self._normalize_size(value, "fusion_config.target_size")
def _resolve_target_size(self, target_size):
value = target_size if target_size is not None else self.default_target_size
if value is None:
return None
size = self._normalize_size(value, "target_size")
if self.product_contract and self.default_target_size is not None:
if size != self.default_target_size:
raise RuntimeError(
"target_size solicitado diverge do module_params homologado: "
f"requested={size}, calibrated_runtime={self.default_target_size}"
)
return size
def _validate_static_product_contract(self):
if not self.product_contract:
return
if self.frame_type != "RAW_BRUTO":
raise RuntimeError(
"OakFcc3Client de produção aceita somente frame_type='RAW_BRUTO'."
)
if self.raw_policy != "require_triple":
raise RuntimeError(
"OakFcc3Client de produção exige raw_policy='require_triple'."
)
if self.capture_mode not in ("TRIPLE", "AUTO"):
raise RuntimeError(
"OakFcc3Client de produção exige capture_mode TRIPLE/AUTO."
)
expected_mx = (
(self.module_params.get("calibration_provenance", {}) or {})
.get("device_mx_id")
)
if expected_mx and self.mx_id and str(expected_mx) != self.mx_id:
raise RuntimeError(
"MX ID solicitado diverge da calibração homologada: "
f"requested={self.mx_id}, calibrated={expected_mx}"
)
def _sync_contract_from_manager_status(self, status):
if not isinstance(status, dict):
return
if self.product_contract and not bool(status.get("product_contract", False)):
raise RuntimeError(
"Client carregou module_params produto, mas Manager não reconheceu o contrato."
)
sizes = status.get("sensor_size_by_role")
if isinstance(sizes, dict):
normalized = {
role: self._normalize_size(sizes.get(role), f"manager.sensor_size_by_role.{role}")
for role in ("rgb", "re", "nir")
}
if self.product_contract and normalized != self.sensor_size_by_role:
raise RuntimeError(
"Manager e Client discordam sobre sensor_size_by_role: "
f"manager={normalized}, client={self.sensor_size_by_role}"
)
self.sensor_size_by_role = normalized
bayer = status.get("bayer_pattern")
if bayer:
bayer = str(bayer).upper()
if self.product_contract and bayer != self.bayer:
raise RuntimeError(
f"Manager Bayer={bayer} != Client/module_params={self.bayer}"
)
def get_contract(self):
return {
"client_version": OAK_FCC3_CLIENT_VERSION,
"product_contract": bool(self.product_contract),
"require_product_contract": bool(self.require_product_contract),
"sensor_size_by_role": {
role: list(size)
for role, size in self.sensor_size_by_role.items()
},
"camera_hardware": json.loads(json.dumps(self.camera_hardware)),
"bayer_pattern": self.bayer,
"default_target_size": (
None if self.default_target_size is None
else list(self.default_target_size)
),
"evaluate_quality": bool(self.evaluate_quality),
"radiometric_controller_enabled": bool(self.radiometric_controller_enabled),
}
return json.load(f)
def apply_module_camera_settings(self):
camera_settings = self.module_params.get("camera_settings", {}) or {}
if self.product_contract:
missing = [
role
for role in ("rgb", "re", "nir")
if not isinstance(camera_settings.get(role), dict)
]
if missing:
raise RuntimeError(
f"module_params produto sem camera_settings para: {missing}"
)
applied = {}
roles = ("rgb", "re", "nir") if self.product_contract else tuple(camera_settings.keys())
for role in roles:
settings = camera_settings.get(role)
for role, settings in camera_settings.items():
if not isinstance(settings, dict):
continue
@ -415,43 +157,17 @@ class OakFcc3Client:
}
self.applied_camera_controls = applied
if self.product_contract:
failed = {
role: value
for role, value in applied.items()
if not bool((value or {}).get("ok", False))
}
if failed:
raise RuntimeError(
f"Falha reaplicando camera_settings homologado: {failed}"
)
return applied
def enable_radiometric_controller(self):
cfg = self.module_params.get("radiometric_config", {}) or {}
enabled = bool(cfg.get("enabled", False))
# No produto atual este controller de exposição em campo é OFF.
# Não instanciamos um segundo piloto para ficar parado dentro do loop.
if not enabled:
self.radiometric_controller = None
self.radiometric_controller_enabled = False
return None
self.radiometric_controller = RadiometricController(
client=self,
config_json_path=self.module_calibration_json,
)
self.radiometric_controller.sync_from_camera_controls(
self.applied_camera_controls
)
self.radiometric_controller.sync_from_camera_controls(self.applied_camera_controls)
self.radiometric_controller.sync_from_actual_camera_controls()
self.radiometric_controller_enabled = bool(
getattr(self.radiometric_controller, "enabled", True)
)
return self.radiometric_controller
def update_radiometry(self, decoded, meta=None):
@ -491,45 +207,17 @@ class OakFcc3Client:
)
try:
status = self.svc.get_status()
self.mx_id = str(status.get("mx_id") or self.mx_id or "") or None
self._sync_contract_from_manager_status(status)
self.mx_id = self.svc.manager.mx_id
except Exception:
# Se a validação de contrato falhar, fecha hardware antes de propagar.
try:
self.svc.disconnect()
except Exception:
pass
raise
pass
# O Manager de produção já aplicou estes controles via initialControl.
# Reaplicamos após start como confirmação operacional e para manter
# compatibilidade com Managers legados durante a migração.
applied = self.apply_module_camera_settings()
self.enable_radiometric_controller()
if print_debug:
print("[OAK CLIENT] START:", resp)
print("[OAK CLIENT] CONTRACT:", self.get_contract())
print("[OAK CLIENT] APPLIED CAMERA SETTINGS:", applied)
if isinstance(resp, dict):
resp = dict(resp)
resp["client_version"] = OAK_FCC3_CLIENT_VERSION
resp["product_contract"] = bool(self.product_contract)
resp["sensor_size_by_role"] = {
role: list(size)
for role, size in self.sensor_size_by_role.items()
}
resp["bayer_pattern"] = self.bayer
resp["default_target_size"] = (
None if self.default_target_size is None
else list(self.default_target_size)
)
resp["radiometric_controller_enabled"] = bool(
self.radiometric_controller_enabled
)
self.enable_radiometric_controller()
return resp
@ -541,29 +229,7 @@ class OakFcc3Client:
return self.svc.get_device_metrics()
def get_status(self):
status = self.svc.get_status()
if not isinstance(status, dict):
status = {}
else:
status = dict(status)
status.update({
"client_version": OAK_FCC3_CLIENT_VERSION,
"client_product_contract": bool(self.product_contract),
"client_require_product_contract": bool(self.require_product_contract),
"client_sensor_size_by_role": {
role: list(size)
for role, size in self.sensor_size_by_role.items()
},
"client_bayer_pattern": self.bayer,
"client_default_target_size": (
None if self.default_target_size is None
else list(self.default_target_size)
),
"evaluate_quality": bool(self.evaluate_quality),
"radiometric_controller_enabled": bool(self.radiometric_controller_enabled),
})
return status
return self.svc.get_status()
def get_next_raw_frame(self, timeout=1.0):
return self.svc.capture_frame(timeout=timeout)
@ -578,12 +244,6 @@ class OakFcc3Client:
frame_type = str(raw_meta.get("frame_type", self.frame_type)).upper()
meta = dict(raw_meta)
if self.product_contract and frame_type != "RAW_BRUTO":
raise RuntimeError(
"OakFcc3Client produto recebeu frame_type não canônico: "
f"{frame_type!r}. Esperado='RAW_BRUTO'."
)
if frame_type == "RAW_BRUTO":
decoded = self.decode_stream_cameras(raw_frame, raw_meta)
@ -680,25 +340,6 @@ class OakFcc3Client:
def build_infer_tensor(self, frame, meta, channels_expected, target_size=None):
channels_expected = self._validate_physical_channel_count(channels_expected)
target_size = self._resolve_target_size(target_size)
frame_type = str(
(meta or {}).get("frame_type", self.frame_type)
if isinstance(meta, dict)
else self.frame_type
).upper()
# No runtime quente evitamos o quality audit pesado. Decodificamos e
# usamos exatamente o mesmo caminho do CameraMultispectral.
if not self.evaluate_quality and frame_type == "RAW_BRUTO":
decoded = self.core.decode_stream_cameras(frame, meta)
return self.build_infer_tensor_from_decoded(
decoded=decoded,
meta=meta,
channels_expected=channels_expected,
target_size=target_size,
)
return self.core.build_infer_tensor_from_stream(
frame,
meta,
@ -714,49 +355,41 @@ class OakFcc3Client:
target_size=None,
evaluate_quality=None,
):
"""
Monta o Raw5 físico a partir das câmeras já decodificadas.
evaluate_quality:
- None -> usa self.evaluate_quality
- True -> executa evaluate_frame_quality() e atualiza
core.last_frame_quality_result
- False -> não executa a auditoria pesada e limpa
core.last_frame_quality_result
A flag altera somente a auditoria de qualidade. Não altera decode,
radiometria, flat-field, homografia, crop/resize ou patch normalization.
"""
channels_expected = self._validate_physical_channel_count(channels_expected)
target_size = self._resolve_target_size(target_size)
if evaluate_quality is None:
evaluate_quality = self.evaluate_quality
evaluate_quality = bool(evaluate_quality)
# Core novo faz a geometria source->target em uma única etapa.
# NÃO redimensionar novamente depois da fusão.
tensor = self.core.fuse_multispec_cameras(
decoded,
meta,
channels_expected,
target_size=target_size,
)
patch_cfg = getattr(
self.core,
"patch_normalization_config",
{},
) or {}
tensor = self.core.fuse_multispec_cameras(decoded, meta, channels_expected)
tensor = self.core.resize_tensor_chw(tensor, target_size=target_size)
# Mantém paridade com build_infer_tensor_from_stream: se a calibração
# habilitar patch normalization, ela também vale no caminho decoded.
patch_cfg = getattr(self.core, "patch_normalization_config", {}) or {}
if bool(patch_cfg.get("enabled", False)):
if self.product_contract:
raise RuntimeError(
"patch_normalization não é permitido no contrato produto."
)
tensor = self.core.apply_patch_normalization_to_tensor(tensor)
if evaluate_quality:
self.core.last_frame_quality_result = self.core.evaluate_frame_quality(tensor)
else:
# Evita deixar resultado antigo no objeto e evita percentis no hot path.
self.core.last_frame_quality_result = {
"status": "skipped",
"usable_for_training": None,
"reason": "disabled_by_oak_fcc3_client",
"client_version": OAK_FCC3_CLIENT_VERSION,
}
# Evita deixar um resultado antigo parecer referente ao frame atual.
self.core.last_frame_quality_result = None
return np.ascontiguousarray(
tensor.astype(np.float32, copy=False)
)
return tensor
def decode_stream_cameras(self, frame, meta):
if str(meta.get("frame_type", self.frame_type)).upper() == "PREVIEW":
@ -872,14 +505,10 @@ class OakFcc3Client:
arr = arr[:, :, 0]
if bit_depth == 10 and arr.ndim == 2:
role_size = self.sensor_size_by_role.get(
str(role).lower(),
[self.rgb_native_width, self.rgb_native_height],
)
raw16 = self.core.unpack_raw10_packed(
arr,
sensor_width=int(info.get("width", role_size[0])),
sensor_height=int(info.get("height", role_size[1])),
sensor_width=int(info.get("width", self.width)),
sensor_height=int(info.get("height", self.height)),
)
if role == "rgb":
@ -941,7 +570,7 @@ class OakFcc3Client:
or cam_meta.get("bayer")
or stream_meta.get("bayer_pattern")
or bayer_pattern
or self.bayer
or "RGGB"
)
bayer = str(bayer).upper()
@ -973,17 +602,19 @@ class OakFcc3Client:
real_w = int(cam_meta.get("width", sensor_width))
real_h = int(cam_meta.get("height", sensor_height))
cache_key = (real_w, real_h, bayer)
preview = self._preview_cache.get(cache_key)
if preview is None:
preview = RawProcessorPreview(
sensor_width=real_w,
sensor_height=real_h,
bayer_pattern=bayer,
)
self._preview_cache[cache_key] = preview
core = RawProcessorCore(
sensor_width=real_w,
sensor_height=real_h,
bayer_pattern=bayer,
)
raw16 = self.core.unpack_raw10_packed(
preview = RawProcessorPreview(
sensor_width=real_w,
sensor_height=real_h,
bayer_pattern=bayer,
)
raw16 = core.unpack_raw10_packed(
packed,
sensor_width=real_w,
sensor_height=real_h,
@ -1050,11 +681,6 @@ class OakFcc3Client:
def decode_oak_aligned_multispec(self, frame, meta):
if self.product_contract:
raise RuntimeError(
"MULTISPEC alinhado pela OAK é legado e não faz parte do contrato produto."
)
"""
Decodifica frames já alinhados pela OAK.
@ -1103,11 +729,6 @@ class OakFcc3Client:
return decoded
def build_multispec_tensor_from_oak_aligned(self, decoded, meta=None):
if self.product_contract:
raise RuntimeError(
"Tensor MULTISPEC pré-alinhado pela OAK é legado no contrato produto."
)
"""
Monta CHW [R,G,B,RE,NIR] sem reaplicar homografia.
"""

737
AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/camera_worker/oak_fcc3_core/oak_fcc3_manager.py
View File

@ -12,60 +12,27 @@ import numpy as np
from itertools import product
OAK_FCC3_MANAGER_VERSION = "production_v1_2026_08_24"
class OakFcc3Manager:
"""
Hardware manager de produção para OAK-FFC-3 multiespectral.
Manager OAK-FFC-3 com dois fluxos principais:
Contrato oficial:
CAM_A = RGB = OV9782 1280x800 OU AR0234 1920x1200
CAM_B = RE = OV9282 1280x800
CAM_C = NIR = OV9282 1280x800
1) RAW_BRUTO
- Mantém o comportamento antigo.
- CAM_A/CAM_B/CAM_C enviam RAW10 packed direto para o PC.
- O PC faz decode, flat/radiometric, homografia/fusão/crop/resize.
Caminho oficial de produto:
RAW_BRUTO -> RAW10 packed nativo por câmera -> RawProcessorCore.
O Manager não calibra imagem. Ele abre/valida hardware, aplica a política
inicial da câmera, sincroniza RAW e publica metadata fiel por câmera.
PREVIEW/MULTISPEC permanecem apenas para compatibilidade legada.
Um module_params produzido pelo assembler oficial exige RAW_BRUTO.
2) MULTISPEC
- Câmeras sempre em 800p nativo.
- OAK aplica homografia/crop/resize via ImageManip.
- PC recebe frames já alinhados:
CAM_A/rgb -> BGR uint8
CAM_B/re -> GRAY uint8
CAM_C/nir -> GRAY uint8
- O Client deve montar o tensor sem reaplicar homografia.
"""
PRODUCT_SCHEMA = "multispec_module_params_v3"
ASSEMBLY_SCHEMA = "multispec_module_params_assembly_v1"
LEGACY_SENSOR_W = 1280
LEGACY_SENSOR_H = 800
PRODUCT_TOPOLOGY = {
"rgb": {"socket": "CAM_A", "allowed_sensors": ("OV9782", "AR0234")},
"re": {"socket": "CAM_B", "allowed_sensors": ("OV9282",)},
"nir": {"socket": "CAM_C", "allowed_sensors": ("OV9282",)},
}
SENSOR_PROFILES = {
"OV9782": {
"kind": "color",
"resolution_name": "THE_800_P",
"width": 1280,
"height": 800,
},
"AR0234": {
"kind": "color",
"resolution_name": "THE_1200_P",
"width": 1920,
"height": 1200,
},
"OV9282": {
"kind": "mono",
"resolution_name": "THE_800_P",
"width": 1280,
"height": 800,
},
}
SENSOR_W = 1280
SENSOR_H = 800
def __init__(
self,
@ -77,75 +44,41 @@ class OakFcc3Manager:
capture_mode="AUTO",
raw_policy="allow_single",
roles=None,
sync_mode="best",
hardware_sync_enabled=True,
frame_sync_master="CAM_A",
sync_tolerance_ms=12.0,
buffer_size=8,
sync_mode=None,
hardware_sync_enabled=None,
frame_sync_master=None,
sync_tolerance_ms=None,
buffer_size=None,
only_camera=None,
mx_id=None,
module_calibration_json=None,
module_params=None,
require_product_contract=False,
imu_modo="rotation_vector",
imu_freq_hz=200,
):
self.hardware_sync_enabled = bool(hardware_sync_enabled)
self.frame_sync_master = str(frame_sync_master).upper()
self.fps = float(fps)
self.fps = fps
# width/height são apenas saída LEGADA. RAW_BRUTO usa resolução nativa.
# Para compatibilidade, mantemos width/height.
# No RAW_BRUTO isso não muda o sensor, pois usamos 800p fixo.
# No MULTISPEC isso representa a saída final alinhada da OAK.
self.width = int(width)
self.height = int(height)
self.size = (self.width, self.height)
self.sensor_width = self.SENSOR_W
self.sensor_height = self.SENSOR_H
self.frame_type = str(frame_type).upper()
self.output_dtype = output_dtype
self.capture_mode = str(capture_mode).upper()
self.raw_policy = str(raw_policy).lower()
self.capture_mode = capture_mode
self.raw_policy = raw_policy
self.only_camera = only_camera
self.module_calibration_json = module_calibration_json
self.module_params = (
copy.deepcopy(module_params)
if isinstance(module_params, dict)
else self._load_module_params(module_calibration_json)
)
self.fusion_config = (self.module_params or {}).get("fusion_config", {}) or {}
self.require_product_contract = bool(require_product_contract)
self.product_contract = self._is_product_module_params()
self.expected_device_mx_id = self._expected_device_mx_id()
if self.product_contract:
self.roles = {
contract["socket"]: role
for role, contract in self.PRODUCT_TOPOLOGY.items()
}
else:
self.roles = roles or {
"CAM_A": "rgb",
"CAM_B": "re",
"CAM_C": "nir",
}
self.expected_camera_hardware = self._resolve_expected_camera_hardware()
self.sensor_size_by_role = self._resolve_sensor_size_by_role()
rgb_size = self.sensor_size_by_role.get(
"rgb",
[self.LEGACY_SENSOR_W, self.LEGACY_SENSOR_H],
)
self.sensor_width = int(rgb_size[0])
self.sensor_height = int(rgb_size[1])
self.bayer_pattern = str(
(self.module_params or {}).get("bayer_pattern", "BGGR")
).upper()
self.sync_mode = str(sync_mode).lower()
self.sync_tolerance_ms = float(sync_tolerance_ms)
self.buffer_size = int(buffer_size)
self.roles = roles or {
"CAM_A": "rgb",
"CAM_B": "re",
"CAM_C": "nir",
}
self.mx_id = str(mx_id) if mx_id else None
self.dev_info = None
@ -157,12 +90,14 @@ class OakFcc3Manager:
self.imu_modo = self._validate_imu_modo(imu_modo)
self.imu_freq_hz = int(imu_freq_hz)
if self.imu_freq_hz <= 0:
raise ValueError(
f"imu_freq_hz deve ser maior que zero: {self.imu_freq_hz}"
)
self.imu_sensor_type = None
self.has_imu_pipeline = False
self.tem_imu = False
self.q_imu = None
@ -170,18 +105,32 @@ class OakFcc3Manager:
self.running = False
self.frame_id = 0
# Serializa start/stop e leituras nativas de telemetria.
# Evita getChipTemperature/getUsbSpeed concorrendo com device.close().
self._device_lock = threading.RLock()
self.control_queues = {}
self._last_raw_dims = {}
self.aligned_geometry = None
# Startup Profile entra ANTES do pipeline existir.
self.camera_controls = {
cam_id: self._default_controls_for_role(role)
for cam_id, role in self.roles.items()
}
self._hydrate_camera_controls_from_module_params()
self._validate_static_product_contract()
self._last_raw_dims = {}
self.module_calibration_json = module_calibration_json
self.module_params = module_params if isinstance(module_params, dict) else self._load_module_params(module_calibration_json)
self.fusion_config = (self.module_params or {}).get("fusion_config", {}) or {}
self.aligned_geometry = None
sync_cfg = (self.module_params.get("capture_synchronization", {}) if isinstance(self.module_params, dict) else {})
if hardware_sync_enabled is None: hardware_sync_enabled = sync_cfg.get("hardware_sync_enabled", False)
if frame_sync_master is None: frame_sync_master = sync_cfg.get("frame_sync_master", "CAM_A")
if sync_mode is None: sync_mode = sync_cfg.get("software_sync_mode", "best")
if sync_tolerance_ms is None: sync_tolerance_ms = sync_cfg.get("sync_tolerance_ms", 12.0)
if buffer_size is None: buffer_size = sync_cfg.get("buffer_size", 8)
self.hardware_sync_enabled = bool(hardware_sync_enabled)
self.frame_sync_master = str(frame_sync_master)
self.sync_mode = str(sync_mode)
self.sync_tolerance_ms = float(sync_tolerance_ms)
self.buffer_size = int(buffer_size)
self.async_capture_enabled = True
self.async_capture_mode = "latest" # latest | queue
@ -228,327 +177,11 @@ class OakFcc3Manager:
# ============================================================
def _load_module_params(self, path):
if not path:
if not path or not os.path.isfile(path):
return {}
if not os.path.isfile(path):
raise FileNotFoundError(
f"module_params não encontrado: {path}"
)
with open(path, "r", encoding="utf-8") as f:
data = json.load(f)
if not isinstance(data, dict):
raise RuntimeError(
f"module_params root deve ser dict/object: {path}"
)
return data
def _is_product_module_params(self):
mp = self.module_params or {}
assembly = mp.get("assembly_metadata", {}) or {}
return bool(
mp.get("schema") == self.PRODUCT_SCHEMA
and assembly.get("schema") == self.ASSEMBLY_SCHEMA
and isinstance(mp.get("camera_hardware"), dict)
and isinstance(mp.get("sensor_size_by_role"), dict)
and isinstance(mp.get("calibration_provenance"), dict)
)
def _expected_device_mx_id(self):
prov = (self.module_params or {}).get("calibration_provenance", {}) or {}
value = prov.get("device_mx_id")
return str(value) if value else None
def _normalize_size(self, value, *, label):
if not (isinstance(value, (list, tuple)) and len(value) == 2):
raise RuntimeError(f"{label} deve ser [W,H], recebido={value!r}")
w = int(value[0])
h = int(value[1])
if w <= 0 or h <= 0:
raise RuntimeError(f"{label} inválido: {value!r}")
return [w, h]
def _resolve_expected_camera_hardware(self):
if not self.product_contract:
return {}
raw = (self.module_params or {}).get("camera_hardware", {}) or {}
out = {}
for role, contract in self.PRODUCT_TOPOLOGY.items():
item = raw.get(role)
if not isinstance(item, dict):
raise RuntimeError(
f"module_params.camera_hardware sem role={role}"
)
socket = str(item.get("socket") or "").upper()
sensor = str(item.get("sensor") or "").upper()
size = self._normalize_size(
item.get("size"),
label=f"camera_hardware.{role}.size",
)
if socket != contract["socket"]:
raise RuntimeError(
f"camera_hardware.{role}.socket={socket!r}, "
f"esperado={contract['socket']!r}"
)
if sensor not in contract["allowed_sensors"]:
raise RuntimeError(
f"camera_hardware.{role}.sensor={sensor!r}, "
f"permitidos={contract['allowed_sensors']}"
)
profile = self.SENSOR_PROFILES.get(sensor)
if profile is None:
raise RuntimeError(
f"Sensor sem profile de runtime: {sensor}"
)
native = [int(profile["width"]), int(profile["height"])]
if size != native:
raise RuntimeError(
f"camera_hardware.{role}.size={size} "
f"não corresponde ao nativo de {sensor}: {native}"
)
out[role] = {
"role": role,
"socket": socket,
"sensor": sensor,
"size": size,
"kind": profile["kind"],
"resolution_name": profile["resolution_name"],
}
return out
def _resolve_sensor_size_by_role(self):
if self.product_contract:
raw = (self.module_params or {}).get("sensor_size_by_role", {}) or {}
out = {}
for role in ("rgb", "re", "nir"):
size = self._normalize_size(
raw.get(role),
label=f"sensor_size_by_role.{role}",
)
expected = self.expected_camera_hardware[role]["size"]
if size != expected:
raise RuntimeError(
f"sensor_size_by_role.{role}={size} != "
f"camera_hardware={expected}"
)
out[role] = size
return out
return {
"rgb": [self.LEGACY_SENSOR_W, self.LEGACY_SENSOR_H],
"re": [self.LEGACY_SENSOR_W, self.LEGACY_SENSOR_H],
"nir": [self.LEGACY_SENSOR_W, self.LEGACY_SENSOR_H],
}
def _validate_static_product_contract(self):
if self.require_product_contract and not self.product_contract:
raise RuntimeError(
"Contrato de produção obrigatório, mas o module_params "
"não foi gerado pelo assembler oficial."
)
if not self.product_contract:
return
if self.frame_type != "RAW_BRUTO":
raise RuntimeError(
"module_params de produção aceita somente frame_type=RAW_BRUTO. "
f"Recebido={self.frame_type!r}"
)
if self.capture_mode not in ("TRIPLE", "AUTO"):
raise RuntimeError(
"Produto multiespectral exige capture_mode TRIPLE "
f"(ou AUTO com require_triple). Recebido={self.capture_mode!r}"
)
if self.raw_policy != "require_triple":
raise RuntimeError(
"module_params de produção exige raw_policy='require_triple'. "
f"Recebido={self.raw_policy!r}"
)
if self.only_camera is not None:
raise RuntimeError(
"only_camera não é permitido no contrato de produção."
)
if self.bayer_pattern not in ("RGGB", "BGGR", "GRBG", "GBRG"):
raise RuntimeError(
f"bayer_pattern inválido no module_params: {self.bayer_pattern!r}"
)
if (
self.mx_id is not None
and self.expected_device_mx_id is not None
and self.mx_id != self.expected_device_mx_id
):
raise RuntimeError(
"MX ID solicitado diverge da calibração homologada: "
f"requested={self.mx_id}, calibrated={self.expected_device_mx_id}"
)
def _role_to_cam_id(self, role):
role = str(role).lower()
for cam_id, mapped_role in self.roles.items():
if str(mapped_role).lower() == role:
return cam_id
return None
def _hydrate_camera_controls_from_module_params(self):
settings = (self.module_params or {}).get("camera_settings", {}) or {}
if not isinstance(settings, dict):
if self.product_contract:
raise RuntimeError(
"module_params de produção sem camera_settings."
)
return
for role, cfg in settings.items():
if not isinstance(cfg, dict):
continue
cam_id = self._role_to_cam_id(role)
if cam_id is None:
continue
state = self.camera_controls.setdefault(
cam_id,
self._default_controls_for_role(role),
)
for key in (
"ae_enable",
"awb_enable",
"exposure_time_us",
"analogue_gain",
"colour_gains",
):
if key in cfg:
state[key] = copy.deepcopy(cfg[key])
if self.product_contract:
for role in ("rgb", "re", "nir"):
cam_id = self._role_to_cam_id(role)
if cam_id is None or cam_id not in self.camera_controls:
raise RuntimeError(
f"camera_settings não resolveu role={role}"
)
def _sensor_profile(self, sensor_name, role=None):
sensor = str(sensor_name or "").upper()
profile = self.SENSOR_PROFILES.get(sensor)
if profile is None:
raise RuntimeError(
f"Sensor não suportado pelo runtime: {sensor!r}"
)
if role is not None:
role = str(role).lower()
expected_kind = "color" if role == "rgb" else "mono"
if profile["kind"] != expected_kind:
raise RuntimeError(
f"Sensor {sensor} é {profile['kind']}, "
f"mas role={role} exige {expected_kind}."
)
return profile
def _feature_rows(self, features):
rows = []
for f in features:
socket = f.socket.name
sensor = str(f.sensorName or "").upper()
rows.append({
"socket": socket,
"sensor": sensor,
"width": int(getattr(f, "width", 0) or 0),
"height": int(getattr(f, "height", 0) or 0),
"role": self.roles.get(socket, "unknown"),
})
return rows
def _validate_connected_hardware(self, features):
rows = self._feature_rows(features)
by_socket = {row["socket"]: row for row in rows}
if not self.product_contract:
return rows
if (
self.expected_device_mx_id is not None
and self.mx_id is not None
and self.mx_id != self.expected_device_mx_id
):
raise RuntimeError(
"OAK conectada não corresponde ao módulo calibrado: "
f"connected={self.mx_id}, calibrated={self.expected_device_mx_id}"
)
errors = []
for role, expected in self.expected_camera_hardware.items():
row = by_socket.get(expected["socket"])
if row is None:
errors.append(f"{role}: {expected['socket']} ausente")
continue
if row["sensor"] != expected["sensor"]:
errors.append(
f"{role}: {expected['socket']} sensor={row['sensor']}, "
f"esperado={expected['sensor']}"
)
if row["width"] > 0 and row["height"] > 0:
advertised = [row["width"], row["height"]]
if advertised != expected["size"]:
errors.append(
f"{role}: {expected['socket']}/{row['sensor']} anunciou "
f"{advertised}, esperado={expected['size']}"
)
if errors:
raise RuntimeError(
"Hardware OAK não corresponde ao module_params homologado:\n - "
+ "\n - ".join(errors)
)
return rows
return json.load(f)
def _default_controls_for_role(self, role: str):
role = str(role).lower()
@ -579,31 +212,12 @@ class OakFcc3Manager:
with dai.Device(dev_info) as dev:
result = []
features = dev.getConnectedCameraFeatures()
for row in self._feature_rows(features):
role = str(row.get("role", "unknown")).lower()
expected = self.expected_camera_hardware.get(role)
for f in dev.getConnectedCameraFeatures():
result.append({
**row,
"expected": copy.deepcopy(expected),
"matches_product_contract": (
None
if not self.product_contract
else bool(
expected is not None
and row["socket"] == expected["socket"]
and row["sensor"] == expected["sensor"]
and (
row["width"] <= 0
or row["height"] <= 0
or [row["width"], row["height"]] == expected["size"]
)
)
),
"socket": f.socket.name,
"sensor": f.sensorName,
"role": self.roles.get(f.socket.name, "unknown"),
})
return result
def _device_id_from_info(self, dev_info):
@ -677,48 +291,35 @@ class OakFcc3Manager:
def _create_camera_node_classic(self, socket, sensor_name: str, role: str):
"""
RAW/PREVIEW sensor-aware.
Fluxo clássico.
Produto:
OV9782 -> ColorCamera THE_800_P 1280x800
AR0234 -> ColorCamera THE_1200_P 1920x1200
OV9282 -> MonoCamera THE_800_P 1280x800
RGB/OV9782:
ColorCamera raw para RAW_BRUTO.
No produto não existe fallback silencioso de resolução.
MONO/OV9282:
MonoCamera raw quando disponível.
"""
sensor = str(sensor_name or "").upper()
role = str(role or "").lower()
sensor_name_u = str(sensor_name or "").upper()
role_u = str(role or "").lower()
if self.product_contract:
profile = self._sensor_profile(sensor, role=role)
else:
profile = self.SENSOR_PROFILES.get(sensor)
is_rgb = (
role_u == "rgb"
or "OV9782" in sensor_name_u
or socket == dai.CameraBoardSocket.CAM_A
)
if profile is None:
profile = {
"kind": "color" if role == "rgb" else "mono",
"resolution_name": "THE_800_P",
"width": self.LEGACY_SENSOR_W,
"height": self.LEGACY_SENSOR_H,
}
if profile["kind"] == "color":
if is_rgb:
cam = self.pipeline.createColorCamera()
cam.setBoardSocket(socket)
resolution = getattr(
dai.ColorCameraProperties.SensorResolution,
profile["resolution_name"],
None,
)
try:
cam.setResolution(dai.ColorCameraProperties.SensorResolution.THE_800_P)
except Exception:
try:
cam.setResolution(dai.ColorCameraProperties.SensorResolution.THE_1080_P)
except Exception:
pass
if resolution is None:
raise RuntimeError(
f"DepthAI não expõe ColorCamera "
f"{profile['resolution_name']} para {sensor}."
)
cam.setResolution(resolution)
cam.setInterleaved(False)
cam.setColorOrder(dai.ColorCameraProperties.ColorOrder.RGB)
cam.setFps(float(self.fps))
@ -730,42 +331,31 @@ class OakFcc3Manager:
except Exception:
return cam, cam.preview
if not hasattr(cam, "raw"):
raise RuntimeError(
f"ColorCamera {sensor} não expõe raw output."
)
return cam, cam.raw
mono = self.pipeline.create(dai.node.MonoCamera)
mono.setBoardSocket(socket)
resolution = getattr(
dai.MonoCameraProperties.SensorResolution,
profile["resolution_name"],
None,
)
try:
mono.setResolution(dai.MonoCameraProperties.SensorResolution.THE_800_P)
except Exception:
try:
mono.setResolution(dai.MonoCameraProperties.SensorResolution.THE_720_P)
except Exception:
try:
mono.setResolution(dai.MonoCameraProperties.SensorResolution.THE_400_P)
except Exception:
pass
if resolution is None:
raise RuntimeError(
f"DepthAI não expõe MonoCamera "
f"{profile['resolution_name']} para {sensor}."
)
mono.setResolution(resolution)
mono.setFps(float(self.fps))
if self._is_preview_mode():
return mono, mono.out
if not hasattr(mono, "raw"):
if self.product_contract:
raise RuntimeError(
f"MonoCamera {sensor} não expõe raw output."
)
return mono, mono.out
if hasattr(mono, "raw"):
return mono, mono.raw
return mono, mono.raw
return mono, mono.out
def _create_imu_node(self, pipeline):
self.has_imu_pipeline = False
@ -1306,13 +896,6 @@ class OakFcc3Manager:
self.pipeline = dai.Pipeline()
features = self.device.getConnectedCameraFeatures()
self._validate_connected_hardware(features)
if self.product_contract and self._is_multispec_mode():
raise RuntimeError(
"MULTISPEC alinhado na OAK é legado e não faz parte do "
"contrato de produção. Use RAW_BRUTO."
)
self.queues.clear()
self.buffers.clear()
@ -1360,44 +943,11 @@ class OakFcc3Manager:
self.buffers[cam_id] = deque(maxlen=self.buffer_size)
self.control_queues[cam_id] = None
sensor_u = str(f.sensorName or "").upper()
if self.product_contract or sensor_u in self.SENSOR_PROFILES:
profile = self._sensor_profile(sensor_u, role=role)
else:
profile = {
"kind": "unknown",
"width": int(getattr(f, "width", 0) or self.LEGACY_SENSOR_W),
"height": int(getattr(f, "height", 0) or self.LEGACY_SENSOR_H),
"resolution_name": None,
}
self.camera_info[cam_id] = {
"id": cam_id,
"socket": socket_name,
"sensor": sensor_u,
"sensor": f.sensorName,
"role": role,
"native_width": int(profile["width"]),
"native_height": int(profile["height"]),
"native_size": [
int(profile["width"]),
int(profile["height"]),
],
"sensor_kind": profile["kind"],
"resolution_mode": profile.get("resolution_name"),
"bayer_pattern": (
self.bayer_pattern
if str(role).lower() == "rgb"
else None
),
"product_expected": (
copy.deepcopy(
self.expected_camera_hardware.get(
str(role).lower()
)
)
if self.product_contract
else None
),
}
self._validate_capture_mode()
@ -1599,14 +1149,6 @@ class OakFcc3Manager:
"height": self.height,
"sensor_width": self.sensor_width,
"sensor_height": self.sensor_height,
"sensor_size_by_role": copy.deepcopy(self.sensor_size_by_role),
"camera_hardware_expected": copy.deepcopy(self.expected_camera_hardware),
"bayer_pattern": self.bayer_pattern,
"product_contract": bool(self.product_contract),
"require_product_contract": bool(self.require_product_contract),
"manager_version": OAK_FCC3_MANAGER_VERSION,
"module_params_schema": (self.module_params or {}).get("schema"),
"module_calibration_json": self.module_calibration_json,
"frame_type": self.frame_type,
"output_dtype": self.output_dtype,
"capture_mode": self.capture_mode,
@ -2184,32 +1726,6 @@ class OakFcc3Manager:
item["shape"] = list(arr.shape)
item["dtype"] = str(arr.dtype)
item["packed"] = True
item["native_width"] = int(
item.get("native_width", item["width"])
)
item["native_height"] = int(
item.get("native_height", item["height"])
)
item["native_size"] = [
item["native_width"],
item["native_height"],
]
if str(item.get("role", "")).lower() == "rgb":
item["bayer_pattern"] = self.bayer_pattern
if self.product_contract:
role = str(item.get("role", "")).lower()
expected = self.expected_camera_hardware.get(role)
if expected is not None:
actual = [int(item["width"]), int(item["height"])]
if actual != expected["size"]:
raise RuntimeError(
f"{cam_id}/{role}: RAW frame {actual} != "
f"hardware homologado {expected['size']}"
)
camera_info[cam_id] = item
@ -2217,18 +1733,6 @@ class OakFcc3Manager:
"frame_id": self.frame_id,
"backend": "oak_fcc3",
"frame_type": self.frame_type,
"product_contract": bool(self.product_contract),
"module_params_schema": (self.module_params or {}).get("schema"),
"module_calibration_json": self.module_calibration_json,
"device_mx_id": self.mx_id,
"reference_camera": "rgb",
"sensor_size_by_role": copy.deepcopy(self.sensor_size_by_role),
"camera_hardware": copy.deepcopy(
self.expected_camera_hardware
if self.product_contract
else {}
),
"bayer_pattern": self.bayer_pattern,
"capture_mode": self.capture_mode,
"output_dtype": self.output_dtype,
"dtype": self.output_dtype,
@ -2417,65 +1921,18 @@ class OakFcc3Manager:
return result
def apply_initial_camera_controls_to_node(self, cam, cam_id):
"""
Aplica Startup Profile antes de startPipeline().
O Client pode reaplicar depois do start como confirmação, mas os
primeiros frames já nascem sob a política homologada.
"""
ctrl_state = self.camera_controls.get(cam_id, {})
ae = bool(ctrl_state.get("ae_enable", False))
awb = bool(ctrl_state.get("awb_enable", False))
exp_us = int(ctrl_state.get("exposure_time_us") or 15000)
gain = float(ctrl_state.get("analogue_gain") or 1.0)
if ae:
fn = getattr(cam.initialControl, "setAutoExposureEnable", None)
if callable(fn):
fn()
elif self.product_contract:
raise RuntimeError(
f"{cam_id}: DepthAI sem setAutoExposureEnable no initialControl."
)
else:
if not ae:
cam.initialControl.setManualExposure(
exp_us,
self._gain_to_iso(gain),
)
role = str(self.roles.get(cam_id, "")).lower()
if role == "rgb":
if awb:
fn = getattr(
cam.initialControl,
"setAutoWhiteBalanceLock",
None,
)
if callable(fn):
fn(False)
else:
try:
cam.initialControl.setAutoWhiteBalanceMode(
dai.CameraControl.AutoWhiteBalanceMode.AUTO
)
except Exception:
if self.product_contract:
raise RuntimeError(
f"{cam_id}: não foi possível habilitar AWB inicial."
)
else:
fn = getattr(
cam.initialControl,
"setAutoWhiteBalanceLock",
None,
)
if callable(fn):
fn(True)
def _send_control(self, cam_id, ctrl):
if cam_id not in self.control_queues:
raise RuntimeError(f"Fila de controle não existe para {cam_id}")

1205
AgroBase/AgroBase/bin/x64/Debug/Python/Scripts/workers/camera_worker/oak_fcc3_core/raw_processor_preview.py
View File

File diff suppressed because it is too large Load Diff

985
Python/OAK/datasets/oak-fcc-3/_0_capture.py
View File

File diff suppressed because it is too large Load Diff

384
Python/OAK/datasets/oak-fcc-3/calibration/_0_module_profile_tool.py
View File

@ -13,7 +13,8 @@ arquivo que só deve existir ao final da linha de calibração.
Responsabilidades:
- selecionar e descobrir o dispositivo DepthAI;
- validar CAM_A RGB + CAM_B RE + CAM_C NIR;
- resolver/confirmar Bayer da RGB usando preview RAW10;
- resolver/confirmar Bayer da RGB usando o RawProcessorPreview oficial;
- confirmar a orientação física/canônica das três câmeras;
- registrar o domínio de decode RGB escolhido;
- criar a pasta oficial do perfil (ex.: calibration/mp_ar0234);
- declarar caminhos oficiais para todas as calibrações seguintes;
@ -32,6 +33,11 @@ Exemplos:
Teclas no preview:
B alterna Bayer RGB
1/2/3 seleciona RGB/RE/NIR
R gira a câmera selecionada 90 graus no sentido horário
H alterna flip horizontal da câmera selecionada
V alterna flip vertical da câmera selecionada
D restaura orientação padrão da câmera selecionada
S salva snapshot
ENTER confirma e promove o profile
Q/ESC cancela sem alterar o profile ativo
@ -55,6 +61,8 @@ import cv2
import depthai as dai
import numpy as np
from core.raw_processor_preview import RawProcessorPreview
SCHEMA = "multispec_module_profile_v1"
SELECTOR_SCHEMA = "multispec_active_module_profile_v1"
@ -256,6 +264,8 @@ class CameraSpec:
resolution_name: str
is_color: bool
preview_rotate_deg: int
preview_flip_horizontal: bool
preview_flip_vertical: bool
stream_name: str
@ -302,6 +312,8 @@ def validate_and_build_specs(rows: list[dict]) -> Dict[str, CameraSpec]:
resolution_name=str(cfg["resolution_enum"]),
is_color=(role == "rgb"),
preview_rotate_deg=int(cfg["preview_rotate_deg"]),
preview_flip_horizontal=False,
preview_flip_vertical=False,
stream_name=f"profile_{role}",
)
@ -364,77 +376,12 @@ def build_preview_pipeline(specs: Dict[str, CameraSpec], fps: float):
return pipeline
def unpack_raw10(data, width: int, height: int, stride: Optional[int] = None):
width = int(width)
height = int(height)
if width <= 0 or height <= 0 or width % 4:
raise ValueError(f"Dimensão RAW10 inválida: {width}x{height}")
raw = np.asarray(data, dtype=np.uint8).reshape(-1)
payload_bytes = (width // 4) * 5
candidates = []
if stride is not None:
try:
candidates.append(int(stride))
except Exception:
pass
if height > 0 and raw.size % height == 0:
candidates.append(raw.size // height)
candidates.append(payload_bytes)
row_stride = next(
(x for x in candidates if x >= payload_bytes and x * height <= raw.size),
None,
)
if row_stride is None:
raise ValueError(
f"RAW10 curto: bytes={raw.size}, payload mínimo={payload_bytes*height}"
)
rows = raw[: row_stride * height].reshape(height, row_stride)
groups = rows[:, :payload_bytes].reshape(height, width // 4, 5)
out = np.empty((height, width // 4, 4), dtype=np.uint16)
out[:, :, 0] = (
groups[:, :, 0].astype(np.uint16) << 2
) | (groups[:, :, 4] & 0x03)
out[:, :, 1] = (
groups[:, :, 1].astype(np.uint16) << 2
) | ((groups[:, :, 4] >> 2) & 0x03)
out[:, :, 2] = (
groups[:, :, 2].astype(np.uint16) << 2
) | ((groups[:, :, 4] >> 4) & 0x03)
out[:, :, 3] = (
groups[:, :, 3].astype(np.uint16) << 2
) | ((groups[:, :, 4] >> 6) & 0x03)
return np.ascontiguousarray(out.reshape(height, width))
def bayer_code(pattern: str, algorithm: str):
pattern = str(pattern).upper()
algorithm = str(algorithm).lower()
if algorithm == "ea":
table = {
"BGGR": cv2.COLOR_BayerRG2RGB_EA,
"RGGB": cv2.COLOR_BayerBG2RGB_EA,
"GRBG": cv2.COLOR_BayerGR2RGB_EA,
"GBRG": cv2.COLOR_BayerGB2RGB_EA,
}
else:
table = {
"BGGR": cv2.COLOR_BayerRG2RGB,
"RGGB": cv2.COLOR_BayerBG2RGB,
"GRBG": cv2.COLOR_BayerGR2RGB,
"GBRG": cv2.COLOR_BayerGB2RGB,
}
return table[pattern]
def rgb_raw_preview(pkt, spec: CameraSpec, pattern: str, algorithm: str):
def rgb_raw_preview(
pkt,
spec: CameraSpec,
processor: RawProcessorPreview,
):
"""Preview RGB oficial; não mantém um segundo mini-ISP neste script."""
raw_type = str(pkt.getType()).upper()
if "PACK10" not in raw_type and "RAW10" not in raw_type:
raise RuntimeError(
@ -454,31 +401,50 @@ def rgb_raw_preview(pkt, spec: CameraSpec, pattern: str, algorithm: str):
except Exception:
stride = None
raw16 = unpack_raw10(pkt.getData(), width, height, stride)
rgb16 = cv2.cvtColor(raw16, bayer_code(pattern, algorithm))
rgb = np.clip(rgb16.astype(np.float32) / 1023.0, 0.0, 1.0)
packed_data = np.asarray(pkt.getData(), dtype=np.uint8).reshape(-1)
payload_per_row = (width // 4) * 5
if (
stride is None
or stride < payload_per_row
or stride * height > packed_data.size
):
# Algumas versões do DepthAI reportam getStride() no domínio dos
# pixels, não dos bytes PACK10. Nesse caso o tamanho real vence.
stride = None
# Stretch comum aos canais: melhora a inspeção sem alterar o balanço RGB.
sample = rgb[::8, ::8]
lo = float(np.percentile(sample, 0.5))
hi = float(np.percentile(sample, 99.5))
if hi <= lo + 1e-6:
hi = lo + 1e-6
rgb = np.clip((rgb - lo) / (hi - lo), 0.0, 1.0)
# OpenCV exibe BGR.
return np.ascontiguousarray((rgb[:, :, ::-1] * 255.0).astype(np.uint8))
raw16 = processor.unpack_raw10_packed(
packed_data,
stride=stride,
)
return processor.raw16_to_preview_bgr(raw16)
def apply_preview_orientation(img: np.ndarray, rotate_deg: int):
rotate_deg = int(rotate_deg) % 360
def normalize_orientation(cfg: dict) -> dict:
rotate_deg = int((cfg or {}).get("rotate_deg", 0)) % 360
if rotate_deg not in (0, 90, 180, 270):
raise ValueError(f"Rotação inválida: {rotate_deg}")
return {
"rotate_deg": rotate_deg,
"flip_horizontal": bool((cfg or {}).get("flip_horizontal", False)),
"flip_vertical": bool((cfg or {}).get("flip_vertical", False)),
"source": str((cfg or {}).get("source") or "unknown"),
}
def apply_preview_orientation(img: np.ndarray, cfg: dict):
cfg = normalize_orientation(cfg)
rotate_deg = cfg["rotate_deg"]
if rotate_deg == 90:
return cv2.rotate(img, cv2.ROTATE_90_CLOCKWISE)
if rotate_deg == 180:
return cv2.rotate(img, cv2.ROTATE_180)
if rotate_deg == 270:
return cv2.rotate(img, cv2.ROTATE_90_COUNTERCLOCKWISE)
return img
img = cv2.rotate(img, cv2.ROTATE_90_CLOCKWISE)
elif rotate_deg == 180:
img = cv2.rotate(img, cv2.ROTATE_180)
elif rotate_deg == 270:
img = cv2.rotate(img, cv2.ROTATE_90_COUNTERCLOCKWISE)
if cfg["flip_horizontal"]:
img = cv2.flip(img, 1)
if cfg["flip_vertical"]:
img = cv2.flip(img, 0)
return np.ascontiguousarray(img)
def overlay(img, lines, x=12, y=24, scale=0.48, step=21):
@ -499,6 +465,7 @@ def preview_and_confirm(
dev_info,
specs: Dict[str, CameraSpec],
initial_bayer: str,
initial_orientation: Dict[str, dict],
algorithm: str,
fps: float,
panel_width: int,
@ -506,8 +473,25 @@ def preview_and_confirm(
snapshots_dir: Path,
):
pattern = initial_bayer
orientation = {
role: normalize_orientation(initial_orientation[role])
for role in ROLES
}
orientation_defaults = {
role: dict(orientation[role])
for role in ROLES
}
selected_role = "rgb"
preview_processors = {
candidate: RawProcessorPreview(
sensor_width=specs["rgb"].width,
sensor_height=specs["rgb"].height,
bayer_pattern=candidate,
)
for candidate in BAYER_PATTERNS
}
pipeline = build_preview_pipeline(specs, fps)
window = "Module Profile - Bayer Confirmation"
window = "Module Profile - Physical Sanity Check"
cv2.namedWindow(window, cv2.WINDOW_NORMAL)
try:
@ -538,7 +522,11 @@ def preview_and_confirm(
spec = specs[role]
pkt = latest[role]
if role == "rgb":
view = rgb_raw_preview(pkt, spec, pattern, algorithm)
view = rgb_raw_preview(
pkt,
spec,
preview_processors[pattern],
)
else:
view = pkt.getCvFrame()
if view.ndim == 2:
@ -546,7 +534,7 @@ def preview_and_confirm(
view = apply_preview_orientation(
view,
spec.preview_rotate_deg,
orientation[role],
)
panel = cv2.resize(
view,
@ -554,16 +542,33 @@ def preview_and_confirm(
interpolation=cv2.INTER_AREA,
)
lines = [
f"{role.upper()} | {spec.socket} | {spec.sensor}",
(
f"> {role.upper()} | {spec.socket} | {spec.sensor}"
if role == selected_role
else f" {role.upper()} | {spec.socket} | {spec.sensor}"
),
f"native={spec.width}x{spec.height}",
f"packet={latest_type[role]}",
(
f"ROT={orientation[role]['rotate_deg']} "
f"FH={int(orientation[role]['flip_horizontal'])} "
f"FV={int(orientation[role]['flip_vertical'])}"
),
]
if role == "rgb":
lines += [
f"BAYER={pattern} | DEMOSAIC={algorithm}",
f"BAYER={pattern} | PREVIEW=RawProcessorPreview",
"Confira vermelho/azul em objeto conhecido",
]
overlay(panel, lines)
if role == selected_role:
cv2.rectangle(
panel,
(2, 2),
(panel_width - 3, panel_height - 3),
(0, 255, 255),
4,
)
panels.append(panel)
info = np.zeros(
@ -575,14 +580,19 @@ def preview_and_confirm(
"",
f"RGB Bayer selecionado: {pattern}",
f"RGB processing: linear_demosaic / {algorithm}",
f"Camera selecionada: {selected_role.upper()}",
"",
"B = alternar Bayer",
"1/2/3 = selecionar RGB/RE/NIR",
"R = girar 90 graus horario",
"H/V = flip horizontal/vertical",
"D = restaurar orientacao padrao",
"S = snapshot",
"ENTER = confirmar e promover",
"Q/ESC = cancelar",
"",
"Use um objeto vermelho e outro azul",
"para confirmar que R/B não estão trocados.",
"Confirme cores e o mesmo sentido fisico",
"nas tres cameras antes de pressionar ENTER.",
])
last_board = np.vstack([
@ -597,21 +607,48 @@ def preview_and_confirm(
key = cv2.waitKey(1) & 0xFF
if key in (ord("q"), ord("Q"), 27):
raise KeyboardInterrupt("Cancelado durante confirmação do Bayer.")
raise KeyboardInterrupt("Cancelado durante sanity check físico.")
if key == ord("1"):
selected_role = "rgb"
elif key == ord("2"):
selected_role = "re"
elif key == ord("3"):
selected_role = "nir"
if key in (ord("b"), ord("B")):
idx = (BAYER_PATTERNS.index(pattern) + 1) % len(BAYER_PATTERNS)
pattern = BAYER_PATTERNS[idx]
if key in (ord("r"), ord("R")):
orientation[selected_role]["rotate_deg"] = (
int(orientation[selected_role]["rotate_deg"]) + 90
) % 360
orientation[selected_role]["source"] = "raw_preview_confirmed"
if key in (ord("h"), ord("H")):
orientation[selected_role]["flip_horizontal"] = not bool(
orientation[selected_role]["flip_horizontal"]
)
orientation[selected_role]["source"] = "raw_preview_confirmed"
if key in (ord("v"), ord("V")):
orientation[selected_role]["flip_vertical"] = not bool(
orientation[selected_role]["flip_vertical"]
)
orientation[selected_role]["source"] = "raw_preview_confirmed"
if key in (ord("d"), ord("D")):
orientation[selected_role] = dict(
orientation_defaults[selected_role]
)
if key in (ord("s"), ord("S")) and last_board is not None:
ensure_dir(snapshots_dir)
cv2.imwrite(
str(snapshots_dir / f"bayer_{pattern}_{stamp()}.png"),
str(snapshots_dir / f"physical_check_{pattern}_{stamp()}.png"),
last_board,
)
if key in (10, 13) and last_board is not None:
ensure_dir(snapshots_dir)
confirmed_path = snapshots_dir / "bayer_confirmed.png"
for role in ROLES:
orientation[role]["source"] = "raw_preview_confirmed"
confirmed_path = snapshots_dir / "physical_sanity_confirmed.png"
cv2.imwrite(str(confirmed_path), last_board)
return pattern, confirmed_path
return pattern, orientation, confirmed_path
time.sleep(0.002)
finally:
@ -625,9 +662,9 @@ def artifact_paths(profile_dir: Path) -> dict:
"intrinsics_json": norm_path(profile_dir / "intrinsics_calibration_v1.json"),
"flatfield_npz": norm_path(profile_dir / "flatfield_maps_v1.npz"),
"flatfield_json": norm_path(profile_dir / "flatfield_maps_v1.json"),
"radiometry_json": norm_path(profile_dir / "radiometry_calibration_v1.json"),
"camera_startup_json": norm_path(profile_dir / "camera_startup_profile_v1.json"),
"homography_json": norm_path(profile_dir / "homography_calibration_v1.json"),
"radiometry_json": norm_path(profile_dir / "radiometry_calibration_v5.json"),
"camera_startup_json": norm_path(profile_dir / "camera_startup_profile_v3.json"),
"homography_json": norm_path(profile_dir / "homography_calibration_v4.json"),
"module_params_json": norm_path(profile_dir / "module_params.json"),
"module_params_assembly_report_json": norm_path(
profile_dir / "module_params_assembly_report.json"
@ -652,19 +689,27 @@ def build_profile(
profile_name: str,
profile_dir: Path,
bayer: str,
preview_orientation: Dict[str, dict],
confirmed_with_raw_preview: bool,
confirmation_snapshot: Optional[Path],
):
orientation = {
role: {
"rotate_deg": int(specs[role].preview_rotate_deg),
"flip_horizontal": False,
"flip_vertical": False,
"source": "sensor_default",
}
role: normalize_orientation(preview_orientation[role])
for role in ROLES
}
resolved_setup = {}
for role in ROLES:
item = asdict(specs[role])
item["preview_rotate_deg"] = int(orientation[role]["rotate_deg"])
item["preview_flip_horizontal"] = bool(
orientation[role]["flip_horizontal"]
)
item["preview_flip_vertical"] = bool(
orientation[role]["flip_vertical"]
)
resolved_setup[role] = item
hardware_signature = {
role: {
"socket": specs[role].socket,
@ -686,10 +731,7 @@ def build_profile(
"depthai_version": getattr(dai, "__version__", "unknown"),
"usb_speed": usb_speed,
"hardware_signature": hardware_signature,
"resolved_setup": {
role: asdict(specs[role])
for role in ROLES
},
"resolved_setup": resolved_setup,
"camera_inventory": rows,
"rgb_decode": {
"bayer_pattern": bayer,
@ -708,9 +750,23 @@ def build_profile(
),
},
"preview_orientation_defaults": orientation,
"mounting_sanity_check": {
"confirmed_with_live_preview": bool(confirmed_with_raw_preview),
"preview_engine": "core.raw_processor_preview.RawProcessorPreview",
"orientation_by_role": orientation,
"confirmation_snapshot": (
norm_path(confirmation_snapshot)
if confirmation_snapshot is not None
else None
),
},
"geometry_bootstrap": {
"native_space": "native_stream_no_external_undistort",
"orientation_status": "preview_default_until_homography",
"orientation_status": (
"raw_preview_confirmed_until_homography"
if confirmed_with_raw_preview
else "default_or_cli_headless_until_homography"
),
"note": (
"A Homography homologada será a autoridade final da orientação "
"científica e do espaço canônico."
@ -730,6 +786,15 @@ def build_profile(
"mode": args.rgb_processing_mode,
"demosaic_algorithm": args.demosaic_algorithm,
},
"camera_orientation_bootstrap": {
"enabled": True,
"input_space": "native_stream_no_external_undistort",
"output_space": (
"canonical_oriented_stream_no_external_undistort"
),
"by_role": orientation,
"authority": "module_profile_until_homography",
},
},
"artifact_paths": artifact_paths(profile_dir),
"traceability": {
@ -740,6 +805,13 @@ def build_profile(
},
"notes": args.notes or "",
},
"capture_synchronization": {
"hardware_sync_enabled": bool(args.hardware_sync_enabled),
"frame_sync_master": args.frame_sync_master,
"software_sync_mode": "best",
"sync_tolerance_ms": 12.0,
"buffer_size": 8,
},
}
@ -789,6 +861,29 @@ def backup_if_exists(path: Path):
return None
def resolve_initial_orientation(
specs: Dict[str, CameraSpec],
args,
) -> Dict[str, dict]:
result = {}
for role in ROLES:
rotate_override = getattr(args, f"{role}_preview_rotate_deg")
flip_h = bool(getattr(args, f"{role}_preview_flip_horizontal"))
flip_v = bool(getattr(args, f"{role}_preview_flip_vertical"))
has_cli_override = rotate_override is not None or flip_h or flip_v
result[role] = normalize_orientation({
"rotate_deg": (
specs[role].preview_rotate_deg
if rotate_override is None
else rotate_override
),
"flip_horizontal": flip_h,
"flip_vertical": flip_v,
"source": "cli" if has_cli_override else "sensor_default",
})
return result
def main():
parser = argparse.ArgumentParser(
description="Bootstrap do profile físico/decode do módulo multiespectral.",
@ -826,6 +921,25 @@ def main():
parser.add_argument("--fps", type=float, default=10.0)
parser.add_argument("--panel-width", type=int, default=640)
parser.add_argument("--panel-height", type=int, default=400)
for role in ROLES:
parser.add_argument(
f"--{role}-preview-rotate-deg",
type=int,
default=None,
choices=[0, 90, 180, 270],
help=(
f"Orientação inicial de {role.upper()}; pode ser ajustada "
"interativamente no preview."
),
)
parser.add_argument(
f"--{role}-preview-flip-horizontal",
action="store_true",
)
parser.add_argument(
f"--{role}-preview-flip-vertical",
action="store_true",
)
parser.add_argument(
"--no-preview",
action="store_true",
@ -841,6 +955,16 @@ def main():
parser.add_argument("--lens-re", default="")
parser.add_argument("--lens-nir", default="")
parser.add_argument("--notes", default="")
parser.add_argument(
"--hardware-sync-enabled",
action=argparse.BooleanOptionalAction,
default=False,
)
parser.add_argument(
"--frame-sync-master",
default="CAM_A",
choices=["CAM_A", "CAM_B", "CAM_C"],
)
args = parser.parse_args()
if args.no_preview and not args.yes:
@ -849,6 +973,7 @@ def main():
dev_info = select_device_info(args.mx_id)
rows, actual_mx, usb_speed = discover(dev_info)
specs = validate_and_build_specs(rows)
selected_orientation = resolve_initial_orientation(specs, args)
rgb_sensor = specs["rgb"].sensor
default_profile_name = f"mp_{rgb_sensor.lower()}"
@ -898,6 +1023,13 @@ def main():
f"RGB decode : Bayer={selected_bayer} ({bayer_initial_source}) | "
f"mode={args.rgb_processing_mode} | algo={args.demosaic_algorithm}"
)
for role in ROLES:
orient = selected_orientation[role]
print(
f"{role.upper():3s} orient. : rotate={orient['rotate_deg']:3d} | "
f"flip_h={orient['flip_horizontal']} | "
f"flip_v={orient['flip_vertical']} | source={orient['source']}"
)
print("=" * 88)
confirmed_with_raw_preview = False
@ -905,10 +1037,15 @@ def main():
try:
if not args.no_preview:
selected_bayer, confirmation_snapshot = preview_and_confirm(
(
selected_bayer,
selected_orientation,
confirmation_snapshot,
) = preview_and_confirm(
dev_info=dev_info,
specs=specs,
initial_bayer=selected_bayer,
initial_orientation=selected_orientation,
algorithm=args.demosaic_algorithm,
fps=args.fps,
panel_width=args.panel_width,
@ -926,6 +1063,7 @@ def main():
profile_name=profile_name,
profile_dir=profile_dir,
bayer=selected_bayer,
preview_orientation=selected_orientation,
confirmed_with_raw_preview=confirmed_with_raw_preview,
confirmation_snapshot=confirmation_snapshot,
)
@ -965,6 +1103,14 @@ def main():
print(f"[SELECTOR] {active_selector}")
print(f"[SHA256] {selector['profile_sha256']}")
print(f"[BAYER] {selected_bayer}")
for role in ROLES:
orient = selected_orientation[role]
print(
f"[ORIENTATION] {role.upper():3s} | "
f"rotate={orient['rotate_deg']} | "
f"flip_h={orient['flip_horizontal']} | "
f"flip_v={orient['flip_vertical']}"
)
if backup_profile:
print(f"[BACKUP PROFILE] {backup_profile}")
if backup_selector:

112
Python/OAK/datasets/oak-fcc-3/calibration/_8_build_module_params.py
View File

@ -108,6 +108,7 @@ from __future__ import annotations
import argparse
import hashlib
import json
import math
import os
import shutil
import zipfile
@ -536,6 +537,78 @@ def resolve_output_path(value: str | Path) -> Path:
return Path(value).expanduser()
def normalize_capture_synchronization(
value: dict,
hardware_signature: dict,
) -> dict:
"""Valida e normaliza o contrato de sincronismo vindo do Module Profile."""
if not isinstance(value, dict):
raise RuntimeError(
"Module Profile sem capture_synchronization. "
"Execute novamente o Script 0 revisado."
)
enabled = value.get("hardware_sync_enabled")
if not isinstance(enabled, bool):
raise RuntimeError(
"capture_synchronization.hardware_sync_enabled deve ser bool."
)
master = str(value.get("frame_sync_master") or "").upper().strip()
valid_sockets = {
str(item["socket"]).upper()
for item in normalize_signature(hardware_signature).values()
}
if master not in valid_sockets:
raise RuntimeError(
"capture_synchronization.frame_sync_master inválido: "
f"{master!r}. Disponíveis={sorted(valid_sockets)}"
)
software_mode = str(
value.get("software_sync_mode") or ""
).lower().strip()
if not software_mode:
raise RuntimeError(
"capture_synchronization.software_sync_mode ausente."
)
try:
tolerance_ms = float(value.get("sync_tolerance_ms"))
except Exception as exc:
raise RuntimeError(
"capture_synchronization.sync_tolerance_ms inválido."
) from exc
if not math.isfinite(tolerance_ms) or tolerance_ms <= 0.0:
raise RuntimeError(
"capture_synchronization.sync_tolerance_ms deve ser finito e > 0."
)
buffer_size = value.get("buffer_size")
if isinstance(buffer_size, bool):
raise RuntimeError(
"capture_synchronization.buffer_size deve ser inteiro positivo."
)
try:
buffer_size = int(buffer_size)
except Exception as exc:
raise RuntimeError(
"capture_synchronization.buffer_size inválido."
) from exc
if buffer_size <= 0:
raise RuntimeError(
"capture_synchronization.buffer_size deve ser > 0."
)
return {
"hardware_sync_enabled": enabled,
"frame_sync_master": master,
"software_sync_mode": software_mode,
"sync_tolerance_ms": tolerance_ms,
"buffer_size": buffer_size,
}
def validate_module_profile(profile: dict, profile_path: Path):
if profile.get("schema") != MODULE_PROFILE_SCHEMA:
raise RuntimeError(
@ -602,6 +675,11 @@ def validate_module_profile(profile: dict, profile_path: Path):
if not artifacts.get(key):
raise RuntimeError(f"Module Profile sem artifact_paths.{key}.")
normalize_capture_synchronization(
profile.get("capture_synchronization"),
signature,
)
def load_module_profile_contract(
module_profile_arg: Optional[str],
@ -2602,6 +2680,10 @@ def runtime_policy_from_module_profile(profile: dict) -> dict:
sensor nem herança de um module_params anterior.
"""
decode = profile["rgb_decode"]
capture_sync = normalize_capture_synchronization(
profile.get("capture_synchronization"),
profile.get("hardware_signature"),
)
declared = profile.get("runtime_policy")
if declared is None:
declared = {}
@ -2614,6 +2696,7 @@ def runtime_policy_from_module_profile(profile: dict) -> dict:
"capture_mode_requested": "AUTO",
"capture_mode_effective": "AUTO",
"raw_policy": "allow_single",
"capture_synchronization": capture_sync,
"rgb_processing": {
"mode": str(decode["mode"]).lower(),
"demosaic_algorithm": str(decode["demosaic_algorithm"]).lower(),
@ -2781,6 +2864,9 @@ def build_final_module_params(
"raw_policy": runtime_policy[
"raw_policy"
],
"capture_synchronization": deepcopy(
runtime_policy["capture_synchronization"]
),
# Compatibilidade root: RGB reference camera
"sensor_width": int(
@ -2958,6 +3044,7 @@ def validate_final_module_params(
"bayer_pattern",
"sensor_size_by_role",
"camera_hardware",
"capture_synchronization",
"camera_orientation",
"rgb_processing",
"camera_settings",
@ -3013,6 +3100,15 @@ def validate_final_module_params(
if str(mp.get("bayer_pattern") or "").upper() not in BAYER_PATTERNS:
raise RuntimeError("bayer_pattern final inválido.")
normalized_sync = normalize_capture_synchronization(
mp.get("capture_synchronization"),
mp.get("camera_hardware"),
)
if mp.get("capture_synchronization") != normalized_sync:
raise RuntimeError(
"capture_synchronization final não está normalizado."
)
rgb_processing = mp.get("rgb_processing")
if not isinstance(rgb_processing, dict):
raise RuntimeError("rgb_processing final inválido.")
@ -3698,6 +3794,9 @@ def main():
provenance = {
"module_profile_contract": module_profile_provenance(profile_contract),
"configuration_source": "module_profile_only",
"capture_synchronization": deepcopy(
runtime_policy["capture_synchronization"]
),
"hardware_signature": deepcopy(
common_signature
),
@ -3832,6 +3931,9 @@ def main():
"camera_orientation": deepcopy(
camera_orientation
),
"capture_synchronization": deepcopy(
runtime_policy["capture_synchronization"]
),
"runtime_policy": runtime_policy,
"calibration_provenance": provenance,
"final_checks": {
@ -3847,6 +3949,7 @@ def main():
"hardware_consistency": "pass",
"geometry_consistency": "pass",
"camera_orientation": "pass",
"capture_synchronization": "pass",
"orientation_bootstrap_consistency": "pass",
"startup_radiometry_consistency": "pass",
"flatfield_npz_integrity": "pass",
@ -3874,6 +3977,15 @@ def main():
f"{common_signature['nir']['size'][0]}x{common_signature['nir']['size'][1]}"
)
print(f"Bayer : {bayer}")
capture_sync = runtime_policy["capture_synchronization"]
print(
"Capture sync: "
f"hardware={'ON' if capture_sync['hardware_sync_enabled'] else 'OFF'} | "
f"master={capture_sync['frame_sync_master']} | "
f"software={capture_sync['software_sync_mode']} | "
f"tolerance={capture_sync['sync_tolerance_ms']:.1f}ms | "
f"buffer={capture_sync['buffer_size']}"
)
print(
f"RGB process : "
f"{runtime_policy['rgb_processing'].get('mode')}"

17
Python/OAK/datasets/oak-fcc-3/core/oak_fcc3_client.py
View File

@ -41,12 +41,17 @@ class OakFcc3Client:
capture_mode="AUTO",
raw_policy="allow_single",
module_calibration_json=None,
sync_mode="best",
sync_tolerance_ms=25.0,
mx_id=None,
imu_modo="rotation_vector",
imu_freq_hz=200,
evaluate_quality=True,
hardware_sync_enabled=None,
frame_sync_master=None,
sync_mode=None,
sync_tolerance_ms=None,
buffer_size=None,
**kwargs,
):
self.width = width
@ -85,14 +90,18 @@ class OakFcc3Client:
output_dtype=output_dtype,
capture_mode=capture_mode,
raw_policy=raw_policy,
sync_mode=sync_mode,
sync_tolerance_ms=sync_tolerance_ms,
mx_id=self.mx_id,
module_calibration_json=module_calibration_json,
imu_modo=self.imu_modo,
imu_freq_hz=self.imu_freq_hz,
hardware_sync_enabled=hardware_sync_enabled,
frame_sync_master=frame_sync_master,
sync_mode=sync_mode,
sync_tolerance_ms=sync_tolerance_ms,
buffer_size=buffer_size,
**kwargs,
)

29
Python/OAK/datasets/oak-fcc-3/core/oak_fcc3_manager.py
View File

@ -44,11 +44,11 @@ class OakFcc3Manager:
capture_mode="AUTO",
raw_policy="allow_single",
roles=None,
sync_mode="best",
hardware_sync_enabled=True,
frame_sync_master="CAM_A",
sync_tolerance_ms=12.0,
buffer_size=8,
sync_mode=None,
hardware_sync_enabled=None,
frame_sync_master=None,
sync_tolerance_ms=None,
buffer_size=None,
only_camera=None,
mx_id=None,
module_calibration_json=None,
@ -56,9 +56,6 @@ class OakFcc3Manager:
imu_modo="rotation_vector",
imu_freq_hz=200,
):
self.hardware_sync_enabled = hardware_sync_enabled
self.frame_sync_master = frame_sync_master
self.fps = fps
# Para compatibilidade, mantemos width/height.
@ -83,10 +80,6 @@ class OakFcc3Manager:
"CAM_C": "nir",
}
self.sync_mode = sync_mode
self.sync_tolerance_ms = sync_tolerance_ms
self.buffer_size = buffer_size
self.mx_id = str(mx_id) if mx_id else None
self.dev_info = None
self.device = None
@ -127,6 +120,18 @@ class OakFcc3Manager:
self.fusion_config = (self.module_params or {}).get("fusion_config", {}) or {}
self.aligned_geometry = None
sync_cfg = (self.module_params.get("capture_synchronization", {}) if isinstance(self.module_params, dict) else {})
if hardware_sync_enabled is None: hardware_sync_enabled = sync_cfg.get("hardware_sync_enabled", False)
if frame_sync_master is None: frame_sync_master = sync_cfg.get("frame_sync_master", "CAM_A")
if sync_mode is None: sync_mode = sync_cfg.get("software_sync_mode", "best")
if sync_tolerance_ms is None: sync_tolerance_ms = sync_cfg.get("sync_tolerance_ms", 12.0)
if buffer_size is None: buffer_size = sync_cfg.get("buffer_size", 8)
self.hardware_sync_enabled = bool(hardware_sync_enabled)
self.frame_sync_master = str(frame_sync_master)
self.sync_mode = str(sync_mode)
self.sync_tolerance_ms = float(sync_tolerance_ms)
self.buffer_size = int(buffer_size)
self.async_capture_enabled = True
self.async_capture_mode = "latest" # latest | queue
self.async_capture_max_queue = 2

741
Python/OAK/datasets/oak-fcc-3/core/raw_processor_preview.py
View File

@ -1,126 +1,689 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
RawProcessorPreview
===================
Mini-ISP exclusivamente visual para gerar previews de anotação a partir do
RAW Bayer da câmera RGB.
Contrato:
- entrada científica permanece imutável;
- saída sempre BGR uint8 para OpenCV;
- preserva exatamente HxW do raster RGB;
- não aplica resize, crop, rotate/flip, undistort, Flat-Field ou Homography;
- não deve ser usado para formar o tensor de treino/inferência.
Pipeline visual padrão:
RAW10/RAW16 -> níveis robustos -> gamma -> demosaico EA/BGR ->
WB por regiões claras/neutras -> CLAHE suave em luminância ->
saturação leve -> contraste -> unsharp suave.
As APIs antigas foram preservadas. Chamadas que antes usavam
raw16_to_preview_bgr() ou raw16_to_preview_jpg_bytes() continuam válidas.
"""
from __future__ import annotations
from copy import deepcopy
from typing import Any, Optional
import cv2
import numpy as np
from typing import Optional
class RawProcessorPreview:
def __init__(self, sensor_width: int, sensor_height: int, bayer_pattern: str = "GBRG"):
self.sensor_width = sensor_width
self.sensor_height = sensor_height
self.bayer_pattern = bayer_pattern.upper()
VERSION = "raw_processor_preview_v2_2026_09_09"
CONTRACT_SCHEMA = "rgb_raw_annotation_preview_v2"
BAYER_PATTERNS = ("RGGB", "BGGR", "GRBG", "GBRG")
def raw16_to_vis8(
self, raw16: np.ndarray,
black_level: Optional[int] = None,
white_level: Optional[int] = None,
gamma: float = 2.2,
bit_depth: int = 10
) -> np.ndarray:
"""
Conversão para visualização:
- auto-level
- gamma
"""
max_val = float((1 << bit_depth) - 1)
DEFAULT_CONFIG = {
"bit_depth": 10,
"levels": {
"mode": "robust_percentile",
"black_percentile": 0.20,
"white_percentile": 99.80,
"sample_max_pixels": 500_000,
"minimum_span_codes": 16.0,
},
"gamma": 2.20,
"demosaic": {
"algorithm": "edge_aware",
"output_color_order": "BGR",
},
"white_balance": {
"enabled": True,
"method": "neutral_bright_with_gray_world_fallback",
"strength": 0.65,
"gain_min": 0.60,
"gain_max": 1.70,
"bright_percentile": 55.0,
"neutral_chroma_percentile": 40.0,
"min_neutral_pixels": 256,
},
"clahe": {
"enabled": True,
"clip_limit": 1.55,
"tile_grid_size": 8,
},
"saturation": {
"enabled": True,
"factor": 1.04,
},
"contrast": {
"enabled": True,
"alpha": 1.04,
"beta": 0.0,
},
"sharpen": {
"enabled": True,
"sigma": 0.85,
"amount": 0.55,
"threshold": 2.0,
},
}
raw = raw16.astype(np.float32)
def __init__(
self,
sensor_width: int,
sensor_height: int,
bayer_pattern: str = "GBRG",
config: Optional[dict] = None,
):
self.sensor_width = int(sensor_width)
self.sensor_height = int(sensor_height)
self.bayer_pattern = str(bayer_pattern).upper()
self.config = self._deep_merge(self.DEFAULT_CONFIG, config or {})
self.last_preview_report: dict[str, Any] = {}
self._validate_contract()
if black_level is None:
black_level = float(raw.min())
if white_level is None:
white_level = float(raw.max())
@staticmethod
def _deep_merge(base: dict, override: dict) -> dict:
result = deepcopy(base)
if not isinstance(override, dict):
raise TypeError("config do RawProcessorPreview deve ser dict.")
for key, value in override.items():
if isinstance(value, dict) and isinstance(result.get(key), dict):
result[key] = RawProcessorPreview._deep_merge(result[key], value)
else:
result[key] = deepcopy(value)
return result
if white_level <= black_level:
norm = raw / max_val
else:
norm = (raw - black_level) / (white_level - black_level)
norm = np.clip(norm, 0.0, 1.0)
if gamma is not None and gamma > 0:
norm = np.power(norm, 1.0 / gamma)
return (norm * 255.0).clip(0, 255).astype(np.uint8)
def _debayer_code(self):
mapping = {
"RGGB": cv2.COLOR_BayerRG2RGB_EA,
"BGGR": cv2.COLOR_BayerBG2RGB_EA,
"GRBG": cv2.COLOR_BayerGR2RGB_EA,
"GBRG": cv2.COLOR_BayerGB2RGB_EA,
}
if self.bayer_pattern not in mapping:
def _validate_contract(self):
if self.sensor_width <= 0 or self.sensor_height <= 0:
raise ValueError(
f"Dimensão de sensor inválida: {self.sensor_width}x{self.sensor_height}"
)
if self.sensor_width % 4 != 0:
raise ValueError(
f"RAW10 packed exige largura múltipla de 4: {self.sensor_width}"
)
if self.bayer_pattern not in self.BAYER_PATTERNS:
raise ValueError(f"Padrão Bayer não suportado: {self.bayer_pattern}")
return mapping[self.bayer_pattern]
bit_depth = int(self.config.get("bit_depth", 10))
if bit_depth < 8 or bit_depth > 16:
raise ValueError(f"bit_depth inválido: {bit_depth}")
def apply_preview_white_balance(self, bgr: np.ndarray, strength: float = 1.0) -> np.ndarray:
levels = self.config["levels"]
low = float(levels["black_percentile"])
high = float(levels["white_percentile"])
if not (0.0 <= low < high <= 100.0):
raise ValueError(f"Percentis de níveis inválidos: {low}, {high}")
gamma = float(self.config["gamma"])
if not np.isfinite(gamma) or gamma <= 0.0:
raise ValueError(f"Gamma inválido: {gamma}")
wb = self.config["white_balance"]
if float(wb["gain_min"]) <= 0.0 or float(wb["gain_max"]) < float(wb["gain_min"]):
raise ValueError("Limites de ganho do white balance inválidos.")
def get_preview_contract(self) -> dict:
"""Contrato serializável para registrar no meta do dataset."""
return {
"schema": self.CONTRACT_SCHEMA,
"version": self.VERSION,
"purpose": "human_annotation_only",
"geometry": "rgb_sensor_native_no_geometric_transform",
"sensor_size": [self.sensor_width, self.sensor_height],
"bayer_pattern": self.bayer_pattern,
"output": {
"layout": "HWC",
"color_order": "BGR",
"dtype": "uint8",
"range": [0, 255],
},
"calibrations_applied": [],
"config": deepcopy(self.config),
}
def get_last_preview_report(self) -> dict:
return deepcopy(self.last_preview_report)
# Nome explícito e amigável para integrações que só precisam registrar
# a configuração, sem necessariamente gerar um frame antes.
def describe_config(self) -> dict:
return self.get_preview_contract()
@staticmethod
def _finite_float(value, name: str) -> float:
result = float(value)
if not np.isfinite(result):
raise ValueError(f"{name} deve ser finito: {value!r}")
return result
@staticmethod
def _sample_for_stats(raw: np.ndarray, max_pixels: int) -> np.ndarray:
total = int(raw.size)
if total <= max_pixels:
return raw.reshape(-1)
stride = max(1, int(np.ceil(np.sqrt(total / float(max_pixels)))))
return raw[::stride, ::stride].reshape(-1)
def _resolve_levels(
self,
raw16: np.ndarray,
black_level: Optional[float],
white_level: Optional[float],
bit_depth: int,
black_percentile: Optional[float],
white_percentile: Optional[float],
) -> tuple[float, float, dict]:
max_code = float((1 << int(bit_depth)) - 1)
levels = self.config["levels"]
low_pct = float(
levels["black_percentile"]
if black_percentile is None
else black_percentile
)
high_pct = float(
levels["white_percentile"]
if white_percentile is None
else white_percentile
)
if not (0.0 <= low_pct < high_pct <= 100.0):
raise ValueError(f"Percentis inválidos: {low_pct}, {high_pct}")
sample = self._sample_for_stats(
raw16,
max(1, int(levels["sample_max_pixels"])),
).astype(np.float32, copy=False)
sample = sample[np.isfinite(sample)]
if sample.size == 0:
raise RuntimeError("RAW não contém pixels finitos para calcular níveis.")
black_source = "explicit"
white_source = "explicit"
if black_level is None:
black = float(np.percentile(sample, low_pct))
black_source = "percentile"
else:
black = self._finite_float(black_level, "black_level")
if white_level is None:
white = float(np.percentile(sample, high_pct))
white_source = "percentile"
else:
white = self._finite_float(white_level, "white_level")
black = float(np.clip(black, 0.0, max_code))
white = float(np.clip(white, 0.0, max_code))
minimum_span = float(levels["minimum_span_codes"])
fallback = False
if white <= black + minimum_span:
black = float(np.clip(np.min(sample), 0.0, max_code))
white = float(np.clip(np.max(sample), 0.0, max_code))
fallback = True
if white <= black:
black, white = 0.0, max_code
fallback = True
return black, white, {
"black_level": black,
"white_level": white,
"black_source": black_source,
"white_source": white_source,
"black_percentile": low_pct,
"white_percentile": high_pct,
"fallback_to_range": fallback,
"sample_pixels": int(sample.size),
}
def raw16_to_vis8(
self,
raw16: np.ndarray,
black_level: Optional[int] = None,
white_level: Optional[int] = None,
gamma: Optional[float] = None,
bit_depth: Optional[int] = None,
black_percentile: Optional[float] = None,
white_percentile: Optional[float] = None,
) -> np.ndarray:
"""Converte mosaico RAW16 para mosaico uint8 de visualização."""
raw = np.asarray(raw16)
if raw.ndim != 2:
raise ValueError(f"RAW16 deve ser 2D; shape={raw.shape}")
if raw.shape != (self.sensor_height, self.sensor_width):
raise ValueError(
f"RAW16 shape={raw.shape}; esperado="
f"{(self.sensor_height, self.sensor_width)}"
)
if not np.issubdtype(raw.dtype, np.integer):
raise TypeError(f"RAW16 deve possuir dtype inteiro; recebido={raw.dtype}")
bit_depth = int(
self.config["bit_depth"] if bit_depth is None else bit_depth
)
gamma_value = self._finite_float(
self.config["gamma"] if gamma is None else gamma,
"gamma",
)
if gamma_value <= 0.0:
raise ValueError("gamma deve ser > 0.")
black, white, level_report = self._resolve_levels(
raw,
black_level,
white_level,
bit_depth,
black_percentile,
white_percentile,
)
norm = (raw.astype(np.float32) - black) / max(white - black, 1.0)
np.clip(norm, 0.0, 1.0, out=norm)
if gamma_value != 1.0:
norm = np.power(norm, 1.0 / gamma_value)
self.last_preview_report = {
"stage": "raw16_to_vis8",
"bit_depth": bit_depth,
"gamma": gamma_value,
"levels": level_report,
}
return np.clip(norm * 255.0 + 0.5, 0, 255).astype(np.uint8)
def _debayer_code(self):
# A API desta classe promete BGR para cv2.imshow/cv2.imwrite.
mapping = {
"RGGB": ("COLOR_BayerRG2BGR_EA", "COLOR_BayerRG2BGR"),
"BGGR": ("COLOR_BayerBG2BGR_EA", "COLOR_BayerBG2BGR"),
"GRBG": ("COLOR_BayerGR2BGR_EA", "COLOR_BayerGR2BGR"),
"GBRG": ("COLOR_BayerGB2BGR_EA", "COLOR_BayerGB2BGR"),
}
preferred, fallback = mapping[self.bayer_pattern]
return getattr(cv2, preferred, getattr(cv2, fallback))
def apply_preview_white_balance(
self,
bgr: np.ndarray,
strength: float = 1.0,
method: Optional[str] = None,
gain_min: Optional[float] = None,
gain_max: Optional[float] = None,
return_report: bool = False,
):
"""
Gray-world simples para deixar o preview mais agradável.
Não usar no raw de treino.
WB visual robusto para campo agrícola.
Primeiro procura pixels claros e de baixo croma, reduzindo o risco de
o gray-world neutralizar toda a vegetação verde. Se não houver amostra
neutra suficiente, usa gray-world com ganhos limitados.
"""
img = bgr.astype(np.float32)
image = np.asarray(bgr)
if image.ndim != 3 or image.shape[2] != 3:
raise ValueError(f"WB requer BGR HWC; shape={image.shape}")
mean_b = float(img[:, :, 0].mean())
mean_g = float(img[:, :, 1].mean())
mean_r = float(img[:, :, 2].mean())
cfg = self.config["white_balance"]
method = str(method or cfg["method"]).lower()
strength = float(np.clip(self._finite_float(strength, "wb_strength"), 0.0, 1.0))
gain_min = float(cfg["gain_min"] if gain_min is None else gain_min)
gain_max = float(cfg["gain_max"] if gain_max is None else gain_max)
if gain_min <= 0.0 or gain_max < gain_min:
raise ValueError("Limites de ganho WB inválidos.")
mean_gray = (mean_b + mean_g + mean_r) / 3.0
sample = image[::4, ::4].astype(np.float32)
flat = sample.reshape(-1, 3)
intensity = flat.mean(axis=1)
chroma = (flat.max(axis=1) - flat.min(axis=1)) / np.maximum(intensity, 1.0)
eps = 1e-6
gain_b = mean_gray / max(mean_b, eps)
gain_g = mean_gray / max(mean_g, eps)
gain_r = mean_gray / max(mean_r, eps)
selected = flat
source = "gray_world"
if method.startswith("neutral_bright") and flat.shape[0] > 0:
bright_limit = float(np.percentile(intensity, float(cfg["bright_percentile"])))
bright_mask = intensity >= bright_limit
bright_chroma = chroma[bright_mask]
if bright_chroma.size:
neutral_limit = float(
np.percentile(
bright_chroma,
float(cfg["neutral_chroma_percentile"]),
)
)
neutral_mask = bright_mask & (chroma <= neutral_limit)
neutral = flat[neutral_mask]
if neutral.shape[0] >= int(cfg["min_neutral_pixels"]):
selected = neutral
source = "neutral_bright"
# strength=1 aplica total, strength=0 não aplica
gain_b = 1.0 + (gain_b - 1.0) * strength
gain_g = 1.0 + (gain_g - 1.0) * strength
gain_r = 1.0 + (gain_r - 1.0) * strength
means = selected.mean(axis=0)
target = float(means.mean())
gains = target / np.maximum(means, 1e-6)
gains = np.clip(gains, gain_min, gain_max)
gains = 1.0 + (gains - 1.0) * strength
img[:, :, 0] *= gain_b
img[:, :, 1] *= gain_g
img[:, :, 2] *= gain_r
out = np.clip(
image.astype(np.float32) * gains[None, None, :],
0,
255,
).astype(np.uint8)
report = {
"method_requested": method,
"source_used": source,
"strength": strength,
"means_bgr": [float(x) for x in means],
"gains_bgr": [float(x) for x in gains],
"selected_pixels": int(selected.shape[0]),
}
return (out, report) if return_report else out
return np.clip(img, 0, 255).astype(np.uint8)
def apply_preview_contrast(
self,
bgr: np.ndarray,
alpha: float = 1.08,
beta: float = 0.0,
) -> np.ndarray:
"""Compatibilidade: contraste/brilho visual sem alterar geometria."""
alpha = self._finite_float(alpha, "contrast_alpha")
beta = self._finite_float(beta, "contrast_beta")
return cv2.convertScaleAbs(bgr, alpha=alpha, beta=beta)
def apply_preview_contrast(self, bgr: np.ndarray, alpha: float = 1.08, beta: float = 0.0) -> np.ndarray:
"""
Ajuste leve de contraste/brilho para preview.
"""
out = cv2.convertScaleAbs(bgr, alpha=alpha, beta=beta)
return out
@staticmethod
def _apply_clahe_luminance(
bgr: np.ndarray,
clip_limit: float,
tile_grid_size: int,
) -> np.ndarray:
lab = cv2.cvtColor(bgr, cv2.COLOR_BGR2LAB)
luma, a_ch, b_ch = cv2.split(lab)
clahe = cv2.createCLAHE(
clipLimit=float(clip_limit),
tileGridSize=(int(tile_grid_size), int(tile_grid_size)),
)
return cv2.cvtColor(
cv2.merge([clahe.apply(luma), a_ch, b_ch]),
cv2.COLOR_LAB2BGR,
)
@staticmethod
def _apply_saturation(bgr: np.ndarray, factor: float) -> np.ndarray:
hsv = cv2.cvtColor(bgr, cv2.COLOR_BGR2HSV)
sat = hsv[:, :, 1].astype(np.float32) * float(factor)
hsv[:, :, 1] = np.clip(sat, 0, 255).astype(np.uint8)
return cv2.cvtColor(hsv, cv2.COLOR_HSV2BGR)
@staticmethod
def _apply_unsharp(
bgr: np.ndarray,
sigma: float,
amount: float,
threshold: float,
) -> np.ndarray:
image = bgr.astype(np.float32)
blurred = cv2.GaussianBlur(
image,
(0, 0),
sigmaX=float(sigma),
sigmaY=float(sigma),
)
sharpened = image + float(amount) * (image - blurred)
if threshold > 0.0:
mask = np.max(np.abs(image - blurred), axis=2) >= float(threshold)
result = image.copy()
result[mask] = sharpened[mask]
else:
result = sharpened
return np.clip(result, 0, 255).astype(np.uint8)
def raw16_to_preview_bgr(
self,
raw16: np.ndarray,
gamma: float = 2.2,
wb_strength: float = 0.8,
gamma: Optional[float] = None,
wb_strength: Optional[float] = None,
apply_wb: bool = True,
apply_contrast: bool = True,
bit_depth: int = 10,
) -> np.ndarray:
"""
Pipeline de preview bonito:
1. auto-level + gamma no mosaico
2. demosaic
3. white balance simples
4. leve contraste final
"""
vis8 = self.raw16_to_vis8(raw16, gamma=gamma, bit_depth=bit_depth)
bit_depth: Optional[int] = None,
*,
black_level: Optional[float] = None,
white_level: Optional[float] = None,
black_percentile: Optional[float] = None,
white_percentile: Optional[float] = None,
apply_clahe: Optional[bool] = None,
apply_saturation: Optional[bool] = None,
apply_sharpen: Optional[bool] = None,
return_report: bool = False,
):
"""Gera preview BGR bonito preservando exatamente o raster de entrada."""
source = np.asarray(raw16)
expected_shape = (self.sensor_height, self.sensor_width)
if source.shape != expected_shape:
raise ValueError(f"RAW16 shape={source.shape}; esperado={expected_shape}")
effective_gamma = float(
self.config["gamma"] if gamma is None else gamma
)
effective_bit_depth = int(
self.config["bit_depth"] if bit_depth is None else bit_depth
)
effective_wb_strength = float(
self.config["white_balance"]["strength"]
if wb_strength is None
else wb_strength
)
vis8 = self.raw16_to_vis8(
source,
black_level=black_level,
white_level=white_level,
gamma=effective_gamma,
bit_depth=effective_bit_depth,
black_percentile=black_percentile,
white_percentile=white_percentile,
)
level_report = deepcopy(self.last_preview_report.get("levels", {}))
bgr = cv2.cvtColor(vis8, self._debayer_code())
wb_report = {"enabled": False}
if apply_wb:
bgr = self.apply_preview_white_balance(bgr, strength=wb_strength)
bgr, wb_report = self.apply_preview_white_balance(
bgr,
strength=effective_wb_strength,
return_report=True,
)
wb_report["enabled"] = True
clahe_cfg = self.config["clahe"]
clahe_enabled = bool(clahe_cfg["enabled"] if apply_clahe is None else apply_clahe)
if clahe_enabled:
bgr = self._apply_clahe_luminance(
bgr,
float(clahe_cfg["clip_limit"]),
int(clahe_cfg["tile_grid_size"]),
)
saturation_cfg = self.config["saturation"]
saturation_enabled = bool(
saturation_cfg["enabled"]
if apply_saturation is None
else apply_saturation
)
if saturation_enabled:
bgr = self._apply_saturation(bgr, float(saturation_cfg["factor"]))
contrast_cfg = self.config["contrast"]
if apply_contrast:
bgr = self.apply_preview_contrast(bgr, alpha=1.08, beta=0.0)
bgr = self.apply_preview_contrast(
bgr,
alpha=float(contrast_cfg["alpha"]),
beta=float(contrast_cfg["beta"]),
)
return bgr
sharpen_cfg = self.config["sharpen"]
sharpen_enabled = bool(
sharpen_cfg["enabled"] if apply_sharpen is None else apply_sharpen
)
if sharpen_enabled:
bgr = self._apply_unsharp(
bgr,
sigma=float(sharpen_cfg["sigma"]),
amount=float(sharpen_cfg["amount"]),
threshold=float(sharpen_cfg["threshold"]),
)
def raw16_to_preview_jpg_bytes(self, raw16: np.ndarray, jpeg_quality: int = 95) -> bytes:
if bgr.shape != (self.sensor_height, self.sensor_width, 3):
raise RuntimeError(
f"Preview alterou geometria: {bgr.shape}; esperado="
f"{(self.sensor_height, self.sensor_width, 3)}"
)
if bgr.dtype != np.uint8:
raise RuntimeError(f"Preview dtype={bgr.dtype}; esperado=uint8")
self.last_preview_report = {
**self.get_preview_contract(),
"effective": {
"bit_depth": effective_bit_depth,
"gamma": effective_gamma,
"levels": level_report,
"white_balance": wb_report,
"clahe_enabled": clahe_enabled,
"saturation_enabled": saturation_enabled,
"contrast_enabled": bool(apply_contrast),
"sharpen_enabled": sharpen_enabled,
},
}
result = np.ascontiguousarray(bgr)
return (result, self.get_last_preview_report()) if return_report else result
def unpack_raw10_packed(
self,
packed_frame,
sensor_width: Optional[int] = None,
sensor_height: Optional[int] = None,
stride: Optional[int] = None,
) -> np.ndarray:
"""
Desempacota RAW10 MIPI (4 pixels/5 bytes), aceitando ndarray 1D/2D,
bytes e stride/padding por linha.
"""
width = self.sensor_width if sensor_width is None else int(sensor_width)
height = self.sensor_height if sensor_height is None else int(sensor_height)
if width != self.sensor_width or height != self.sensor_height:
raise ValueError(
"RawProcessorPreview preserva o raster configurado; dimensão "
f"solicitada={width}x{height}, configurada="
f"{self.sensor_width}x{self.sensor_height}."
)
if width % 4 != 0:
raise ValueError(f"RAW10 exige largura múltipla de 4: {width}")
if isinstance(packed_frame, (bytes, bytearray, memoryview)):
packed = np.frombuffer(packed_frame, dtype=np.uint8)
else:
packed = np.asarray(packed_frame)
if packed.ndim == 3 and packed.shape[2] == 1:
packed = packed[:, :, 0]
useful_width = (width // 4) * 5
if packed.dtype != np.uint8:
raise TypeError(
f"RAW10 packed deve ser uint8; shape={packed.shape}, dtype={packed.dtype}"
)
if packed.ndim == 1:
if stride is None:
if packed.size % height != 0:
raise ValueError(
f"RAW10 1D possui {packed.size} bytes, não divisível por "
f"height={height}; informe stride."
)
stride = packed.size // height
stride = int(stride)
expected_bytes = stride * height
if stride < useful_width or packed.size < expected_bytes:
raise ValueError(
f"RAW10 1D curto/inválido: bytes={packed.size}, stride={stride}, "
f"esperado>={expected_bytes}, payload/linha={useful_width}."
)
packed = packed[:expected_bytes].reshape(height, stride)
elif packed.ndim == 2:
if stride is not None and int(stride) != packed.shape[1]:
raise ValueError(
f"stride={stride} diverge da largura do array={packed.shape[1]}."
)
else:
raise TypeError(
f"RAW10 packed deve ser 1D ou 2D; shape={packed.shape}."
)
if packed.shape[0] != height or packed.shape[1] < useful_width:
raise ValueError(
f"RAW10 packed shape={packed.shape}; esperado >=({height}, {useful_width})"
)
padding = int(packed.shape[1] - useful_width)
if padding > max(4096, useful_width):
raise ValueError(
f"Padding RAW10 implausível: {padding} bytes/linha; "
"verifique width/height/stride."
)
payload = np.ascontiguousarray(packed[:, :useful_width])
groups = payload.reshape(height, width // 4, 5)
high = groups[:, :, :4].astype(np.uint16)
low = groups[:, :, 4].astype(np.uint16)
out = np.empty((height, width // 4, 4), dtype=np.uint16)
out[:, :, 0] = (high[:, :, 0] << 2) | (low & 0x03)
out[:, :, 1] = (high[:, :, 1] << 2) | ((low >> 2) & 0x03)
out[:, :, 2] = (high[:, :, 2] << 2) | ((low >> 4) & 0x03)
out[:, :, 3] = (high[:, :, 3] << 2) | ((low >> 6) & 0x03)
return np.ascontiguousarray(out.reshape(height, width))
def packed_raw10_to_preview_bgr(self, packed_frame: np.ndarray, **kwargs):
"""Atalho oficial RAW10 packed -> preview BGR."""
raw16 = self.unpack_raw10_packed(packed_frame)
return self.raw16_to_preview_bgr(raw16, **kwargs)
def raw16_to_preview_jpg_bytes(
self,
raw16: np.ndarray,
jpeg_quality: int = 95,
) -> bytes:
quality = int(np.clip(int(jpeg_quality), 1, 100))
bgr = self.raw16_to_preview_bgr(raw16)
ok, enc = cv2.imencode(".jpg", bgr, [int(cv2.IMWRITE_JPEG_QUALITY), int(jpeg_quality)])
ok, encoded = cv2.imencode(
".jpg",
bgr,
[int(cv2.IMWRITE_JPEG_QUALITY), quality],
)
if not ok:
raise RuntimeError("Falha ao codificar preview JPG")
return enc.tobytes()
return encoded.tobytes()
def raw16_to_preview_png_bytes(
self,
raw16: np.ndarray,
compression: int = 3,
) -> bytes:
level = int(np.clip(int(compression), 0, 9))
bgr = self.raw16_to_preview_bgr(raw16)
ok, encoded = cv2.imencode(
".png",
bgr,
[int(cv2.IMWRITE_PNG_COMPRESSION), level],
)
if not ok:
raise RuntimeError("Falha ao codificar preview PNG")
return encoded.tobytes()