import os import json import time import argparse from datetime import datetime import cv2 import numpy as np from core.oak_fcc3_client import OakFcc3Client as MultiSpectralClient # ============================================================ # Helpers gerais # ============================================================ def now_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) def clamp(v, lo, hi): return max(lo, min(hi, v)) def overlay_hud( img_bgr, lines, x=12, y=24, font_scale=0.58, line_step=22, color=(255, 255, 255), shadow=(0, 0, 0), ): yy = int(y) for s in lines: cv2.putText(img_bgr, str(s), (int(x), yy), cv2.FONT_HERSHEY_SIMPLEX, font_scale, shadow, 3, cv2.LINE_AA) cv2.putText(img_bgr, str(s), (int(x), yy), cv2.FONT_HERSHEY_SIMPLEX, font_scale, color, 1, cv2.LINE_AA) yy += int(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 resize_if_needed(img: np.ndarray, target_hw: tuple[int, int]) -> np.ndarray: if img is None: return None th, tw = target_hw if img.shape[:2] == (th, tw): return img return cv2.resize(img, (tw, th), interpolation=cv2.INTER_LINEAR) def compute_stats(img01: np.ndarray, roi_px=None) -> dict: if img01 is None: return { "valid": False, "pixels": 0, "mean": 0.0, "std": 0.0, "p05": 0.0, "p50": 0.0, "p95": 0.0, "sat_pct": 0.0, "dark_pct": 0.0, } if img01.ndim == 3: arr = ( 0.299 * img01[:, :, 0] + 0.587 * img01[:, :, 1] + 0.114 * img01[:, :, 2] ).astype(np.float32) else: arr = img01.astype(np.float32) if roi_px is not None: x0, y0, x1, y1 = roi_px x0, x1 = sorted((int(x0), int(x1))) y0, y1 = sorted((int(y0), int(y1))) x0 = clamp(x0, 0, arr.shape[1] - 1) x1 = clamp(x1, x0 + 1, arr.shape[1]) y0 = clamp(y0, 0, arr.shape[0] - 1) y1 = clamp(y1, y0 + 1, arr.shape[0]) arr = arr[y0:y1, x0:x1] flat = arr.reshape(-1).astype(np.float32) if flat.size == 0: return { "valid": False, "pixels": 0, "mean": 0.0, "std": 0.0, "p05": 0.0, "p50": 0.0, "p95": 0.0, "sat_pct": 0.0, "dark_pct": 0.0, } return { "valid": True, "pixels": int(flat.size), "mean": float(flat.mean()), "std": float(flat.std()), "p05": float(np.percentile(flat, 5)), "p50": float(np.percentile(flat, 50)), "p95": float(np.percentile(flat, 95)), "sat_pct": float((flat >= 0.98).mean() * 100.0), "dark_pct": float((flat <= 0.02).mean() * 100.0), } def pct_to_px(roi_pct: dict, w: int, h: int): x0 = int(float(roi_pct.get("x0", 0.0)) * w) y0 = int(float(roi_pct.get("y0", 0.0)) * h) x1 = int(float(roi_pct.get("x1", 1.0)) * w) y1 = int(float(roi_pct.get("y1", 1.0)) * h) x0 = clamp(x0, 0, w - 1) x1 = clamp(x1, x0 + 1, w) y0 = clamp(y0, 0, h - 1) y1 = clamp(y1, y0 + 1, h) return x0, y0, x1, y1 def px_to_pct(rect, w: int, h: int): x0, y0, x1, y1 = rect x0, x1 = sorted((int(x0), int(x1))) y0, y1 = sorted((int(y0), int(y1))) x0 = clamp(x0, 0, w - 1) x1 = clamp(x1, x0 + 1, w) y0 = clamp(y0, 0, h - 1) y1 = clamp(y1, y0 + 1, h) return { "x0": round(x0 / float(w), 6), "y0": round(y0 / float(h), 6), "x1": round(x1 / float(w), 6), "y1": round(y1 / float(h), 6), } def get_decoded_by_role(decoded: dict, role: str): role = str(role).lower() for cam_id, item in (decoded or {}).items(): if str(item.get("role", "")).lower() == role: return cam_id, item return None, None 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_visual_preview_by_role(visual_previews: dict, meta: dict, role: str): """ Busca uma imagem visual BGR dentro do retorno de cam.build_visual_preview_from_raw(), usando camera_info para descobrir o role rgb/re/nir. Retorna: cam_id, img_bgr """ if not visual_previews: return None, None camera_info = (meta or {}).get("camera_info", {}) or {} role = str(role).lower() for cam_id, img in visual_previews.items(): cam_role = str(camera_info.get(cam_id, {}).get("role", "")).lower() if cam_role == role: return cam_id, img return None, None def validate_module_ready(status: dict, raw_policy: str): if not status.get("ok", True): raise RuntimeError(f"Status inválido retornado pelo modulo: {status}") active_roles = status.get("active_roles", {}) or {} active_count = int(status.get("camera_count_active", 0)) if raw_policy == "require_triple": missing = [role for role in ("rgb", "nir", "re") if role not in active_roles] if missing: raise RuntimeError( f"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.") # ============================================================ # Config radiométrico # ============================================================ def base_ae_contract(): return { "enabled": True, "interval_s": 0.20, "verbose": True, "control_metric": "p50", "target_value": 0.34, "deadband": 0.035, "p95_limit": 0.94, "saturation_limit_pct": 0.50, "dark_limit_pct": 35.0, # Novo controle proporcional por razão "control_strategy": "ratio", "ratio_alpha": 0.35, "ratio_min": 0.65, "ratio_max": 1.35, # Redução rápida quando satura "reduce_fast_factor": 0.80, # Mantém compatibilidade com o modo antigo "alpha": 0.18, "exp_step_gain": 0.55, "factor_min": 0.55, "factor_max": 1.28, "prefer_exposure": True, "exp_min_us": 100, "exp_max_us": 80000, "gain_min": 1.0, "gain_max": 4.0, "gain_return_enabled": True, "gain_reduce_on_saturation": True, "gain_increase_required_cycles": 5, "gain_decrease_required_cycles": 2, "gain_step_up": 0.20, "gain_step_down": 0.50, "gain_hard_reset_on_saturation": False, "exp_high_ratio_for_gain": 0.95, "exp_low_ratio_for_gain_return": 0.75, "role_limits": { "rgb": {"exp_min_us": 100, "exp_max_us": 80000, "gain_min": 1.0, "gain_max": 2.0}, "re": {"exp_min_us": 100, "exp_max_us": 2500, "gain_min": 1.0, "gain_max": 2.0}, "nir": {"exp_min_us": 100, "exp_max_us": 3000, "gain_min": 1.0, "gain_max": 2.0}, }, "exp_apply_threshold_us": 80, "gain_apply_threshold": 0.05, "ready_required_cycles": 3, "apply_same_spectral_to_both": True, "spectral_roles": ["re", "nir"], } def default_profile_global(): cfg = base_ae_contract() base = { "x0": 0.08, "y0": 0.08, "x1": 0.92, "y1": 0.92, } cfg.update({ "metering_mode": "global", "spectral_control_mode": "shared", "global_roi_pct": base, "global_roi_pct_by_role": { "rgb": dict(base), "re": dict(base), "nir": dict(base), }, }) return { "radiometric_config": cfg } def make_default_patch(patch_type: str, target: float, weight: float): return { "name": f"{patch_type}_reference", "type": patch_type, "roles": ROLES[:] if "ROLES" in globals() else ["rgb", "re", "nir"], "target_value": target, "weight": weight, # Compatibilidade com o controller antigo: primeira ROI ativa de RGB. "roi_pct": {}, "roi_pct_by_role": {"rgb": {}, "re": {}, "nir": {}}, # Formato novo: lista dinâmica por câmera/role. # Cada item: {name, enabled, roi_pct, created_at, updated_at} "roi_list_by_role": {"rgb": [], "re": [], "nir": []}, } def default_profile_patches(): cfg = base_ae_contract() gray_patch = make_default_patch("gray", 0.34, 1.0) gray_patch["target_value_by_role"] = { "rgb": 0.34, "re": 0.24, "nir": 0.30, } cfg.update({ "metering_mode": "reference_patches", "spectral_control_mode": "shared", "deadband": 0.035, "patch_control_mode": "gray_primary", "patch_require_order": True, "patch_min_separation": 0.08, "patch_white_sat_limit_pct": 0.50, "patch_white_p95_limit": 0.90, "patch_black_dark_limit_pct": 80.0, "patch_black_max_p50": 0.20, "patch_gray_min_p50": 0.08, "patch_gray_max_p50": 0.85, # Novo contrato: o runtime pode combinar N ROIs por cor/camera. "patch_roi_contract": "multi_roi_by_role_v1", "patch_roi_reduce_method": "median_valid_rois", "patch_roi_outlier_reject": True, "patch_roi_max_p50_delta": 0.12, "reference_patches": [ make_default_patch("black", 0.06, 0.25), gray_patch, make_default_patch("white", 0.78, 0.7), ], }) return { "radiometric_config": cfg } def get_active_profile_name(data: dict) -> str: name = str(data.get("active_profile", "global_scene_mode")) if name not in ("global_scene_mode", "three_reference_patches_mode"): return "global_scene_mode" return name def set_active_profile_name(data: dict, profile_name: str): if profile_name not in ("global_scene_mode", "three_reference_patches_mode"): profile_name = "global_scene_mode" data["active_profile"] = profile_name def get_active_radiometric_config(data: dict) -> dict: profile_name = get_active_profile_name(data) profile = data.get(profile_name, {}) or {} cfg = profile.get("radiometric_config", {}) or {} return json.loads(json.dumps(cfg)) def update_root_radiometric_config(data: dict): data["radiometric_config"] = get_active_radiometric_config(data) def load_or_default_config(path: str): if path and os.path.isfile(path): with open(path, "r", encoding="utf-8") as f: data = json.load(f) else: data = {} data.setdefault("schema", "multispec_radiometric_config_profiles_v3") data.setdefault("saved_at", now_str()) data.setdefault("active_profile", "global_scene_mode") data.setdefault("global_scene_mode", default_profile_global()) data.setdefault("three_reference_patches_mode", default_profile_patches()) data.setdefault("patch_normalization", { "enabled": True, "apply_when_metering_mode": "reference_patches", "apply_stage": "after_fusion", "method": "gray_scale_with_white_guard", "space": "multispec_tensor", "targets_by_patch_channel": { "black": { "R": 0.06, "G": 0.06, "B": 0.06, "RE": 0.06, "NIR": 0.06 }, "gray": { "R": 0.34, "G": 0.34, "B": 0.34, "RE": 0.24, "NIR": 0.30 }, "white": { "R": 0.78, "G": 0.78, "B": 0.78, "RE": 0.78, "NIR": 0.78 } }, "white_guard_max": 0.92, "white_guard_max_by_channel": { "R": 0.92, "G": 0.92, "B": 0.92, "RE": 0.88, "NIR": 0.88 }, "scale_min": 0.35, "scale_max": 2.50, "clip_output": True, "require_valid_gray": True, "use_black_for_offset": False, "save_patch_stats": True }) # Migração: se vier arquivo antigo sem contrato novo, injeta defaults novos for profile_name, default_fn in ( ("global_scene_mode", default_profile_global), ("three_reference_patches_mode", default_profile_patches), ): default_profile = default_fn() data.setdefault(profile_name, default_profile) data[profile_name].setdefault("radiometric_config", {}) default_cfg = default_profile["radiometric_config"] cfg = data[profile_name]["radiometric_config"] for k, v in default_cfg.items(): cfg.setdefault(k, v) update_root_radiometric_config(data) return data def save_config(path: str, data: dict): ensure_dir(os.path.dirname(path) or ".") data = dict(data) data["schema"] = "multispec_radiometric_config_profiles_v3" data["saved_at"] = now_str() update_root_radiometric_config(data) with open(path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) ROLES = ["rgb", "re", "nir"] def normalize_role(role: str) -> str: role = str(role or "rgb").lower() return role if role in ROLES else "rgb" def default_roi(): return {"x0": 0.08, "y0": 0.08, "x1": 0.92, "y1": 0.92} def clone_roi(roi: dict) -> dict: roi = roi or {} return { "x0": float(roi.get("x0", 0.08)), "y0": float(roi.get("y0", 0.08)), "x1": float(roi.get("x1", 0.92)), "y1": float(roi.get("y1", 0.92)), } def make_roi_by_role(base_roi=None): base = clone_roi(base_roi or default_roi()) return {role: dict(base) for role in ROLES} def ensure_global_roi_by_role(data: dict): data.setdefault("global_scene_mode", default_profile_global()) cfg = data["global_scene_mode"].setdefault( "radiometric_config", default_profile_global()["radiometric_config"], ) legacy = cfg.get("global_roi_pct", default_roi()) by_role = cfg.setdefault("global_roi_pct_by_role", make_roi_by_role(legacy)) for role in ROLES: if role not in by_role or not by_role[role]: by_role[role] = clone_roi(legacy) return by_role def get_global_roi_for_role(data: dict, role: str): role = normalize_role(role) by_role = ensure_global_roi_by_role(data) return by_role.get(role, clone_roi(default_roi())) def set_global_roi_for_role(data: dict, role: str, roi_pct: dict): role = normalize_role(role) by_role = ensure_global_roi_by_role(data) by_role[role] = roi_pct # Compatibilidade: mantém uma ROI antiga preenchida. # Uso: média/legado/visual antigo. O controller novo deverá usar by_role. data["global_scene_mode"]["radiometric_config"]["global_roi_pct"] = by_role.get("rgb", roi_pct) def get_patches(data: dict): return ( data.get("three_reference_patches_mode", {}) .get("radiometric_config", {}) .get("reference_patches", []) ) def make_roi_entry(roi_pct: dict, name: str | None = None, enabled: bool = True) -> dict: return { "name": name or "roi_01", "enabled": bool(enabled), "roi_pct": clone_roi(roi_pct), "created_at": now_str(), "updated_at": now_str(), } def normalize_roi_entry(entry, idx: int) -> dict | None: """Aceita formatos antigos e novos, devolvendo sempre um item padrão.""" if not entry: return None if isinstance(entry, dict) and "roi_pct" in entry: roi = entry.get("roi_pct") or {} if not roi: return None out = dict(entry) out["name"] = str(out.get("name") or f"roi_{idx + 1:02d}") out["enabled"] = bool(out.get("enabled", True)) out["roi_pct"] = clone_roi(roi) out.setdefault("created_at", now_str()) out["updated_at"] = str(out.get("updated_at") or now_str()) return out if isinstance(entry, dict) and all(k in entry for k in ("x0", "y0", "x1", "y1")): return make_roi_entry(entry, name=f"roi_{idx + 1:02d}", enabled=True) return None def sync_patch_legacy_roi_fields(patch: dict): """Mantém roi_pct e roi_pct_by_role compatíveis com scripts antigos.""" roi_lists = patch.setdefault("roi_list_by_role", {}) by_role = patch.setdefault("roi_pct_by_role", {}) for role in ROLES: entries = roi_lists.setdefault(role, []) first_active = next((e.get("roi_pct") for e in entries if e.get("enabled", True) and e.get("roi_pct")), {}) by_role[role] = clone_roi(first_active) if first_active else {} patch["roi_pct"] = by_role.get("rgb", {}) or {} def ensure_patch_roi_lists_by_role(patch: dict): """ Migra o formato antigo: roi_pct_by_role[role] = {x0,y0,x1,y1} para o formato novo: roi_list_by_role[role] = [{name, enabled, roi_pct, ...}, ...] Também aceita, por tolerância, caso alguém já tenha salvo uma lista dentro de roi_pct_by_role. """ legacy_global = patch.get("roi_pct", {}) or {} legacy_by_role = patch.get("roi_pct_by_role", {}) or {} roi_lists = patch.setdefault("roi_list_by_role", {}) for role in ROLES: raw_list = roi_lists.get(role, []) # Caso raro: formato novo foi salvo diretamente em roi_pct_by_role. if not raw_list and isinstance(legacy_by_role.get(role), list): raw_list = legacy_by_role.get(role) or [] normalized = [] if isinstance(raw_list, list): for idx, item in enumerate(raw_list): entry = normalize_roi_entry(item, idx) if entry is not None: normalized.append(entry) elif isinstance(raw_list, dict) and raw_list: entry = normalize_roi_entry(raw_list, 0) if entry is not None: normalized.append(entry) # Migração do formato antigo, se ainda não houver lista. if not normalized: old_roi = legacy_by_role.get(role, {}) if isinstance(legacy_by_role, dict) else {} if not old_roi and legacy_global: old_roi = legacy_global if isinstance(old_roi, dict) and old_roi: normalized.append(make_roi_entry(old_roi, name=f"{role}_legacy_01", enabled=True)) # Garante nomes estáveis e únicos. seen = set() for idx, entry in enumerate(normalized): name = str(entry.get("name") or f"roi_{idx + 1:02d}") if name in seen: name = f"{name}_{idx + 1:02d}" seen.add(name) entry["name"] = name roi_lists[role] = normalized sync_patch_legacy_roi_fields(patch) return roi_lists # Alias antigo mantido para não quebrar chamadas existentes. def ensure_patch_roi_by_role(patch: dict): ensure_patch_roi_lists_by_role(patch) return patch.setdefault("roi_pct_by_role", {}) def get_patch_by_type(data: dict, patch_type: str): patch_type = str(patch_type).lower() for p in get_patches(data): if str(p.get("type", "")).lower() == patch_type: ensure_patch_roi_lists_by_role(p) return p return None def ensure_patch_exists(data: dict, patch_type: str): data.setdefault("three_reference_patches_mode", default_profile_patches()) cfg = data["three_reference_patches_mode"].setdefault( "radiometric_config", default_profile_patches()["radiometric_config"], ) patches = cfg.setdefault( "reference_patches", default_profile_patches()["radiometric_config"]["reference_patches"], ) patch_type = str(patch_type).lower() target = {"black": 0.06, "gray": 0.40, "white": 0.78}.get(patch_type, 0.40) weight = {"black": 0.25, "gray": 1.0, "white": 0.7}.get(patch_type, 1.0) for p in patches: if str(p.get("type", "")).lower() == patch_type: ensure_patch_roi_lists_by_role(p) return p patch = make_default_patch(patch_type, target, weight) patches.append(patch) ensure_patch_roi_lists_by_role(patch) return patch def get_patch_roi_entries_for_role(data: dict, patch_type: str, role: str, enabled_only: bool = False): role = normalize_role(role) patch = get_patch_by_type(data, patch_type) if not patch: return [] roi_lists = ensure_patch_roi_lists_by_role(patch) entries = list(roi_lists.get(role, []) or []) if enabled_only: entries = [e for e in entries if e.get("enabled", True) and e.get("roi_pct")] return entries def get_patch_roi_for_role(data: dict, patch_type: str, role: str): """Compatibilidade: retorna a primeira ROI ativa da lista.""" entries = get_patch_roi_entries_for_role(data, patch_type, role, enabled_only=True) if entries: return entries[0].get("roi_pct", {}) or {} patch = get_patch_by_type(data, patch_type) if not patch: return {} return patch.get("roi_pct_by_role", {}).get(normalize_role(role), {}) or patch.get("roi_pct", {}) or {} def get_patch_roi_entry(data: dict, patch_type: str, role: str, index: int): entries = get_patch_roi_entries_for_role(data, patch_type, role, enabled_only=False) if not entries: return None, -1 index = clamp(int(index), 0, len(entries) - 1) return entries[index], index def set_patch_roi_for_role(data: dict, patch_type: str, role: str, roi_pct: dict, index: int | None = None, append: bool = False): patch = ensure_patch_exists(data, patch_type) role = normalize_role(role) roi_lists = ensure_patch_roi_lists_by_role(patch) entries = roi_lists.setdefault(role, []) if append or index is None or index >= len(entries) or index < 0: entry = make_roi_entry( roi_pct, name=f"{patch_type}_{role}_{len(entries) + 1:02d}", enabled=True, ) entries.append(entry) saved_index = len(entries) - 1 else: saved_index = int(index) old = entries[saved_index] old["roi_pct"] = clone_roi(roi_pct) old["enabled"] = bool(old.get("enabled", True)) old["updated_at"] = now_str() sync_patch_legacy_roi_fields(patch) return saved_index def delete_patch_roi_for_role(data: dict, patch_type: str, role: str, index: int): patch = get_patch_by_type(data, patch_type) if not patch: return False, 0 role = normalize_role(role) roi_lists = ensure_patch_roi_lists_by_role(patch) entries = roi_lists.setdefault(role, []) if not entries: return False, 0 index = clamp(int(index), 0, len(entries) - 1) entries.pop(index) sync_patch_legacy_roi_fields(patch) return True, len(entries) def toggle_patch_roi_enabled_for_role(data: dict, patch_type: str, role: str, index: int): entry, idx = get_patch_roi_entry(data, patch_type, role, index) if entry is None: return False, False entry["enabled"] = not bool(entry.get("enabled", True)) entry["updated_at"] = now_str() patch = get_patch_by_type(data, patch_type) if patch: sync_patch_legacy_roi_fields(patch) return True, bool(entry["enabled"]) def add_empty_patch_roi_slot(data: dict, patch_type: str, role: str): # Usa uma ROI pequena central como placeholder, para o usuário arrastar por cima depois. return set_patch_roi_for_role( data, patch_type, role, {"x0": 0.45, "y0": 0.45, "x1": 0.55, "y1": 0.55}, append=True, ) def set_shared_mode(data: dict, shared: bool): for profile in ("global_scene_mode", "three_reference_patches_mode"): data.setdefault(profile, default_profile_global() if profile == "global_scene_mode" else default_profile_patches()) cfg = data[profile].setdefault("radiometric_config", {}) cfg["spectral_control_mode"] = "shared" if shared else "independent" cfg["apply_same_spectral_to_both"] = bool(shared) update_root_radiometric_config(data) # ============================================================ # UI # ============================================================ PATCH_COLORS = { "global": (0, 255, 255), "black": (80, 80, 80), "gray": (180, 180, 180), "white": (255, 255, 255), } PATCH_ORDER = ["black", "gray", "white"] def draw_roi_on_panel(panel, roi_pct, label, color, thickness=2): if roi_pct is None: return h, w = panel.shape[:2] x0, y0, x1, y1 = pct_to_px(roi_pct, w, h) cv2.rectangle(panel, (x0, y0), (x1, y1), color, thickness) cv2.putText(panel, label, (x0 + 5, max(20, y0 - 6)), cv2.FONT_HERSHEY_SIMPLEX, 0.55, (0, 0, 0), 3, cv2.LINE_AA) cv2.putText(panel, label, (x0 + 5, max(20, y0 - 6)), cv2.FONT_HERSHEY_SIMPLEX, 0.55, color, 1, cv2.LINE_AA) def draw_all_rois(panel, data, selected_target, mode, panel_role, edit_role, selected_roi_index=0): panel_role = normalize_role(panel_role) edit_role = normalize_role(edit_role) is_edit_panel = panel_role == edit_role if mode == "global": roi = get_global_roi_for_role(data, panel_role) label = f"GLOBAL/{panel_role.upper()}" thickness = 3 if is_edit_panel else 2 draw_roi_on_panel(panel, roi, label, PATCH_COLORS["global"], thickness) else: for p in get_patches(data): typ = str(p.get("type", "")).lower() color = PATCH_COLORS.get(typ, (0, 255, 255)) entries = get_patch_roi_entries_for_role(data, typ, panel_role, enabled_only=False) for idx, entry in enumerate(entries): roi = entry.get("roi_pct", {}) if not roi: continue enabled = bool(entry.get("enabled", True)) selected = typ == selected_target and is_edit_panel and idx == selected_roi_index thickness = 3 if selected else 1 if not enabled else 2 label = f"{typ.upper()}/{panel_role.upper()}#{idx + 1}" if not enabled: label += " OFF" draw_roi_on_panel(panel, roi, label, color, thickness) def build_board( decoded, data, mode, selected_target, edit_role, selected_roi_index, drag_rect_local, drag_role, panel_rects, preview_scale=1.0, visual_previews=None, meta=None, beauty_preview=True, ): rgb_id, rgb01 = get_image_by_role(decoded, "rgb") re_id, re01 = get_image_by_role(decoded, "re") nir_id, nir01 = get_image_by_role(decoded, "nir") # ------------------------------------------------------------ # Tamanho base SEMPRE vem do decoded, porque ROI/stats usam dado real. # O preview visual é só para desenhar bonito. # ------------------------------------------------------------ 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 = 800, 1280 # ------------------------------------------------------------ # Preview bonito, igual ao capture. # ------------------------------------------------------------ rgb_vis_id, rgb_vis = get_visual_preview_by_role(visual_previews, meta, "rgb") re_vis_id, re_vis = get_visual_preview_by_role(visual_previews, meta, "re") nir_vis_id, nir_vis = get_visual_preview_by_role(visual_previews, meta, "nir") if beauty_preview and rgb_vis is not None: rgb_panel = rgb_vis.copy() if rgb_panel.shape[:2] != (base_h, base_w): rgb_panel = cv2.resize(rgb_panel, (base_w, base_h), interpolation=cv2.INTER_LINEAR) rgb_id = rgb_vis_id else: if rgb01 is not None: rgb_panel = to_bgr_u8_from_rgb01(rgb01) else: rgb_panel = np.zeros((base_h, base_w, 3), dtype=np.uint8) overlay_hud(rgb_panel, ["RGB", "sem frame"]) if beauty_preview and re_vis is not None: re_panel = re_vis.copy() if re_panel.shape[:2] != (base_h, base_w): re_panel = cv2.resize(re_panel, (base_w, base_h), interpolation=cv2.INTER_LINEAR) re_id = re_vis_id else: re01_show = resize_if_needed(re01, (base_h, base_w)) if re01 is not None else None re_panel = gray_to_bgr_u8(re01_show) if re01_show is not None else np.zeros_like(rgb_panel) if beauty_preview and nir_vis is not None: nir_panel = nir_vis.copy() if nir_panel.shape[:2] != (base_h, base_w): nir_panel = cv2.resize(nir_panel, (base_w, base_h), interpolation=cv2.INTER_LINEAR) nir_id = nir_vis_id else: nir01_show = resize_if_needed(nir01, (base_h, base_w)) if nir01 is not None else None nir_panel = gray_to_bgr_u8(nir01_show) if nir01_show is not None else np.zeros_like(rgb_panel) draw_all_rois(rgb_panel, data, selected_target, mode, "rgb", edit_role, selected_roi_index) draw_all_rois(re_panel, data, selected_target, mode, "re", edit_role, selected_roi_index) draw_all_rois(nir_panel, data, selected_target, mode, "nir", edit_role, selected_roi_index) if drag_rect_local is not None: x0, y0, x1, y1 = drag_rect_local color = PATCH_COLORS["global"] if mode == "global" else PATCH_COLORS.get(selected_target, (0, 255, 255)) if drag_role == "rgb": cv2.rectangle(rgb_panel, (x0, y0), (x1, y1), color, 1) elif drag_role == "re": cv2.rectangle(re_panel, (x0, y0), (x1, y1), color, 1) elif drag_role == "nir": cv2.rectangle(nir_panel, (x0, y0), (x1, y1), color, 1) overlay_hud(rgb_panel, [f"RGB ({rgb_id})"], y=24) overlay_hud(re_panel, [f"RE ({re_id})"], y=24) overlay_hud(nir_panel, [f"NIR ({nir_id})"], y=24) ph = max(rgb_panel.shape[0], re_panel.shape[0], nir_panel.shape[0]) pw = max(rgb_panel.shape[1], re_panel.shape[1], nir_panel.shape[1]) def fit_panel(img): if img.shape[:2] != (ph, pw): return cv2.resize(img, (pw, ph), interpolation=cv2.INTER_NEAREST) return img rgb_panel = fit_panel(rgb_panel) re_panel = fit_panel(re_panel) nir_panel = fit_panel(nir_panel) data_panel = np.zeros((ph, pw, 3), dtype=np.uint8) 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]) x0, y0, x1, y1 = panel_rects["data"] lines = build_data_lines(decoded, data, mode, selected_target, edit_role, selected_roi_index, base_w, base_h) overlay_hud(board, lines, x=x0 + 16, y=y0 + 28, font_scale=0.50, line_step=20) if preview_scale != 1.0: board = cv2.resize( board, (int(board.shape[1] * preview_scale), int(board.shape[0] * preview_scale)), interpolation=cv2.INTER_NEAREST, ) return board def build_data_lines(decoded, data, mode, selected_target, edit_role, selected_roi_index, base_w, base_h): edit_role = normalize_role(edit_role) active_profile = get_active_profile_name(data) active_cfg = get_active_radiometric_config(data) lines = [ "RADIOMETRIC CONFIG TOOL", f"modo={mode.upper()} | camera={edit_role.upper()} | active={active_profile}", f"spectral={active_cfg.get('spectral_control_mode')} | strategy={active_cfg.get('control_strategy')}", f"roi_contract={active_cfg.get('patch_roi_contract', 'legacy_single_roi')}", "", "Arraste no painel da camera editada para definir/atualizar ROI.", "PATCHES agora suportam N ROIs por cor e por camera.", "", ] if mode == "global": lines.append("GLOBAL ROI por camera:") for role in ROLES: roi_pct = get_global_roi_for_role(data, role) marker = "*" if role == edit_role else " " lines.append(f"{marker} {role.upper()}: roi={roi_pct}") lines.append("") lines.append("Stats GLOBAL:") lines.extend(stats_lines_for_mode(data, decoded, mode="global", patch_type=None)) else: entries_edit = get_patch_roi_entries_for_role(data, selected_target, edit_role, enabled_only=False) n_edit = len(entries_edit) selected_roi_index = clamp(selected_roi_index, 0, max(0, n_edit - 1)) if n_edit else 0 lines.append(f"PATCH selecionado: {selected_target.upper()}") lines.append(f"ROI selecionada {edit_role.upper()}: #{selected_roi_index + 1 if n_edit else 0}/{n_edit}") lines.append("Contagem de ROIs por camera:") for role in ROLES: entries = get_patch_roi_entries_for_role(data, selected_target, role, enabled_only=False) enabled = sum(1 for e in entries if e.get("enabled", True)) marker = "*" if role == edit_role else " " lines.append(f"{marker} {role.upper()}: {enabled}/{len(entries)} ativas") if n_edit: entry = entries_edit[selected_roi_index] lines.append(f"ROI atual: {entry.get('name')} | enabled={entry.get('enabled', True)}") lines.append(f"rect={entry.get('roi_pct')}") else: lines.append("ROI atual: nenhuma. Arraste para criar a primeira.") sel_patch = get_patch_by_type(data, selected_target) if sel_patch: lines.append( f"target={float(sel_patch.get('target_value', 0.0)):.2f} " f"weight={float(sel_patch.get('weight', 1.0)):.2f}" ) lines.append("") lines.append(f"Stats robustas {selected_target.upper()}:") lines.extend(stats_lines_for_mode(data, decoded, mode="patches", patch_type=selected_target)) lines.extend([ "", "M = GLOBAL/PATCHES | C = camera | V = preview bonito/bruto", "1/2/3 = BLACK/GRAY/WHITE | S = shared/independent", "N = nova ROI | [ ] = troca ROI | D = apaga ROI | T = liga/desliga ROI", "P ou SPACE = salva JSON | R = defaults | Q/Esc = sai", ]) return lines def aggregate_roi_stats(stats_list: list[dict]) -> dict: valid = [s for s in stats_list if s.get("valid")] if not valid: return {"valid": False, "count": 0} p50 = np.array([s["p50"] for s in valid], dtype=np.float32) p95 = np.array([s["p95"] for s in valid], dtype=np.float32) sat = np.array([s["sat_pct"] for s in valid], dtype=np.float32) dark = np.array([s["dark_pct"] for s in valid], dtype=np.float32) std = np.array([s["std"] for s in valid], dtype=np.float32) return { "valid": True, "count": len(valid), "p50": float(np.median(p50)), "p95": float(np.median(p95)), "sat_pct": float(np.median(sat)), "dark_pct": float(np.median(dark)), "std": float(np.median(std)), "p50_spread": float(p50.max() - p50.min()) if len(p50) > 1 else 0.0, "p50_min": float(p50.min()), "p50_max": float(p50.max()), } def stats_lines_for_mode(data, decoded, mode: str, patch_type: str | None = None): lines = [] for role in ROLES: _, img = get_image_by_role(decoded, role) if img is None: lines.append(f"{role.upper()}: sem frame") continue h, w = img.shape[:2] if mode == "global": roi_pct = get_global_roi_for_role(data, role) if not roi_pct: lines.append(f"{role.upper()}: sem ROI") continue roi = pct_to_px(roi_pct, w, h) st = compute_stats(img, roi) lines.append( f"{role.upper()}: p50={st['p50']:.3f} p95={st['p95']:.3f} " f"sat={st['sat_pct']:.2f}% dark={st['dark_pct']:.1f}%" ) continue entries = get_patch_roi_entries_for_role(data, patch_type, role, enabled_only=True) if not entries: lines.append(f"{role.upper()}: sem ROI ativa") continue stats = [] p50_each = [] for entry in entries: roi_pct = entry.get("roi_pct", {}) if not roi_pct: continue roi = pct_to_px(roi_pct, w, h) st = compute_stats(img, roi) stats.append(st) if st.get("valid"): p50_each.append(st["p50"]) ag = aggregate_roi_stats(stats) if not ag.get("valid"): lines.append(f"{role.upper()}: ROIs invalidas") continue mini = ",".join(f"{v:.2f}" for v in p50_each[:4]) if len(p50_each) > 4: mini += ",..." lines.append( f"{role.upper()}: n={ag['count']} p50_med={ag['p50']:.3f} " f"spread={ag['p50_spread']:.3f} sat_med={ag['sat_pct']:.2f}%" ) lines.append(f" p50_each=[{mini}]") return lines def rect_inside(rect, x, y): if rect is None: return False x0, y0, x1, y1 = rect return x0 <= x < x1 and y0 <= y < y1 def local_from_rect(rect, x, y): x0, y0, _, _ = rect return int(x - x0), int(y - y0) # ============================================================ # Main # ============================================================ def main(): parser = argparse.ArgumentParser( description="Ferramenta visual para parametrizar o radiometric_config global ou por 3 patches.", 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="RGGB", 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("--module_calibration_json", default="calibration/module_params.json") parser.add_argument("--out_json", default="calibration/radiometric_config.json") parser.add_argument("--load_json", default="") parser.add_argument("--preview_scale", type=float, default=0.75) args = parser.parse_args() config_path = args.load_json or args.out_json data = load_or_default_config(config_path) mode = "global" selected_target = "gray" selected_roi_index = 0 edit_role = "rgb" drag_role = None beauty_preview = True visual_previews_last = {} raw_meta_last = {} panel_rects = {"rgb": None, "re": None, "nir": None, "data": None} dragging = False drag_start = None drag_rect_local = None last_msg = "" last_msg_t = 0.0 last_frame_id = -1 decoded_last = {} window_name = "Radiometric Config Tool" def on_mouse(event, x, y, flags, param): nonlocal dragging, drag_start, drag_rect_local, last_msg, last_msg_t, data, drag_role, selected_roi_index # Coordenadas vêm depois do preview_scale. Reescala para board real. if args.preview_scale != 1.0: x = int(x / args.preview_scale) y = int(y / args.preview_scale) edit_rect = panel_rects.get(edit_role) if not rect_inside(edit_rect, x, y): return lx, ly = local_from_rect(edit_rect, x, y) if event == cv2.EVENT_LBUTTONDOWN: dragging = True drag_role = edit_role drag_start = (lx, ly) drag_rect_local = (lx, ly, lx + 1, ly + 1) elif event == cv2.EVENT_MOUSEMOVE and dragging: sx, sy = drag_start drag_rect_local = (sx, sy, lx, ly) elif event == cv2.EVENT_LBUTTONUP and dragging: dragging = False sx, sy = drag_start rect = (sx, sy, lx, ly) drag_rect_local = None # Descobre tamanho local do painel da camera editada. edit_rect = panel_rects.get(drag_role or edit_role) if edit_rect is None: return _, _, x1, y1 = edit_rect x0r, y0r, _, _ = edit_rect w = x1 - x0r h = y1 - y0r roi_pct = px_to_pct(rect, w, h) role_to_save = normalize_role(drag_role or edit_role) if mode == "global": set_global_roi_for_role(data, role_to_save, roi_pct) last_msg = f"GLOBAL ROI {role_to_save.upper()} atualizada: {roi_pct}" else: selected_roi_index = set_patch_roi_for_role( data, selected_target, role_to_save, roi_pct, index=selected_roi_index, append=False ) last_msg = ( f"{selected_target.upper()} ROI {role_to_save.upper()} " f"#{selected_roi_index + 1} atualizada: {roi_pct}" ) drag_role = None last_msg_t = time.time() cv2.namedWindow(window_name, cv2.WINDOW_NORMAL) 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 ) as cam: validate_module_ready(cam.get_status(), args.raw_policy) while True: raw_frame, raw_meta, decoded = cam.get_next_decoded(timeout=2.0) visual_previews = {} try: if isinstance(raw_frame, dict): visual_previews = cam.build_visual_preview_from_raw(raw_frame, raw_meta) except Exception as e: visual_previews = {} print(f"[WARN] Falha ao gerar beauty preview: {e}") if raw_meta is not None and raw_meta.get("frame_id") != last_frame_id: last_frame_id = raw_meta.get("frame_id") decoded_last = decoded visual_previews_last = visual_previews raw_meta_last = raw_meta if decoded_last: board = build_board( decoded=decoded_last, data=data, mode=mode, selected_target=selected_target, edit_role=edit_role, selected_roi_index=selected_roi_index, drag_rect_local=drag_rect_local, drag_role=drag_role, panel_rects=panel_rects, preview_scale=args.preview_scale, visual_previews=visual_previews_last, meta=raw_meta_last, beauty_preview=beauty_preview, ) if last_msg and (time.time() - last_msg_t) < 2.5: cv2.putText(board, last_msg, (18, board.shape[0] - 20), cv2.FONT_HERSHEY_SIMPLEX, 0.65, (0, 255, 0), 2, cv2.LINE_AA) cv2.imshow(window_name, board) else: blank = np.zeros((720, 1280, 3), dtype=np.uint8) overlay_hud(blank, ["Aguardando frames..."], x=40, y=80, font_scale=1.0) cv2.imshow(window_name, blank) k = cv2.waitKey(1) & 0xFF if k in (ord("q"), ord("Q"), 27): break elif k in (ord("m"), ord("M")): mode = "patches" if mode == "global" else "global" if mode == "global": set_active_profile_name(data, "global_scene_mode") else: set_active_profile_name(data, "three_reference_patches_mode") update_root_radiometric_config(data) selected_roi_index = 0 last_msg = f"Modo -> {mode} | active_profile={data['active_profile']}" last_msg_t = time.time() elif k == ord("1"): mode = "patches" selected_target = "black" selected_roi_index = 0 set_active_profile_name(data, "three_reference_patches_mode") update_root_radiometric_config(data) last_msg = "Selecionado: BLACK" last_msg_t = time.time() elif k == ord("2"): mode = "patches" selected_target = "gray" selected_roi_index = 0 set_active_profile_name(data, "three_reference_patches_mode") update_root_radiometric_config(data) last_msg = "Selecionado: GRAY" last_msg_t = time.time() elif k == ord("3"): mode = "patches" selected_target = "white" selected_roi_index = 0 set_active_profile_name(data, "three_reference_patches_mode") update_root_radiometric_config(data) last_msg = "Selecionado: WHITE" last_msg_t = time.time() elif k in (ord("s"), ord("S")): cfg = data.get("global_scene_mode", {}).get("radiometric_config", {}) curr = str(cfg.get("spectral_control_mode", "shared")).lower() set_shared_mode(data, shared=(curr != "shared")) new_mode = ( data.get("global_scene_mode", {}) .get("radiometric_config", {}) .get("spectral_control_mode", "shared") ) last_msg = f"spectral_control_mode -> {new_mode}" last_msg_t = time.time() elif k in (ord("r"), ord("R")): data = { "schema": "multispec_radiometric_config_profiles_v3", "saved_at": now_str(), "active_profile": "global_scene_mode", "global_scene_mode": default_profile_global(), "three_reference_patches_mode": default_profile_patches(), "patch_normalization": { "enabled": True, "apply_when_metering_mode": "reference_patches", "apply_stage": "after_fusion", "method": "gray_scale_with_white_guard", "space": "multispec_tensor", "targets_by_patch_channel": { "black": { "R": 0.06, "G": 0.06, "B": 0.06, "RE": 0.06, "NIR": 0.06 }, "gray": { "R": 0.34, "G": 0.34, "B": 0.34, "RE": 0.24, "NIR": 0.30 }, "white": { "R": 0.78, "G": 0.78, "B": 0.78, "RE": 0.78, "NIR": 0.78 } }, "white_guard_max": 0.92, "white_guard_max_by_channel": { "R": 0.92, "G": 0.92, "B": 0.92, "RE": 0.88, "NIR": 0.88 }, "scale_min": 0.35, "scale_max": 2.50, "clip_output": True, "require_valid_gray": True, "use_black_for_offset": False, "save_patch_stats": True } } update_root_radiometric_config(data) last_msg = "Defaults restaurados" last_msg_t = time.time() elif k in (ord("p"), ord("P"), 32): save_config(args.out_json, data) last_msg = f"Salvo em: {args.out_json}" last_msg_t = time.time() print(f"[OK] radiometric config salvo em: {args.out_json}") elif k in (ord("c"), ord("C")): idx = ROLES.index(edit_role) if edit_role in ROLES else 0 edit_role = ROLES[(idx + 1) % len(ROLES)] selected_roi_index = 0 last_msg = f"Camera editada -> {edit_role.upper()}" last_msg_t = time.time() elif mode == "patches" and k in (ord("n"), ord("N")): selected_roi_index = add_empty_patch_roi_slot(data, selected_target, edit_role) last_msg = f"Nova ROI {selected_target.upper()}/{edit_role.upper()} #{selected_roi_index + 1}. Arraste para posicionar." last_msg_t = time.time() elif mode == "patches" and k in (ord("["), ord(",")): entries = get_patch_roi_entries_for_role(data, selected_target, edit_role, enabled_only=False) if entries: selected_roi_index = (selected_roi_index - 1) % len(entries) last_msg = f"ROI selecionada -> #{selected_roi_index + 1}/{len(entries)}" else: last_msg = "Nenhuma ROI para selecionar" last_msg_t = time.time() elif mode == "patches" and k in (ord("]"), ord(".")): entries = get_patch_roi_entries_for_role(data, selected_target, edit_role, enabled_only=False) if entries: selected_roi_index = (selected_roi_index + 1) % len(entries) last_msg = f"ROI selecionada -> #{selected_roi_index + 1}/{len(entries)}" else: last_msg = "Nenhuma ROI para selecionar" last_msg_t = time.time() elif mode == "patches" and k in (ord("d"), ord("D")): ok, n_left = delete_patch_roi_for_role(data, selected_target, edit_role, selected_roi_index) selected_roi_index = clamp(selected_roi_index, 0, max(0, n_left - 1)) last_msg = f"ROI apagada. Restam {n_left}." if ok else "Nenhuma ROI para apagar" last_msg_t = time.time() elif mode == "patches" and k in (ord("t"), ord("T")): ok, enabled = toggle_patch_roi_enabled_for_role(data, selected_target, edit_role, selected_roi_index) last_msg = f"ROI #{selected_roi_index + 1} enabled={enabled}" if ok else "Nenhuma ROI para alternar" last_msg_t = time.time() elif k in (ord("v"), ord("V")): beauty_preview = not beauty_preview last_msg = f"Beauty Preview -> {beauty_preview}" last_msg_t = time.time() finally: cv2.destroyAllWindows() print("Fim da parametrizacao radiometrica.") if __name__ == "__main__": main()