From 21decdbbe9a235456e47ff07cf16fd331a546dc6 Mon Sep 17 00:00:00 2001 From: Diego Freitas Date: Mon, 24 Aug 2026 14:16:32 -0300 Subject: [PATCH] ajustado focus_calibration_tool para aceitar as cameras ov9782 e ar0234 --- .../oak-fcc-3/utils/focus_calibration_tool.py | 1738 ++++++++++++----- 1 file changed, 1258 insertions(+), 480 deletions(-) diff --git a/Python/OAK/datasets/oak-fcc-3/utils/focus_calibration_tool.py b/Python/OAK/datasets/oak-fcc-3/utils/focus_calibration_tool.py index 61fe7fe71..4a3a564a0 100644 --- a/Python/OAK/datasets/oak-fcc-3/utils/focus_calibration_tool.py +++ b/Python/OAK/datasets/oak-fcc-3/utils/focus_calibration_tool.py @@ -1,24 +1,94 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +""" +Focus Calibration Tool - OAK-FFC-3P - Sensor Aware +==================================================== + +Objetivo +-------- +Ferramenta standalone para: + - detectar automaticamente os sensores conectados; + - suportar CAM_A com OV9782 (1280x800) OU AR0234 (1920x1200); + - visualizar RGB + RE + NIR simultaneamente; + - medir foco manual por Laplacian / Tenengrad / Brenner; + - permitir ROI por câmera; + - salvar melhores scores e snapshots em JSON/PNG. + +Importante +---------- +Este script NÃO usa o OakFcc3Client e NÃO depende do pipeline RAW_BRUTO. +Para foco óptico ele usa: + - saída ISP da câmera colorida (RGB); + - saída nativa das câmeras mono (RE/NIR). + +Assim, Bayer pattern e normalização do dataset não interferem no teste de foco. + +Configuração padrão esperada: + CAM_A = RGB -> OV9782 ou AR0234 + CAM_B = RE -> OV9282 + CAM_C = NIR -> OV9282 + +Exemplos +-------- +# Detecta e abre tudo que estiver disponível: +python focus_calibration_sensor_aware.py + +# Primeiro teste somente da RGB nova: +python focus_calibration_sensor_aware.py --mode rgb + +# Exige as três câmeras: +python focus_calibration_sensor_aware.py --mode triple + +# Se RE e NIR estiverem fisicamente invertidas: +python focus_calibration_sensor_aware.py --re-socket CAM_C --nir-socket CAM_B + +Teclas +------ +1 = RGB +2 = RE +3 = NIR +M = troca métrica +D = informa sentido atual da lente +E = equalização ON/OFF +L = trava/destrava ROI +C = centraliza ROI +R = reseta score da câmera/métrica ativa +S = salva snapshot PNG + registro +SPACE = salva JSON completo +Q / ESC = sair +""" + import os import json import time import argparse +from dataclasses import dataclass, asdict from datetime import datetime from collections import deque +from typing import Dict, Optional, Tuple, List import cv2 import numpy as np - -from core.oak_fcc3_client import OakFcc3Client as MultiSpectralClient +import depthai as dai # ============================================================ # Helpers gerais # ============================================================ +METHODS = ("laplacian", "tenengrad", "brenner") +ROLES = ("rgb", "re", "nir") + + def now_str() -> str: return datetime.now().strftime("%Y-%m-%d %H:%M:%S") +def now_file_str() -> str: + return datetime.now().strftime("%Y%m%d_%H%M%S") + + def ensure_dir(path: str): if path: os.makedirs(path, exist_ok=True) @@ -36,170 +106,495 @@ def overlay_hud( ): yy = y h, _ = img_bgr.shape[:2] + for s in lines: if yy > h - 8: break + + text = str(s) + if shadow: - cv2.putText(img_bgr, str(s), (x, yy), cv2.FONT_HERSHEY_SIMPLEX, - font_scale, (0, 0, 0), 3, cv2.LINE_AA) - cv2.putText(img_bgr, str(s), (x, yy), cv2.FONT_HERSHEY_SIMPLEX, - font_scale, color, 1, cv2.LINE_AA) + cv2.putText( + img_bgr, + text, + (x, yy), + cv2.FONT_HERSHEY_SIMPLEX, + font_scale, + (0, 0, 0), + 3, + cv2.LINE_AA, + ) + + cv2.putText( + img_bgr, + text, + (x, yy), + cv2.FONT_HERSHEY_SIMPLEX, + font_scale, + color, + 1, + cv2.LINE_AA, + ) + yy += line_step -def to_bgr_u8_from_rgb01(rgb01: np.ndarray) -> np.ndarray: - rgb_u8 = np.clip(rgb01 * 255.0, 0, 255).astype(np.uint8) - return cv2.cvtColor(rgb_u8, cv2.COLOR_RGB2BGR) - - -def gray_to_bgr_u8(gray01: np.ndarray) -> np.ndarray: - g = np.clip(gray01 * 255.0, 0, 255).astype(np.uint8) - return cv2.cvtColor(g, cv2.COLOR_GRAY2BGR) - - -def gray_to_color_bgr(gray01: np.ndarray, color_name: str) -> np.ndarray: - g = np.clip(gray01 * 255.0, 0, 255).astype(np.uint8) - z = np.zeros_like(g, dtype=np.uint8) - - color_name = str(color_name).upper() - if color_name == "RE": - rgb = np.stack([g, z, z], axis=2) - elif color_name == "NIR": - rgb = np.stack([z, g, g], axis=2) - else: - rgb = np.stack([g, g, g], axis=2) - - return cv2.cvtColor(rgb, cv2.COLOR_RGB2BGR) - - -def resize_if_needed(img: np.ndarray, target_hw: tuple[int, int]) -> np.ndarray: - if img is None: - return None - target_h, target_w = target_hw - if img.shape[:2] == (target_h, target_w): - return img - return cv2.resize(img, (target_w, target_h), interpolation=cv2.INTER_LINEAR) - - -def build_empty_panel(shape_hw: tuple[int, int], title: str) -> np.ndarray: +def build_empty_panel(shape_hw: Tuple[int, int], title: str, text="sem frame disponivel"): h, w = shape_hw img = np.zeros((h, w, 3), dtype=np.uint8) - overlay_hud(img, [title, "sem frame disponivel"], x=18, y=44, font_scale=0.8, line_step=32) + overlay_hud( + img, + [title, text], + x=18, + y=44, + font_scale=0.75, + line_step=32, + ) return img -def get_decoded_by_role(decoded: dict, role: str): - role = str(role).lower() - for cam_id, item in decoded.items(): - if str(item.get("role", "")).lower() == role: - return cam_id, item - return None, None +def socket_name(socket) -> str: + name = getattr(socket, "name", None) + if name: + return str(name) + + s = str(socket) + for candidate in ("CAM_A", "CAM_B", "CAM_C", "CAM_D"): + if candidate in s: + return candidate + + return s -def get_image_by_role(decoded: dict, role: str): - cam_id, item = get_decoded_by_role(decoded, role) - if item is None: - return cam_id, None - return cam_id, item.get("image") +def get_socket(name: str): + name = str(name).upper().strip() + mapping = { + "CAM_A": dai.CameraBoardSocket.CAM_A, + "CAM_B": dai.CameraBoardSocket.CAM_B, + "CAM_C": dai.CameraBoardSocket.CAM_C, + } + + if hasattr(dai.CameraBoardSocket, "CAM_D"): + mapping["CAM_D"] = dai.CameraBoardSocket.CAM_D + + if name not in mapping: + raise ValueError(f"Socket inválido: {name}. Opções: {sorted(mapping)}") + + return mapping[name] -def get_preview_panel_by_role(previews: dict, meta: dict, role: str): - camera_info = (meta or {}).get("camera_info", {}) or {} - for cam_id, preview in (previews or {}).items(): - info = camera_info.get(cam_id, {}) or {} - if str(info.get("role", "")).lower() == role: - return preview - return None +def supported_type_strings(feature) -> List[str]: + values = getattr(feature, "supportedTypes", []) or [] + return [str(v).upper() for v in values] -def validate_module_ready(status: dict, frame_type: str, raw_policy: str): - if not status.get("ok", True): - raise RuntimeError(f"Status inválido retornado pelo módulo: {status}") +def feature_is_color(feature) -> bool: + types = supported_type_strings(feature) + sensor = str(getattr(feature, "sensorName", "") or "").upper() - active_roles = status.get("active_roles", {}) or {} - active_count = int(status.get("camera_count_active", 0)) + if any("COLOR" in t for t in types): + return True - if frame_type == "RAW_BRUTO": - if raw_policy == "require_triple": - missing = [role for role in ("rgb", "nir", "re") if role not in active_roles] - if missing: - raise RuntimeError( - "RAW_BRUTO com require_triple exige rgb/nir/re ativas. " - f"Faltando: {missing}. Ativas: {active_roles}" - ) - elif active_count < 1: - raise RuntimeError("RAW_BRUTO requer ao menos uma câmera ativa.") - return + if any("MONO" in t for t in types): + return False - raise RuntimeError(f"frame_type desconhecido para validação: {frame_type}") + # Fallback conhecido do nosso módulo. + return sensor in {"OV9782", "AR0234"} -def normalize_gray01(img01: np.ndarray) -> np.ndarray: - """ - Converte RGB/mono float 0..1 para mono float 0..1. - Para foco, o objetivo é medir borda, então usamos luminância no RGB. - """ - if img01 is None: +def feature_is_mono(feature) -> bool: + types = supported_type_strings(feature) + + if any("MONO" in t for t in types): + return True + + if any("COLOR" in t for t in types): + return False + + return not feature_is_color(feature) + + +def bgr_to_gray01(img_bgr: np.ndarray) -> Optional[np.ndarray]: + if img_bgr is None: return None - arr = img01.astype(np.float32) - - if arr.ndim == 3: - # img01 vem em RGB, não BGR. - r = arr[:, :, 0] - g = arr[:, :, 1] - b = arr[:, :, 2] - gray = 0.299 * r + 0.587 * g + 0.114 * b + if img_bgr.ndim == 2: + gray = img_bgr + elif img_bgr.ndim == 3 and img_bgr.shape[2] == 1: + gray = img_bgr[:, :, 0] else: - gray = arr + gray = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2GRAY) - gray = np.nan_to_num(gray, nan=0.0, posinf=1.0, neginf=0.0) + gray = gray.astype(np.float32) / 255.0 return np.clip(gray, 0.0, 1.0) -def crop_rect(img: np.ndarray, rect): - if img is None: +def resize_panel(img: np.ndarray, target_hw: Tuple[int, int]) -> np.ndarray: + th, tw = target_hw + return cv2.resize(img, (tw, th), interpolation=cv2.INTER_AREA) + + +def colorize_mono_for_view(img_bgr: np.ndarray, role: str) -> np.ndarray: + if img_bgr is None: return None - h, w = img.shape[:2] - x0, y0, x1, y1 = rect - x0, x1 = sorted((int(x0), int(x1))) - y0, y1 = sorted((int(y0), int(y1))) - x0 = max(0, min(w - 1, x0)) - x1 = max(0, min(w, x1)) - y0 = max(0, min(h - 1, y0)) - y1 = max(0, min(h, y1)) + if img_bgr.ndim == 3: + gray = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2GRAY) + else: + gray = img_bgr - if x1 <= x0 or y1 <= y0: - return None + z = np.zeros_like(gray) - return img[y0:y1, x0:x1] + role = str(role).lower() + + # Apenas visual. As métricas são calculadas no frame original. + if role == "re": + # Vermelho no BGR. + return np.dstack([z, z, gray]) + + if role == "nir": + # Ciano no BGR. + return np.dstack([gray, gray, z]) + + return cv2.cvtColor(gray, cv2.COLOR_GRAY2BGR) def default_roi_for_shape(shape_hw, frac=0.42): h, w = shape_hw - rw = int(w * frac) - rh = int(h * frac) + rw = max(8, int(w * frac)) + rh = max(8, int(h * frac)) x0 = (w - rw) // 2 y0 = (h - rh) // 2 return (x0, y0, x0 + rw, y0 + rh) +def sanitize_roi(rect, shape_hw): + if rect is None: + return None + + h, w = shape_hw + x0, y0, x1, y1 = rect + + x0, x1 = sorted((int(x0), int(x1))) + y0, y1 = sorted((int(y0), int(y1))) + + x0 = max(0, min(w - 1, x0)) + x1 = max(1, min(w, x1)) + y0 = max(0, min(h - 1, y0)) + y1 = max(1, min(h, y1)) + + if x1 - x0 < 4 or y1 - y0 < 4: + return None + + return (x0, y0, x1, y1) + + +def crop_rect(img: np.ndarray, rect): + if img is None or rect is None: + return None + + rect = sanitize_roi(rect, img.shape[:2]) + if rect is None: + return None + + x0, y0, x1, y1 = rect + return img[y0:y1, x0:x1] + + +# ============================================================ +# Descoberta / resolução sensor-aware +# ============================================================ + +@dataclass +class SensorSpec: + role: str + socket_name: str + sensor_name: str + feature_width: int + feature_height: int + is_color: bool + configured_width: int + configured_height: int + resolution_name: str + stream_name: str + source: str + + +def discover_cameras(mx_id: Optional[str]): + device_info = dai.DeviceInfo(mx_id) if mx_id else None + + if device_info is not None: + ctx = dai.Device(device_info) + else: + ctx = dai.Device() + + with ctx as device: + features = list(device.getConnectedCameraFeatures()) + + actual_mx = None + for attr in ("getMxId", "getDeviceId"): + if hasattr(device, attr): + try: + actual_mx = str(getattr(device, attr)()) + if actual_mx: + break + except Exception: + pass + + usb_speed = None + try: + usb_speed = str(device.getUsbSpeed()) + except Exception: + pass + + rows = [] + for f in features: + rows.append({ + "socket_obj": f.socket, + "socket_name": socket_name(f.socket), + "sensor_name": str(getattr(f, "sensorName", "") or ""), + "width": int(getattr(f, "width", 0) or 0), + "height": int(getattr(f, "height", 0) or 0), + "supported_types": supported_type_strings(f), + "is_color": feature_is_color(f), + "is_mono": feature_is_mono(f), + "has_autofocus_ic": int(getattr(f, "hasAutofocusIC", 0) or 0), + }) + + return rows, actual_mx, usb_speed + + +def enum_if_exists(enum_cls, name: str): + return getattr(enum_cls, name, None) + + +def pick_color_resolution(sensor_name: str, width: int, height: int): + """ + Retorna: + (enum_resolution, resolution_name, configured_width, configured_height) + """ + sensor = str(sensor_name or "").upper() + enum_cls = dai.ColorCameraProperties.SensorResolution + + if "AR0234" in sensor: + enum_value = enum_if_exists(enum_cls, "THE_1200_P") + if enum_value is None: + raise RuntimeError( + "Seu depthai não possui ColorCameraProperties.SensorResolution.THE_1200_P. " + "Atualize a biblioteca DepthAI antes de testar a AR0234." + ) + return enum_value, "THE_1200_P", 1920, 1200 + + if "OV9782" in sensor: + enum_value = enum_if_exists(enum_cls, "THE_800_P") + if enum_value is None: + raise RuntimeError("DepthAI sem THE_800_P para ColorCamera.") + return enum_value, "THE_800_P", 1280, 800 + + # Fallback por resolução anunciada pelo próprio sensor. + candidates = [ + ((1920, 1200), "THE_1200_P"), + ((1280, 800), "THE_800_P"), + ((1920, 1080), "THE_1080_P"), + ((1280, 720), "THE_720_P"), + ((3840, 2160), "THE_4_K"), + ] + + for (w, h), enum_name in candidates: + if (width, height) == (w, h): + enum_value = enum_if_exists(enum_cls, enum_name) + if enum_value is not None: + return enum_value, enum_name, w, h + + raise RuntimeError( + f"Sensor colorido não mapeado: {sensor_name} ({width}x{height}). " + "Adicione o modo em pick_color_resolution()." + ) + + +def pick_mono_resolution(sensor_name: str, width: int, height: int): + sensor = str(sensor_name or "").upper() + enum_cls = dai.MonoCameraProperties.SensorResolution + + if "OV9282" in sensor or (width, height) == (1280, 800): + enum_value = enum_if_exists(enum_cls, "THE_800_P") + if enum_value is None: + raise RuntimeError("DepthAI sem THE_800_P para MonoCamera.") + return enum_value, "THE_800_P", 1280, 800 + + candidates = [ + ((1280, 800), "THE_800_P"), + ((1280, 720), "THE_720_P"), + ((640, 480), "THE_480_P"), + ((640, 400), "THE_400_P"), + ] + + for (w, h), enum_name in candidates: + if (width, height) == (w, h): + enum_value = enum_if_exists(enum_cls, enum_name) + if enum_value is not None: + return enum_value, enum_name, w, h + + raise RuntimeError( + f"Sensor mono não mapeado: {sensor_name} ({width}x{height}). " + "Adicione o modo em pick_mono_resolution()." + ) + + +def build_specs(camera_rows, args) -> Dict[str, SensorSpec]: + by_socket = {row["socket_name"]: row for row in camera_rows} + + role_socket = { + "rgb": args.rgb_socket.upper(), + "re": args.re_socket.upper(), + "nir": args.nir_socket.upper(), + } + + if args.mode == "rgb": + requested_roles = ["rgb"] + else: + requested_roles = ["rgb", "re", "nir"] + + specs = {} + + for role in requested_roles: + sock_name = role_socket[role] + row = by_socket.get(sock_name) + + if row is None: + if args.mode == "triple": + raise RuntimeError( + f"Modo triple exige {role.upper()} em {sock_name}, " + f"mas esse socket não foi detectado." + ) + continue + + if role == "rgb": + if not row["is_color"]: + raise RuntimeError( + f"{sock_name} foi escolhido como RGB, mas o sensor detectado " + f"({row['sensor_name']}) não foi anunciado como COLOR." + ) + + _, res_name, cw, ch = pick_color_resolution( + row["sensor_name"], row["width"], row["height"] + ) + + specs[role] = SensorSpec( + role=role, + socket_name=sock_name, + sensor_name=row["sensor_name"], + feature_width=row["width"], + feature_height=row["height"], + is_color=True, + configured_width=cw, + configured_height=ch, + resolution_name=res_name, + stream_name="focus_rgb", + source="ISP", + ) + + else: + if not row["is_mono"]: + raise RuntimeError( + f"{sock_name} foi escolhido como {role.upper()}, mas o sensor detectado " + f"({row['sensor_name']}) não foi anunciado como MONO." + ) + + _, res_name, cw, ch = pick_mono_resolution( + row["sensor_name"], row["width"], row["height"] + ) + + specs[role] = SensorSpec( + role=role, + socket_name=sock_name, + sensor_name=row["sensor_name"], + feature_width=row["width"], + feature_height=row["height"], + is_color=False, + configured_width=cw, + configured_height=ch, + resolution_name=res_name, + stream_name=f"focus_{role}", + source="MONO_OUT", + ) + + if "rgb" not in specs: + raise RuntimeError( + f"RGB não encontrada em {args.rgb_socket}. " + "Confira os flats e/ou use --rgb-socket." + ) + + if args.mode == "triple": + missing = [r for r in ROLES if r not in specs] + if missing: + raise RuntimeError(f"Modo triple: faltando roles {missing}") + + return specs + + +def build_pipeline(specs: Dict[str, SensorSpec], fps: float): + pipeline = dai.Pipeline() + + for role, spec in specs.items(): + socket = get_socket(spec.socket_name) + + if spec.is_color: + cam = pipeline.createColorCamera() + cam.setBoardSocket(socket) + + enum_value, _, _, _ = pick_color_resolution( + spec.sensor_name, + spec.feature_width, + spec.feature_height, + ) + + cam.setResolution(enum_value) + cam.setFps(float(fps)) + cam.setInterleaved(False) + + try: + cam.setColorOrder(dai.ColorCameraProperties.ColorOrder.BGR) + except Exception: + pass + + xout = pipeline.createXLinkOut() + xout.setStreamName(spec.stream_name) + + # ISP mantém a resolução útil do sensor e getCvFrame() entrega + # uma imagem pronta para inspeção, sem depender do nosso Bayer. + cam.isp.link(xout.input) + + else: + cam = pipeline.createMonoCamera() + cam.setBoardSocket(socket) + + enum_value, _, _, _ = pick_mono_resolution( + spec.sensor_name, + spec.feature_width, + spec.feature_height, + ) + + cam.setResolution(enum_value) + cam.setFps(float(fps)) + + xout = pipeline.createXLinkOut() + xout.setStreamName(spec.stream_name) + cam.out.link(xout.input) + + return pipeline + + # ============================================================ # Métricas de foco # ============================================================ -def preprocess_focus_gray(gray01: np.ndarray, equalize=False, blur_ksize=0) -> np.ndarray: +def preprocess_focus_gray(gray01: np.ndarray, equalize=False): g = np.clip(gray01 * 255.0, 0, 255).astype(np.uint8) if equalize: g = cv2.equalizeHist(g) - if blur_ksize and blur_ksize >= 3: - if blur_ksize % 2 == 0: - blur_ksize += 1 - g = cv2.GaussianBlur(g, (blur_ksize, blur_ksize), 0) - return g @@ -211,20 +606,21 @@ def focus_laplacian_var(gray_u8: np.ndarray) -> float: def focus_tenengrad(gray_u8: np.ndarray) -> float: sx = cv2.Sobel(gray_u8, cv2.CV_64F, 1, 0, ksize=3) sy = cv2.Sobel(gray_u8, cv2.CV_64F, 0, 1, ksize=3) - mag2 = sx * sx + sy * sy - return float(np.mean(mag2)) + return float(np.mean(sx * sx + sy * sy)) def focus_brenner(gray_u8: np.ndarray) -> float: arr = gray_u8.astype(np.float32) + if arr.shape[1] < 3: return 0.0 + diff = arr[:, 2:] - arr[:, :-2] return float(np.mean(diff * diff)) -def compute_focus_metrics(img01: np.ndarray, roi_rect, equalize=False) -> dict: - gray01 = normalize_gray01(img01) +def compute_focus_metrics(img_bgr: np.ndarray, roi_rect, equalize=False): + gray01 = bgr_to_gray01(img_bgr) roi = crop_rect(gray01, roi_rect) if roi is None or roi.size < 64: @@ -242,8 +638,8 @@ def compute_focus_metrics(img01: np.ndarray, roi_rect, equalize=False) -> dict: } gray_u8 = preprocess_focus_gray(roi, equalize=equalize) - arr = roi.astype(np.float32).reshape(-1) + return { "valid": True, "laplacian": focus_laplacian_var(gray_u8), @@ -262,14 +658,15 @@ def metric_value(metrics: dict, method: str) -> float: return float(metrics.get(method, 0.0) or 0.0) -def smooth_score(history, window: int) -> float: +def smooth_from_history(history, window: int) -> float: if not history: return 0.0 + vals = [float(x["score"]) for x in list(history)[-max(1, window):]] return float(np.mean(vals)) -def analyze_trend(history, best_score, direction_name: str, drop_warn_pct=3.0) -> dict: +def analyze_trend(history, best_score, direction_name: str, drop_warn_pct=3.0): if len(history) < 6: return { "status": "coletando", @@ -279,7 +676,11 @@ def analyze_trend(history, best_score, direction_name: str, drop_warn_pct=3.0) - } recent = [float(x["smooth"]) for x in list(history)[-5:]] - old = [float(x["smooth"]) for x in list(history)[-12:-7]] if len(history) >= 12 else [float(x["smooth"]) for x in list(history)[:5]] + + if len(history) >= 12: + old = [float(x["smooth"]) for x in list(history)[-12:-7]] + else: + old = [float(x["smooth"]) for x in list(history)[:5]] recent_mean = float(np.mean(recent)) old_mean = float(np.mean(old)) @@ -291,12 +692,11 @@ def analyze_trend(history, best_score, direction_name: str, drop_warn_pct=3.0) - if best_score > 0 and drop_from_best >= drop_warn_pct: return { "status": "passou_do_pico", - "instruction": f"volte um pouco no sentido contrario de {direction_name}", + "instruction": f"volte um pouco: contrario de {direction_name}", "delta": delta, "pct_of_best": pct_of_best, } - # Faixa morta para evitar feedback nervoso. eps = max(best_score * 0.002, 1e-6) if delta > eps: @@ -310,14 +710,14 @@ def analyze_trend(history, best_score, direction_name: str, drop_warn_pct=3.0) - if delta < -eps: return { "status": "piorando", - "instruction": f"inverta o sentido: contrario de {direction_name}", + "instruction": f"inverta: contrario de {direction_name}", "delta": delta, "pct_of_best": pct_of_best, } return { "status": "estavel", - "instruction": "ajuste bem fino ou trave a lente", + "instruction": "ajuste fino ou trave a lente", "delta": delta, "pct_of_best": pct_of_best, } @@ -327,52 +727,121 @@ def analyze_trend(history, best_score, direction_name: str, drop_warn_pct=3.0) - # Desenho # ============================================================ -def draw_roi(panel: np.ndarray, rect, active=False): - if rect is None: +def draw_roi(panel, roi_src, src_shape_hw, active=False): + if roi_src is None: return - x0, y0, x1, y1 = map(int, rect) + + ph, pw = panel.shape[:2] + sh, sw = src_shape_hw + + x0, y0, x1, y1 = roi_src + + px0 = int(x0 * pw / max(1, sw)) + px1 = int(x1 * pw / max(1, sw)) + py0 = int(y0 * ph / max(1, sh)) + py1 = int(y1 * ph / max(1, sh)) + color = (0, 255, 255) if active else (0, 180, 255) - cv2.rectangle(panel, (x0, y0), (x1, y1), color, 2) - cv2.putText(panel, "FOCUS ROI", (x0 + 6, max(20, y0 - 8)), - cv2.FONT_HERSHEY_SIMPLEX, 0.55, color, 2, cv2.LINE_AA) + + cv2.rectangle(panel, (px0, py0), (px1, py1), color, 2) + cv2.putText( + panel, + "FOCUS ROI", + (px0 + 6, max(20, py0 - 8)), + cv2.FONT_HERSHEY_SIMPLEX, + 0.55, + color, + 2, + cv2.LINE_AA, + ) -def draw_crosshair(panel: np.ndarray): +def draw_crosshair(panel): h, w = panel.shape[:2] - cv2.line(panel, (w // 2 - 18, h // 2), (w // 2 + 18, h // 2), (255, 255, 255), 1, cv2.LINE_AA) - cv2.line(panel, (w // 2, h // 2 - 18), (w // 2, h // 2 + 18), (255, 255, 255), 1, cv2.LINE_AA) + + cv2.line( + panel, + (w // 2 - 18, h // 2), + (w // 2 + 18, h // 2), + (255, 255, 255), + 1, + cv2.LINE_AA, + ) + + cv2.line( + panel, + (w // 2, h // 2 - 18), + (w // 2, h // 2 + 18), + (255, 255, 255), + 1, + cv2.LINE_AA, + ) -def draw_score_bar(panel: np.ndarray, pct: float, x: int, y: int, w: int, h: int, label: str): +def draw_panel_title(panel, title, selected=False): + color = (0, 255, 255) if selected else (255, 255, 255) + overlay_hud( + panel, + [title], + x=12, + y=24, + font_scale=0.60, + line_step=24, + color=color, + ) + + +def draw_score_bar(panel, pct, x, y, w, h, label): pct = float(max(0.0, min(100.0, pct))) + cv2.rectangle(panel, (x, y), (x + w, y + h), (80, 80, 80), 1) + fill_w = int((pct / 100.0) * w) cv2.rectangle(panel, (x, y), (x + fill_w, y + h), (230, 230, 230), -1) + cv2.rectangle(panel, (x, y), (x + w, y + h), (180, 180, 180), 1) - cv2.putText(panel, f"{label}: {pct:5.1f}%", (x, y - 8), - cv2.FONT_HERSHEY_SIMPLEX, 0.55, (255, 255, 255), 1, cv2.LINE_AA) + + cv2.putText( + panel, + f"{label}: {pct:5.1f}%", + (x, y - 8), + cv2.FONT_HERSHEY_SIMPLEX, + 0.50, + (255, 255, 255), + 1, + cv2.LINE_AA, + ) -def draw_history_graph(panel: np.ndarray, history, x: int, y: int, w: int, h: int, best_score: float): +def draw_history_graph(panel, history, x, y, w, h, best_score): cv2.rectangle(panel, (x, y), (x + w, y + h), (35, 35, 35), -1) cv2.rectangle(panel, (x, y), (x + w, y + h), (120, 120, 120), 1) if len(history) < 2: - cv2.putText(panel, "grafico aguardando historico...", (x + 10, y + h // 2), - cv2.FONT_HERSHEY_SIMPLEX, 0.55, (180, 180, 180), 1, cv2.LINE_AA) + cv2.putText( + panel, + "grafico aguardando historico...", + (x + 10, y + h // 2), + cv2.FONT_HERSHEY_SIMPLEX, + 0.50, + (180, 180, 180), + 1, + cv2.LINE_AA, + ) return vals = np.array([float(item["smooth"]) for item in history], dtype=np.float32) - vals = vals[-w:] # no máximo um ponto por pixel horizontal + vals = vals[-max(2, w):] max_val = max(float(np.max(vals)), float(best_score), 1e-6) min_val = min(float(np.min(vals)), max_val * 0.90) span = max(max_val - min_val, 1e-6) pts = [] - for i, v in enumerate(vals): + + for i, value in enumerate(vals): px = x + int((i / max(1, len(vals) - 1)) * (w - 1)) - py = y + h - 1 - int(((float(v) - min_val) / span) * (h - 1)) + py = y + h - 1 - int(((float(value) - min_val) / span) * (h - 1)) pts.append((px, py)) for p0, p1 in zip(pts[:-1], pts[1:]): @@ -381,113 +850,190 @@ def draw_history_graph(panel: np.ndarray, history, x: int, y: int, w: int, h: in if best_score > 0: by = y + h - 1 - int(((best_score - min_val) / span) * (h - 1)) by = max(y, min(y + h - 1, by)) + cv2.line(panel, (x, by), (x + w, by), (0, 255, 255), 1, cv2.LINE_AA) - cv2.putText(panel, "best", (x + 6, max(y + 16, by - 4)), - cv2.FONT_HERSHEY_SIMPLEX, 0.45, (0, 255, 255), 1, cv2.LINE_AA) + + cv2.putText( + panel, + "best", + (x + 6, max(y + 16, by - 4)), + cv2.FONT_HERSHEY_SIMPLEX, + 0.43, + (0, 255, 255), + 1, + cv2.LINE_AA, + ) def make_data_panel( shape_hw, + spec: SensorSpec, selected_role, method, metrics, score, smooth, - best_score, - best_pct, + best, trend, - fps_stream, + fps_by_role, fps_view, direction_name, history, - roi_rect, roi_locked, equalize, ): h, w = shape_hw panel = np.zeros((h, w, 3), dtype=np.uint8) + best_score = float(best.get("smooth", 0.0) or 0.0) score_pct = 0.0 if best_score <= 0 else (smooth / best_score) * 100.0 - score_pct = max(0.0, min(120.0, score_pct)) - - status = trend.get("status", "coletando") - instruction = trend.get("instruction", "gire devagar") lines = [ - "FOCUS CALIBRATION TOOL", - f"camera ativa: {selected_role.upper()} | metodo={method}", + "FOCUS CALIBRATION - SENSOR AWARE", + f"ativa={selected_role.upper()} | metodo={method}", + f"sensor={spec.sensor_name} | socket={spec.socket_name}", + f"modo={spec.resolution_name} | {spec.configured_width}x{spec.configured_height}", + f"fonte={spec.source}", f"score={score:.1f} | smooth={smooth:.1f}", f"best={best_score:.1f} | atual/best={score_pct:.1f}%", - f"status={status}", - f"acao: {instruction}", - f"sentido atual: {direction_name}", - f"fps_stream={fps_stream:.1f} | fps_view={fps_view:.1f}", - f"roi={'travada' if roi_locked else 'editavel'} | equalize={'ON' if equalize else 'OFF'}", + f"status={trend.get('status', 'coletando')}", + f"acao={trend.get('instruction', 'gire devagar')}", + f"sentido={direction_name}", + f"fps RGB/RE/NIR={fps_by_role.get('rgb', 0):.1f}/{fps_by_role.get('re', 0):.1f}/{fps_by_role.get('nir', 0):.1f}", + f"fps_view={fps_view:.1f}", + f"ROI={'travada' if roi_locked else 'editavel'} | equalize={'ON' if equalize else 'OFF'}", ] if metrics and metrics.get("valid"): lines.extend([ "-", f"mean={metrics['mean']:.3f} std={metrics['std']:.3f} p95={metrics['p95']:.3f}", - f"sat={metrics['pct_saturated']:.2f}% dark={metrics['pct_dark']:.2f}% pixels={metrics['pixels']}", + f"sat={metrics['pct_saturated']:.2f}% dark={metrics['pct_dark']:.2f}%", ]) - overlay_hud(panel, lines, x=14, y=28, font_scale=0.58, line_step=23) + overlay_hud( + panel, + lines, + x=14, + y=26, + font_scale=0.48, + line_step=20, + ) - bar_y = min(h - 170, 310) - draw_score_bar(panel, min(100.0, score_pct), 18, bar_y, max(80, w - 36), 24, "nitidez relativa") + bar_y = min(h - 165, 315) + bar_y = max(255, bar_y) - graph_y = bar_y + 52 - graph_h = max(90, h - graph_y - 78) - draw_history_graph(panel, history, 18, graph_y, max(100, w - 36), graph_h, best_score) + draw_score_bar( + panel, + min(100.0, score_pct), + 18, + bar_y, + max(80, w - 36), + 22, + "nitidez relativa", + ) + + graph_y = bar_y + 46 + graph_h = max(65, h - graph_y - 62) + + draw_history_graph( + panel, + history, + 18, + graph_y, + max(100, w - 36), + graph_h, + best_score, + ) help_lines = [ - "1=RGB | 2=RE | 3=NIR | M troca metrica | D troca sentido", - "mouse arrasta ROI | C centraliza ROI | L trava ROI | E equalize", - "R reset score | S snapshot JSON | SPACE salva resultado | Q sai", + "1 RGB | 2 RE | 3 NIR | M metrica | D sentido", + "mouse ROI | C centraliza | L trava | E equalize | R reset", + "S snapshot PNG | SPACE salva JSON | Q sai", ] - overlay_hud(panel, help_lines, x=14, y=h - 56, font_scale=0.48, line_step=18) + + overlay_hud( + panel, + help_lines, + x=14, + y=h - 48, + font_scale=0.40, + line_step=16, + ) return panel -def fit_panel(img, target_hw): - th, tw = target_hw - if img.shape[:2] == (th, tw): - return img - return cv2.resize(img, (tw, th), interpolation=cv2.INTER_NEAREST) - - -def draw_panel_title(panel, title, selected=False): - color = (0, 255, 255) if selected else (255, 255, 255) - overlay_hud(panel, [title], x=12, y=24, font_scale=0.65, line_step=24, color=color) - - # ============================================================ -# Persistência +# Estado / persistência # ============================================================ -def build_result_payload(args, results_by_role, snapshots): +def empty_best(): return { - "schema": "multispec_focus_calibration_v1", + "score": 0.0, + "smooth": 0.0, + "metrics": None, + "timestamp": None, + "roi": None, + } + + +def json_safe_best(best): + return { + "score": float(best.get("score", 0.0) or 0.0), + "smooth": float(best.get("smooth", 0.0) or 0.0), + "metrics": best.get("metrics"), + "timestamp": best.get("timestamp"), + "roi": best.get("roi"), + } + + +def build_result_payload( + args, + camera_rows, + specs, + best_by_role_method, + snapshots, + actual_mx, + usb_speed, +): + results = {} + + for role in ROLES: + results[role] = {} + for method in METHODS: + results[role][method] = json_safe_best(best_by_role_method[role][method]) + + return { + "schema": "oak_ffc_focus_calibration_sensor_aware_v2", "saved_at": now_str(), - "frame_type": "RAW_BRUTO", - "capture_mode_requested": args.capture_mode, - "raw_policy": args.raw_policy, - "sensor_width": args.width, - "sensor_height": args.height, - "bayer_pattern": args.bayer, - "focus_method_default": args.method, - "notes": args.notes or "", - "results_by_role": results_by_role, + "depthai_version": getattr(dai, "__version__", "unknown"), + "device_mx_id": actual_mx, + "usb_speed": usb_speed, + "mode": args.mode, + "fps_requested": args.fps, + "role_socket_map": { + "rgb": args.rgb_socket, + "re": args.re_socket, + "nir": args.nir_socket, + }, + "camera_inventory": [ + {k: v for k, v in row.items() if k != "socket_obj"} + for row in camera_rows + ], + "active_specs": { + role: asdict(spec) + for role, spec in specs.items() + }, + "results_by_role_and_method": results, "snapshots": snapshots, + "notes": args.notes or "", } def save_json(path, payload): ensure_dir(os.path.dirname(path) or ".") - payload = dict(payload) - payload["saved_at"] = now_str() + with open(path, "w", encoding="utf-8") as f: json.dump(payload, f, ensure_ascii=False, indent=2) @@ -498,397 +1044,629 @@ def save_json(path, payload): def main(): parser = argparse.ArgumentParser( - description="Ferramenta de auxílio para foco manual das câmeras RGB/RE/NIR do módulo multiespectral.", + description=( + "Focus Calibration Tool standalone para OAK-FFC-3P, " + "sensor-aware para OV9782 e AR0234." + ), formatter_class=argparse.ArgumentDefaultsHelpFormatter, ) - parser.add_argument("--fps", type=int, default=20) - parser.add_argument("--width", type=int, default=1280) - parser.add_argument("--height", type=int, default=800) - parser.add_argument("--bayer", default="BGGR", choices=["GBRG", "GRBG", "RGGB", "BGGR"]) - parser.add_argument("--capture_mode", default="AUTO", choices=["AUTO", "SINGLE", "DOUBLE", "TRIPLE"]) - parser.add_argument("--raw_policy", default="allow_single", choices=["allow_single", "require_triple"]) - parser.add_argument("--preview_scale", type=float, default=1.0) - parser.add_argument("--module_calibration_json", default="calibration/module_params.json") - parser.add_argument("--out_json", default="calibration/focus_calibration.json") - parser.add_argument("--method", default="laplacian", choices=["laplacian", "tenengrad", "brenner"]) + parser.add_argument( + "--mode", + default="auto", + choices=["auto", "rgb", "triple"], + help="auto abre o que existir; rgb abre só CAM_A/RGB; triple exige RGB+RE+NIR", + ) + + parser.add_argument("--fps", type=float, default=20.0) + parser.add_argument("--mx-id", default=None) + + parser.add_argument("--rgb-socket", default="CAM_A") + parser.add_argument("--re-socket", default="CAM_B") + parser.add_argument("--nir-socket", default="CAM_C") + + parser.add_argument( + "--panel-width", + type=int, + default=640, + help="Largura visual de cada quadrante. Métrica continua na resolução nativa.", + ) + parser.add_argument( + "--panel-height", + type=int, + default=400, + help="Altura visual de cada quadrante.", + ) + + parser.add_argument( + "--method", + default="laplacian", + choices=list(METHODS), + ) parser.add_argument("--history", type=int, default=260) - parser.add_argument("--smooth_window", type=int, default=5) - parser.add_argument("--drop_warn_pct", type=float, default=3.0) - parser.add_argument("--equalize", action="store_true", help="Equaliza histograma da ROI antes de medir foco") - parser.add_argument("--only_camera", default=None, choices=["CAM_A", "CAM_B", "CAM_C"]) + parser.add_argument("--smooth-window", type=int, default=5) + parser.add_argument("--drop-warn-pct", type=float, default=3.0) + parser.add_argument("--equalize", action="store_true") + + parser.add_argument( + "--out-json", + default="calibration/focus_calibration_sensor_aware.json", + ) + parser.add_argument( + "--snapshot-dir", + default="calibration/focus_snapshots", + ) parser.add_argument("--notes", default="") + args = parser.parse_args() + # -------------------------------------------------------- + # Descoberta + # -------------------------------------------------------- + + camera_rows, actual_mx, usb_speed = discover_cameras(args.mx_id) + + print("=" * 74) + print("FOCUS CALIBRATION TOOL - SENSOR AWARE") + print(f"DepthAI : {getattr(dai, '__version__', 'unknown')}") + print(f"MX ID : {actual_mx}") + print(f"USB : {usb_speed}") + print("-" * 74) + + for row in camera_rows: + print( + f"{row['socket_name']:5s} | " + f"{row['sensor_name']:12s} | " + f"{row['width']}x{row['height']} | " + f"types={row['supported_types']} | " + f"AF_IC={row['has_autofocus_ic']}" + ) + + specs = build_specs(camera_rows, args) + + print("-" * 74) + + for role, spec in specs.items(): + print( + f"[PIPE] {role.upper():3s} <- {spec.socket_name} " + f"{spec.sensor_name} | {spec.resolution_name} " + f"{spec.configured_width}x{spec.configured_height} | {spec.source}" + ) + + print("=" * 74) + + # -------------------------------------------------------- + # Pipeline + # -------------------------------------------------------- + + pipeline = build_pipeline(specs, args.fps) + + device_info = dai.DeviceInfo(actual_mx) if actual_mx else None + + if device_info is not None: + device_ctx = dai.Device(pipeline, device_info) + else: + device_ctx = dai.Device(pipeline) + + # -------------------------------------------------------- + # Estado da UI + # -------------------------------------------------------- + selected_role = "rgb" method = args.method equalize = bool(args.equalize) + direction_idx = 0 direction_names = ["rosqueando", "desrosqueando"] - decoded_last = {} - previews_last = {} - meta_last = None - last_frame_id = -1 + frames: Dict[str, Optional[np.ndarray]] = {role: None for role in ROLES} + roi_rects = {role: None for role in ROLES} + + history_by_role_method = { + role: { + m: deque(maxlen=args.history) + for m in METHODS + } + for role in ROLES + } + + best_by_role_method = { + role: { + m: empty_best() + for m in METHODS + } + for role in ROLES + } + + snapshots = [] + + fps_by_role = {role: 0.0 for role in ROLES} + fps_count = {role: 0 for role in ROLES} + fps_t0 = {role: time.time() for role in ROLES} fps_view = 0.0 - fps_stream = 0.0 - t_view_fps = time.time() - t_stream_fps = time.time() - view_frames = 0 - stream_frames_accum = 0 - last_stream_frame_id = None + view_count = 0 + view_t0 = time.time() - panel_rects = {"rgb": None, "re": None, "nir": None, "data": None} - roi_rects = {"rgb": None, "re": None, "nir": None} roi_locked = False dragging_roi = False - drag_start = None + drag_start_src = None - history_by_role = {role: deque(maxlen=args.history) for role in ("rgb", "re", "nir")} - best_by_role = { - role: {"score": 0.0, "smooth": 0.0, "metrics": None, "timestamp": None, "roi": None, "method": method} - for role in ("rgb", "re", "nir") + panel_rects = { + "rgb": None, + "re": None, + "nir": None, + "data": None, } - snapshots = [] + + last_board = None last_msg = "" last_msg_t = 0.0 - window_name = "Focus Calibration Tool - Multispec" + window_name = "Focus Calibration Tool - Sensor Aware" cv2.namedWindow(window_name, cv2.WINDOW_NORMAL) + def set_msg(text): + nonlocal last_msg, last_msg_t + last_msg = str(text) + last_msg_t = time.time() + def inside(rect, px, py): if rect is None: return False + x0, y0, x1, y1 = rect return x0 <= px < x1 and y0 <= py < y1 - def local_from_rect(rect, px, py): - x0, y0, _, _ = rect - return int(px - x0), int(py - y0) + def display_to_source(role, px, py): + rect = panel_rects.get(role) + img = frames.get(role) + + if rect is None or img is None: + return None + + x0, y0, x1, y1 = rect + pw = max(1, x1 - x0) + ph = max(1, y1 - y0) + + sh, sw = img.shape[:2] + + lx = max(0, min(pw - 1, int(px - x0))) + ly = max(0, min(ph - 1, int(py - y0))) + + sx = int(lx * sw / pw) + sy = int(ly * sh / ph) + + sx = max(0, min(sw - 1, sx)) + sy = max(0, min(sh - 1, sy)) + + return sx, sy def on_mouse(event, x, y, flags, param): - nonlocal dragging_roi, drag_start, last_msg, last_msg_t + nonlocal dragging_roi, drag_start_src if roi_locked: return - active_rect = panel_rects.get(selected_role) - if active_rect is None or not inside(active_rect, x, y): + rect = panel_rects.get(selected_role) + + if rect is None or not inside(rect, x, y): return - lx, ly = local_from_rect(active_rect, x, y) + src_pt = display_to_source(selected_role, x, y) + + if src_pt is None: + return if event == cv2.EVENT_LBUTTONDOWN: dragging_roi = True - drag_start = (lx, ly) - roi_rects[selected_role] = (lx, ly, lx + 1, ly + 1) + drag_start_src = src_pt - elif event == cv2.EVENT_MOUSEMOVE and dragging_roi and drag_start is not None: - x0, y0 = drag_start - roi_rects[selected_role] = (x0, y0, lx, ly) + sx, sy = src_pt + roi_rects[selected_role] = (sx, sy, sx + 1, sy + 1) + + elif event == cv2.EVENT_MOUSEMOVE and dragging_roi and drag_start_src: + x0, y0 = drag_start_src + x1, y1 = src_pt + roi_rects[selected_role] = (x0, y0, x1, y1) + + elif event == cv2.EVENT_LBUTTONUP and dragging_roi and drag_start_src: + x0, y0 = drag_start_src + x1, y1 = src_pt + + rect_src = sanitize_roi( + (x0, y0, x1, y1), + frames[selected_role].shape[:2], + ) + + if rect_src is not None: + roi_rects[selected_role] = rect_src + set_msg(f"ROI atualizada: {selected_role.upper()}") - elif event == cv2.EVENT_LBUTTONUP and dragging_roi and drag_start is not None: - x0, y0 = drag_start - roi_rects[selected_role] = (x0, y0, lx, ly) dragging_roi = False - drag_start = None - last_msg = f"ROI atualizada para {selected_role.upper()}" - last_msg_t = time.time() + drag_start_src = None cv2.setMouseCallback(window_name, on_mouse) - try: - with MultiSpectralClient( - width=args.width, - height=args.height, - bayer=args.bayer, - fps=args.fps, - frame_type="RAW_BRUTO", - output_dtype="uint8", - capture_mode=args.capture_mode, - raw_policy=args.raw_policy, - module_calibration_json=args.module_calibration_json, - only_camera=args.only_camera, - ) as cam: + ensure_dir(args.snapshot_dir) - validate_module_ready(cam.get_status(), "RAW_BRUTO", args.raw_policy) + # -------------------------------------------------------- + # Execução + # -------------------------------------------------------- + + try: + with device_ctx as device: + queues = { + role: device.getOutputQueue( + name=spec.stream_name, + maxSize=2, + blocking=False, + ) + for role, spec in specs.items() + } + + print("[OK] Pipeline iniciado. Ajuste a lente devagar.") + print("[OK] Q/ESC sai | 1/2/3 troca câmera | S salva snapshot") while True: - t0 = time.time() + # ------------------------------------------------ + # Coleta independente: foco não precisa sync rígido. + # ------------------------------------------------ + for role, queue in queues.items(): + packet = queue.tryGet() + + if packet is None: + continue - frame, meta, decoded = cam.get_next_decoded(timeout=2.0) - if frame is not None and meta is not None: try: - previews_last = cam.build_visual_preview_from_raw(frame, meta) - except Exception: - previews_last = {} + img = packet.getCvFrame() + except Exception as exc: + set_msg(f"Falha getCvFrame {role}: {exc}") + continue - if meta is not None and frame is not None and meta.get("frame_id") != last_frame_id: - last_frame_id = meta["frame_id"] + if img is None: + continue - if not isinstance(frame, dict): - raise RuntimeError("Este calibrador espera RAW_BRUTO multi-payload como dict de câmeras.") + if img.ndim == 2: + img = cv2.cvtColor(img, cv2.COLOR_GRAY2BGR) - decoded_last = decoded - meta_last = meta + frames[role] = img - curr_frame_id = meta.get("frame_id") - if curr_frame_id is not None and last_stream_frame_id != curr_frame_id: - stream_frames_accum += 1 - last_stream_frame_id = curr_frame_id + if roi_rects[role] is None: + roi_rects[role] = default_roi_for_shape(img.shape[:2], frac=0.42) - dt_stream = time.time() - t_stream_fps - if dt_stream >= 1.0: - fps_stream = stream_frames_accum / dt_stream - stream_frames_accum = 0 - t_stream_fps = time.time() + fps_count[role] += 1 + dt = time.time() - fps_t0[role] - view_frames += 1 - dt_view = time.time() - t_view_fps - if dt_view >= 1.0: - fps_view = view_frames / dt_view - view_frames = 0 - t_view_fps = time.time() + if dt >= 1.0: + fps_by_role[role] = fps_count[role] / dt + fps_count[role] = 0 + fps_t0[role] = time.time() - if decoded_last: - rgb_id, rgb01 = get_image_by_role(decoded_last, "rgb") - re_id, re01 = get_image_by_role(decoded_last, "re") - nir_id, nir01 = get_image_by_role(decoded_last, "nir") + # ------------------------------------------------ + # Se ativa não existe, cai para a primeira disponível. + # ------------------------------------------------ + if selected_role not in specs or frames.get(selected_role) is None: + available = [ + r for r in ROLES + if r in specs and frames.get(r) is not None + ] - # Base de escala visual. - if rgb01 is not None: - base_h, base_w = rgb01.shape[:2] - elif re01 is not None: - base_h, base_w = re01.shape[:2] - elif nir01 is not None: - base_h, base_w = nir01.shape[:2] - else: - base_h, base_w = args.height, args.width + if available: + selected_role = available[0] - for role in ("rgb", "re", "nir"): - if roi_rects[role] is None: - roi_rects[role] = default_roi_for_shape((base_h, base_w), frac=0.42) + active_img = frames.get(selected_role) - re01 = resize_if_needed(re01, (base_h, base_w)) - nir01 = resize_if_needed(nir01, (base_h, base_w)) + # ------------------------------------------------ + # Métrica + # ------------------------------------------------ + metrics = None + score = 0.0 + smooth = 0.0 + trend = { + "status": "aguardando", + "instruction": "aguardando frame", + "pct_of_best": 0.0, + } - rgb_panel = get_preview_panel_by_role(previews_last, meta_last, "rgb") - re_panel = get_preview_panel_by_role(previews_last, meta_last, "re") - nir_panel = get_preview_panel_by_role(previews_last, meta_last, "nir") + history = history_by_role_method[selected_role][method] + best = best_by_role_method[selected_role][method] - if rgb_panel is None: - rgb_panel = to_bgr_u8_from_rgb01(rgb01) if rgb01 is not None else build_empty_panel((base_h, base_w), "RGB") - if re_panel is None: - re_panel = gray_to_color_bgr(re01, "RE") if re01 is not None else build_empty_panel((base_h, base_w), "RE") - if nir_panel is None: - nir_panel = gray_to_color_bgr(nir01, "NIR") if nir01 is not None else build_empty_panel((base_h, base_w), "NIR") - - # Garante que painéis e ROIs estão na mesma resolução visual. - ph = max(rgb_panel.shape[0], re_panel.shape[0], nir_panel.shape[0], base_h) - pw = max(rgb_panel.shape[1], re_panel.shape[1], nir_panel.shape[1], base_w) - - rgb_panel = fit_panel(rgb_panel, (ph, pw)) - re_panel = fit_panel(re_panel, (ph, pw)) - nir_panel = fit_panel(nir_panel, (ph, pw)) - - # Se a resolução visual mudou em relação ao decoded, escalamos a ROI para desenhar corretamente. - sx = pw / float(base_w) - sy = ph / float(base_h) - - def scaled_roi(role): - r = roi_rects[role] - return (int(r[0] * sx), int(r[1] * sy), int(r[2] * sx), int(r[3] * sy)) - - active_img_map = {"rgb": rgb01, "re": re01, "nir": nir01} - active_img = active_img_map.get(selected_role) - active_roi = roi_rects[selected_role] - - metrics = compute_focus_metrics(active_img, active_roi, equalize=equalize) - score = metric_value(metrics, method) if metrics.get("valid") else 0.0 - - hist = history_by_role[selected_role] - smooth_tmp = score - hist.append({ - "t": time.time(), - "score": score, - "smooth": smooth_tmp, - "method": method, - }) - smooth = smooth_score(hist, args.smooth_window) - hist[-1]["smooth"] = smooth - - best = best_by_role[selected_role] - if smooth > best["smooth"]: - best.update({ - "score": score, - "smooth": smooth, - "metrics": metrics, - "timestamp": now_str(), - "roi": list(map(int, active_roi)), - "method": method, - }) - - trend = analyze_trend( - hist, - best["smooth"], - direction_names[direction_idx], - drop_warn_pct=args.drop_warn_pct, - ) - - # Painéis com ROI - draw_roi(rgb_panel, scaled_roi("rgb"), active=(selected_role == "rgb")) - draw_roi(re_panel, scaled_roi("re"), active=(selected_role == "re")) - draw_roi(nir_panel, scaled_roi("nir"), active=(selected_role == "nir")) - - draw_crosshair(rgb_panel) - draw_crosshair(re_panel) - draw_crosshair(nir_panel) - - draw_panel_title(rgb_panel, f"RGB ({rgb_id}) | 1 seleciona", selected_role == "rgb") - draw_panel_title(re_panel, f"RE ({re_id}) | 2 seleciona", selected_role == "re") - draw_panel_title(nir_panel, f"NIR ({nir_id}) | 3 seleciona", selected_role == "nir") - - data_panel = make_data_panel( - (ph, pw), - selected_role, - method, - metrics, - score, - smooth, - best["smooth"], - trend.get("pct_of_best", 0.0), - trend, - fps_stream, - fps_view, - direction_names[direction_idx], - hist, - active_roi, - roi_locked, - equalize, - ) - - panel_rects["rgb"] = (0, 0, pw, ph) - panel_rects["re"] = (pw, 0, pw * 2, ph) - panel_rects["nir"] = (0, ph, pw, ph * 2) - panel_rects["data"] = (pw, ph, pw * 2, ph * 2) - - top = np.hstack([rgb_panel, re_panel]) - bottom = np.hstack([nir_panel, data_panel]) - board = np.vstack([top, bottom]) - - if last_msg and (time.time() - last_msg_t) < 2.5: - cv2.putText(board, last_msg, (16, board.shape[0] - 76), - cv2.FONT_HERSHEY_SIMPLEX, 0.65, (0, 255, 0), 2, cv2.LINE_AA) - - if args.preview_scale != 1.0: - board = cv2.resize( - board, - (int(board.shape[1] * args.preview_scale), int(board.shape[0] * args.preview_scale)), - interpolation=cv2.INTER_NEAREST, + if active_img is not None: + if roi_rects[selected_role] is None: + roi_rects[selected_role] = default_roi_for_shape( + active_img.shape[:2], + frac=0.42, ) - cv2.imshow(window_name, board) + metrics = compute_focus_metrics( + active_img, + roi_rects[selected_role], + equalize=equalize, + ) - else: - blank = np.zeros((720, 1280, 3), dtype=np.uint8) - overlay_hud(blank, ["Aguardando frames do modulo..."], x=40, y=80, font_scale=1.0, line_step=34) - cv2.imshow(window_name, blank) + if metrics.get("valid"): + score = metric_value(metrics, method) + history.append({ + "t": time.time(), + "score": score, + "smooth": score, + }) + + smooth = smooth_from_history( + history, + args.smooth_window, + ) + + history[-1]["smooth"] = smooth + + if smooth > float(best.get("smooth", 0.0) or 0.0): + best.update({ + "score": float(score), + "smooth": float(smooth), + "metrics": metrics, + "timestamp": now_str(), + "roi": list(map(int, roi_rects[selected_role])), + }) + + trend = analyze_trend( + history, + float(best.get("smooth", 0.0) or 0.0), + direction_names[direction_idx], + drop_warn_pct=args.drop_warn_pct, + ) + + # ------------------------------------------------ + # Painéis + # ------------------------------------------------ + ph = int(args.panel_height) + pw = int(args.panel_width) + target_hw = (ph, pw) + + panels = {} + + for role in ROLES: + img = frames.get(role) + spec = specs.get(role) + + if img is None: + if spec is None: + panels[role] = build_empty_panel( + target_hw, + role.upper(), + "camera nao ativa neste modo", + ) + else: + panels[role] = build_empty_panel( + target_hw, + role.upper(), + f"aguardando {spec.sensor_name}", + ) + continue + + view = img.copy() + + if role in ("re", "nir"): + view = colorize_mono_for_view(view, role) + + panel = resize_panel(view, target_hw) + + roi = roi_rects.get(role) + if roi is not None: + draw_roi( + panel, + roi, + img.shape[:2], + active=(role == selected_role), + ) + + draw_crosshair(panel) + + spec = specs[role] + title = ( + f"{role.upper()} | {spec.socket_name} | " + f"{spec.sensor_name} | {img.shape[1]}x{img.shape[0]}" + ) + + draw_panel_title( + panel, + title, + selected=(role == selected_role), + ) + + panels[role] = panel + + active_spec = specs.get(selected_role) + + if active_spec is None: + # Fallback apenas defensivo. + active_spec = next(iter(specs.values())) + + data_panel = make_data_panel( + target_hw, + active_spec, + selected_role, + method, + metrics or {}, + score, + smooth, + best, + trend, + fps_by_role, + fps_view, + direction_names[direction_idx], + history, + roi_locked, + equalize, + ) + + # Layout: + # RGB | RE + # NIR | DATA + panel_rects["rgb"] = (0, 0, pw, ph) + panel_rects["re"] = (pw, 0, pw * 2, ph) + panel_rects["nir"] = (0, ph, pw, ph * 2) + panel_rects["data"] = (pw, ph, pw * 2, ph * 2) + + top = np.hstack([panels["rgb"], panels["re"]]) + bottom = np.hstack([panels["nir"], data_panel]) + board = np.vstack([top, bottom]) + + if last_msg and (time.time() - last_msg_t) < 2.5: + cv2.putText( + board, + last_msg, + (16, board.shape[0] - 14), + cv2.FONT_HERSHEY_SIMPLEX, + 0.58, + (0, 255, 0), + 2, + cv2.LINE_AA, + ) + + last_board = board.copy() + + cv2.imshow(window_name, board) + + # FPS da UI + view_count += 1 + dt_view = time.time() - view_t0 + + if dt_view >= 1.0: + fps_view = view_count / dt_view + view_count = 0 + view_t0 = time.time() + + # ------------------------------------------------ + # Teclas + # ------------------------------------------------ k = cv2.waitKey(1) & 0xFF if k in (ord("q"), ord("Q"), 27): break elif k == ord("1"): - selected_role = "rgb" - last_msg = "Selecionada: RGB" - last_msg_t = time.time() + if "rgb" in specs: + selected_role = "rgb" + set_msg("Selecionada: RGB") elif k == ord("2"): - selected_role = "re" - last_msg = "Selecionada: RE" - last_msg_t = time.time() + if "re" in specs: + selected_role = "re" + set_msg("Selecionada: RE") + else: + set_msg("RE nao ativa neste modo") elif k == ord("3"): - selected_role = "nir" - last_msg = "Selecionada: NIR" - last_msg_t = time.time() + if "nir" in specs: + selected_role = "nir" + set_msg("Selecionada: NIR") + else: + set_msg("NIR nao ativa neste modo") elif k in (ord("m"), ord("M")): - methods = ["laplacian", "tenengrad", "brenner"] - method = methods[(methods.index(method) + 1) % len(methods)] - last_msg = f"Metrica -> {method}" - last_msg_t = time.time() + method = METHODS[(METHODS.index(method) + 1) % len(METHODS)] + set_msg(f"Metrica -> {method}") elif k in (ord("d"), ord("D")): direction_idx = 1 - direction_idx - last_msg = f"Sentido informado -> {direction_names[direction_idx]}" - last_msg_t = time.time() + set_msg(f"Sentido -> {direction_names[direction_idx]}") elif k in (ord("e"), ord("E")): equalize = not equalize - last_msg = f"Equalize -> {'ON' if equalize else 'OFF'}" - last_msg_t = time.time() + set_msg(f"Equalize -> {'ON' if equalize else 'OFF'}") elif k in (ord("l"), ord("L")): roi_locked = not roi_locked - last_msg = f"ROI -> {'travada' if roi_locked else 'editavel'}" - last_msg_t = time.time() + set_msg(f"ROI -> {'travada' if roi_locked else 'editavel'}") elif k in (ord("c"), ord("C")): - # Centraliza ROI da câmera ativa usando a resolução do último frame ativo. - active_img = {"rgb": get_image_by_role(decoded_last, "rgb")[1], - "re": get_image_by_role(decoded_last, "re")[1], - "nir": get_image_by_role(decoded_last, "nir")[1]}.get(selected_role) - if active_img is not None: - roi_rects[selected_role] = default_roi_for_shape(active_img.shape[:2], frac=0.42) - last_msg = f"ROI centralizada em {selected_role.upper()}" - else: - last_msg = "Sem imagem ativa para centralizar ROI" - last_msg_t = time.time() + img = frames.get(selected_role) + + if img is not None: + roi_rects[selected_role] = default_roi_for_shape( + img.shape[:2], + frac=0.42, + ) + set_msg(f"ROI centralizada: {selected_role.upper()}") elif k in (ord("r"), ord("R")): - history_by_role[selected_role].clear() - best_by_role[selected_role] = { - "score": 0.0, - "smooth": 0.0, - "metrics": None, - "timestamp": None, - "roi": list(map(int, roi_rects[selected_role])) if roi_rects[selected_role] else None, - "method": method, - } - last_msg = f"Score resetado: {selected_role.upper()}" - last_msg_t = time.time() + history_by_role_method[selected_role][method].clear() + best_by_role_method[selected_role][method] = empty_best() + + set_msg( + f"Reset: {selected_role.upper()} / {method}" + ) elif k in (ord("s"), ord("S")): - best = best_by_role[selected_role] + stamp = now_file_str() + + png_path = os.path.join( + args.snapshot_dir, + f"focus_{selected_role}_{method}_{stamp}.png", + ) + + if last_board is not None: + cv2.imwrite(png_path, last_board) + snap = { "timestamp": now_str(), "role": selected_role, "method": method, - "roi": list(map(int, roi_rects[selected_role])) if roi_rects[selected_role] else None, - "current_best": json.loads(json.dumps(best)), + "sensor": asdict(specs[selected_role]), + "roi": ( + list(map(int, roi_rects[selected_role])) + if roi_rects[selected_role] is not None + else None + ), + "current_score": float(score), + "current_smooth": float(smooth), + "best": json_safe_best( + best_by_role_method[selected_role][method] + ), "direction_name": direction_names[direction_idx], - "equalize": equalize, + "equalize": bool(equalize), + "png_path": png_path, } + snapshots.append(snap) - last_msg = f"Snapshot salvo em memoria: {selected_role.upper()}" - last_msg_t = time.time() + set_msg(f"Snapshot -> {png_path}") elif k == 32: - payload = build_result_payload(args, best_by_role, snapshots) - save_json(args.out_json, payload) - last_msg = f"Resultado salvo em: {args.out_json}" - last_msg_t = time.time() + payload = build_result_payload( + args, + camera_rows, + specs, + best_by_role_method, + snapshots, + actual_mx, + usb_speed, + ) - dt_loop = time.time() - t0 - if dt_loop < 0.001: - time.sleep(0.001) + save_json(args.out_json, payload) + set_msg(f"JSON salvo -> {args.out_json}") finally: cv2.destroyAllWindows() - print("Fim da calibração de foco.") + + # Salva automaticamente no fechamento também. + try: + payload = build_result_payload( + args, + camera_rows, + specs, + best_by_role_method, + snapshots, + actual_mx, + usb_speed, + ) + save_json(args.out_json, payload) + print(f"[OK] Resultado final salvo em: {args.out_json}") + except Exception as exc: + print(f"[WARN] Não foi possível salvar resultado final: {exc}") + + print("Fim da calibração de foco.") if __name__ == "__main__":