agrobot_base/Python/OAK/datasets/oak-fcc-3/utils/radiometric_config_tool.py

1618 lines
63 KiB
Python
Raw Normal View History

2026-05-06 13:24:17 +00:00
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
2026-05-06 13:24:17 +00:00
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}")
2026-05-06 13:24:17 +00:00
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
# ============================================================
# Defaults espelhados do module_params.json atual.
# Este bloco é a "semente boa" do AE Rad: patches + Global Saturation Guard + Sun Guard.
DEFAULT_RADIOMETRIC_CONFIG = {'enabled': True,
'interval_s': 0.25,
'verbose': True,
'metering_mode': 'reference_patches',
'spectral_control_mode': 'shared',
'control_metric': 'p50',
'target_value': 0.5,
'deadband': 0.035,
'p95_limit': 0.94,
'saturation_limit_pct': 0.5,
'alpha': 0.18,
'exp_step_gain': 0.55,
'prefer_exposure': True,
'exp_min_us': 100,
'exp_max_us': 80000,
'gain_min': 1.0,
'gain_max': 4.0,
'reference_patches': [{'name': 'black_reference',
'type': 'black',
'roles': ['rgb', 're', 'nir'],
'roi_pct': {'x0': 0.365625, 'y0': 0.8225, 'x1': 0.432812, 'y1': 0.995},
'target_value': 0.08,
'weight': 0.7,
'roi_pct_by_role': {'rgb': {'x0': 0.365625, 'y0': 0.8225, 'x1': 0.432812, 'y1': 0.995},
're': {'x0': 0.395313, 'y0': 0.745, 'x1': 0.4625, 'y1': 0.9225},
'nir': {'x0': 0.353125, 'y0': 0.78, 'x1': 0.420312, 'y1': 0.9525}},
'roi_list_by_role': {'rgb': [{'name': 'rgb_legacy_01',
'enabled': True,
'roi_pct': {'x0': 0.365625,
'y0': 0.8225,
'x1': 0.432812,
'y1': 0.995},
'created_at': '2026-05-07 14:36:25',
'updated_at': '2026-05-08 09:19:46'}],
're': [{'name': 're_legacy_01',
'enabled': True,
'roi_pct': {'x0': 0.395313,
'y0': 0.745,
'x1': 0.4625,
'y1': 0.9225},
'created_at': '2026-05-07 14:36:25',
'updated_at': '2026-05-08 09:20:14'}],
'nir': [{'name': 'nir_legacy_01',
'enabled': True,
'roi_pct': {'x0': 0.353125,
'y0': 0.78,
'x1': 0.420312,
'y1': 0.9525},
'created_at': '2026-05-07 14:36:25',
'updated_at': '2026-05-08 09:20:47'}]}},
{'name': 'gray_reference',
'type': 'gray',
'roles': ['rgb', 're', 'nir'],
'roi_pct': {'x0': 0.29375, 'y0': 0.825, 'x1': 0.3625, 'y1': 0.995},
'target_value': 0.35,
'target_value_by_role': {'rgb': 0.34, 're': 0.24, 'nir': 0.3},
'weight': 1.0,
'roi_pct_by_role': {'rgb': {'x0': 0.29375, 'y0': 0.825, 'x1': 0.3625, 'y1': 0.995},
're': {'x0': 0.325, 'y0': 0.75, 'x1': 0.389062, 'y1': 0.915},
'nir': {'x0': 0.284375, 'y0': 0.79, 'x1': 0.35, 'y1': 0.9525}},
'roi_list_by_role': {'rgb': [{'name': 'rgb_legacy_01',
'enabled': True,
'roi_pct': {'x0': 0.29375,
'y0': 0.825,
'x1': 0.3625,
'y1': 0.995},
'created_at': '2026-05-07 14:36:25',
'updated_at': '2026-05-08 09:19:29'}],
're': [{'name': 're_legacy_01',
'enabled': True,
'roi_pct': {'x0': 0.325, 'y0': 0.75, 'x1': 0.389062, 'y1': 0.915},
'created_at': '2026-05-07 14:36:25',
'updated_at': '2026-05-08 09:20:22'}],
'nir': [{'name': 'nir_legacy_01',
'enabled': True,
'roi_pct': {'x0': 0.284375, 'y0': 0.79, 'x1': 0.35, 'y1': 0.9525},
'created_at': '2026-05-07 14:36:25',
'updated_at': '2026-05-08 09:20:53'}]}},
{'name': 'white_reference',
'type': 'white',
'roles': ['rgb', 're', 'nir'],
'roi_pct': {'x0': 0.220312, 'y0': 0.8225, 'x1': 0.2875, 'y1': 0.995},
'target_value': 0.82,
'weight': 0.8,
'roi_pct_by_role': {'rgb': {'x0': 0.220312, 'y0': 0.8225, 'x1': 0.2875, 'y1': 0.995},
're': {'x0': 0.25, 'y0': 0.7525, 'x1': 0.315625, 'y1': 0.9225},
'nir': {'x0': 0.214062, 'y0': 0.785, 'x1': 0.282813, 'y1': 0.9525}},
'roi_list_by_role': {'rgb': [{'name': 'rgb_legacy_01',
'enabled': True,
'roi_pct': {'x0': 0.220312,
'y0': 0.8225,
'x1': 0.2875,
'y1': 0.995},
'created_at': '2026-05-07 14:36:25',
'updated_at': '2026-05-08 09:19:04'}],
're': [{'name': 're_legacy_01',
'enabled': True,
'roi_pct': {'x0': 0.25,
'y0': 0.7525,
'x1': 0.315625,
'y1': 0.9225},
'created_at': '2026-05-07 14:36:25',
'updated_at': '2026-05-08 09:20:27'}],
'nir': [{'name': 'nir_legacy_01',
'enabled': True,
'roi_pct': {'x0': 0.214062,
'y0': 0.785,
'x1': 0.282813,
'y1': 0.9525},
'created_at': '2026-05-07 14:36:25',
'updated_at': '2026-05-08 09:21:02'}]}}],
'exp_apply_threshold_us': 80,
'gain_apply_threshold': 0.05,
'apply_same_spectral_to_both': True,
'spectral_roles': ['re', 'nir'],
'dark_limit_pct': 35.0,
'control_strategy': 'ratio',
'ratio_alpha': 0.35,
'ratio_min': 0.65,
'ratio_max': 1.35,
'reduce_fast_factor': 0.8,
'factor_min': 0.55,
'factor_max': 1.28,
'gain_return_enabled': True,
'gain_reduce_on_saturation': True,
'gain_increase_required_cycles': 5,
'gain_decrease_required_cycles': 2,
'gain_step_up': 0.2,
'gain_step_down': 0.5,
'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}},
'ready_required_cycles': 3,
'patch_control_mode': 'gray_primary',
'patch_require_order': True,
'patch_min_separation': 0.08,
'patch_white_sat_limit_pct': 0.5,
'patch_white_p95_limit': 0.94,
'patch_black_dark_limit_pct': 80.0,
'patch_black_max_p50': 0.2,
'patch_gray_min_p50': 0.08,
'patch_gray_max_p50': 0.85,
'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,
'global_saturation_guard_enabled': True,
'global_guard_roi_pct': {'x0': 0.05, 'y0': 0.05, 'x1': 0.95, 'y1': 0.95},
'global_guard_sat_threshold': 0.985,
'global_guard_near_sat_threshold': 0.94,
'global_guard_sat_pct_soft': 0.05,
'global_guard_sat_pct_hard': 0.2,
'global_guard_sat_pct_extreme': 0.8,
'global_guard_blob_pct_soft': 0.015,
'global_guard_blob_pct_hard': 0.08,
'global_guard_blob_pct_extreme': 0.25,
'global_guard_min_blob_px': 48,
'global_guard_downsample_max_side': 320,
'global_guard_reduce_factor_soft': 0.82,
'global_guard_reduce_factor_hard': 0.6,
'global_guard_reduce_factor_extreme': 0.35,
'sun_guard_enabled': True,
'sun_guard_p99_threshold': 0.9,
'sun_guard_near_sat_pct_threshold': 0.8,
'sun_guard_freeze_increase_cycles': 2,
'sun_guard_allow_decrease': True,
'guard_force_apply_enabled': True,
'guard_force_apply_soft': True,
'guard_force_apply_hard': True,
'guard_force_apply_extreme': True,
'guard_force_apply_on_patch_saturation': True,
'guard_freeze_cycles_soft': 3,
'guard_freeze_cycles_hard': 5,
'guard_freeze_cycles_extreme': 8,
'guard_reapply_min_exp_on_emergency': True,
'guard_min_exp_margin_us': 80}
DEFAULT_PATCH_NORMALIZATION = {'enabled': True,
'apply_when_metering_mode': 'reference_patches',
'apply_stage': 'after_fusion',
'method': 'gray_scale_with_white_guard',
'space': 'multispec_tensor',
'targets': {'black': 0.06, 'gray': 0.4, 'white': 0.78},
'white_guard_max': 0.92,
'scale_min': 0.35,
'scale_max': 2.5,
'clip_output': True,
'require_valid_gray': True,
'use_black_for_offset': False,
'save_patch_stats': True}
PROFILE_SCHEMA = "multispec_radiometric_config_profiles_v4"
MODULE_PARAMS_SCHEMA = "multispec_module_params_v3"
def deep_clone(obj):
return json.loads(json.dumps(obj))
def is_module_params_contract(data: dict) -> bool:
"""Detecta o contrato completo do module_params.json, para não salvar wrappers do tool nele."""
if not isinstance(data, dict):
return False
schema = str(data.get("schema", ""))
if schema == MODULE_PARAMS_SCHEMA:
return True
module_keys = ("camera_settings", "fusion_config", "rgb_calibration", "flatfield_config")
return "radiometric_config" in data and any(k in data for k in module_keys)
def sanitize_radiometric_config(cfg: dict) -> dict:
"""Garante que o radiometric_config salvo siga o contrato runtime atual."""
out = deep_clone(DEFAULT_RADIOMETRIC_CONFIG)
if isinstance(cfg, dict):
# Preserva valores/ROIs escolhidos no tool, mas injeta qualquer chave nova faltante.
for k, v in cfg.items():
out[k] = v
out.setdefault("reference_patches", deep_clone(DEFAULT_RADIOMETRIC_CONFIG.get("reference_patches", [])))
# Garante contrato multi_roi_by_role_v1 em todos os patches.
out["patch_roi_contract"] = "multi_roi_by_role_v1"
for patch in out.get("reference_patches", []) or []:
if not isinstance(patch, dict):
continue
patch.setdefault("roles", ROLES[:] if "ROLES" in globals() else ["rgb", "re", "nir"])
patch.setdefault("roi_pct", {})
patch.setdefault("roi_pct_by_role", {"rgb": {}, "re": {}, "nir": {}})
patch.setdefault("roi_list_by_role", {"rgb": [], "re": [], "nir": []})
ensure_patch_roi_lists_by_role(patch)
# Garante guardas parrudas mesmo em arquivos antigos.
for k, v in DEFAULT_RADIOMETRIC_CONFIG.items():
if k.startswith("global_guard_") or k.startswith("sun_guard_") or k.startswith("guard_"):
out.setdefault(k, v)
return out
def base_ae_contract():
cfg = deep_clone(DEFAULT_RADIOMETRIC_CONFIG)
# Removemos somente campos específicos de patches quando usado como base global.
cfg.pop("reference_patches", None)
cfg.pop("patch_control_mode", None)
cfg.pop("patch_require_order", None)
cfg.pop("patch_min_separation", None)
cfg.pop("patch_white_sat_limit_pct", None)
cfg.pop("patch_white_p95_limit", None)
cfg.pop("patch_black_dark_limit_pct", None)
cfg.pop("patch_black_max_p50", None)
cfg.pop("patch_gray_min_p50", None)
cfg.pop("patch_gray_max_p50", None)
cfg.pop("patch_roi_contract", None)
cfg.pop("patch_roi_reduce_method", None)
cfg.pop("patch_roi_outlier_reject", None)
cfg.pop("patch_roi_max_p50_delta", None)
return cfg
2026-05-06 13:24:17 +00:00
def default_profile_global():
cfg = base_ae_contract()
base = {"x0": 0.08, "y0": 0.08, "x1": 0.92, "y1": 0.92}
guard_base = deep_clone(DEFAULT_RADIOMETRIC_CONFIG.get("global_guard_roi_pct", {"x0": 0.05, "y0": 0.05, "x1": 0.95, "y1": 0.95}))
2026-05-06 13:24:17 +00:00
cfg.update({
"metering_mode": "global",
"spectral_control_mode": DEFAULT_RADIOMETRIC_CONFIG.get("spectral_control_mode", "shared"),
"global_roi_pct": base,
"global_roi_pct_by_role": {"rgb": dict(base), "re": dict(base), "nir": dict(base)},
"global_guard_roi_pct": guard_base,
})
return {"radiometric_config": cfg}
2026-05-06 13:24:17 +00:00
def make_default_patch(patch_type: str, target: float, weight: float):
# Preferimos copiar o patch correspondente do module_params.json atual.
for p in DEFAULT_RADIOMETRIC_CONFIG.get("reference_patches", []) or []:
if str(p.get("type", "")).lower() == str(patch_type).lower():
patch = deep_clone(p)
patch.setdefault("target_value", target)
patch.setdefault("weight", weight)
patch.setdefault("roles", ROLES[:] if "ROLES" in globals() else ["rgb", "re", "nir"])
patch.setdefault("roi_pct", {})
patch.setdefault("roi_pct_by_role", {"rgb": {}, "re": {}, "nir": {}})
patch.setdefault("roi_list_by_role", {"rgb": [], "re": [], "nir": []})
return patch
return {
"name": f"{patch_type}_reference",
"type": patch_type,
"roles": ROLES[:] if "ROLES" in globals() else ["rgb", "re", "nir"],
"target_value": target,
"weight": weight,
"roi_pct": {},
"roi_pct_by_role": {"rgb": {}, "re": {}, "nir": {}},
"roi_list_by_role": {"rgb": [], "re": [], "nir": []},
}
2026-05-06 13:24:17 +00:00
def default_profile_patches():
cfg = sanitize_radiometric_config(DEFAULT_RADIOMETRIC_CONFIG)
cfg["metering_mode"] = "reference_patches"
cfg["spectral_control_mode"] = DEFAULT_RADIOMETRIC_CONFIG.get("spectral_control_mode", "shared")
return {"radiometric_config": cfg}
2026-05-06 13:24:17 +00:00
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)
2026-05-06 13:24:17 +00:00
def load_or_default_config(path: str):
"""
Carrega tanto:
1) calibration/module_params.json completo, contrato multispec_module_params_v3;
2) arquivo isolado do tool com perfis.
Em ambos os casos, o root radiometric_config é mantido no mesmo contrato do runtime.
"""
2026-05-06 13:24:17 +00:00
if path and os.path.isfile(path):
with open(path, "r", encoding="utf-8") as f:
data = json.load(f)
else:
data = {}
module_contract = is_module_params_contract(data)
root_cfg = data.get("radiometric_config") if isinstance(data.get("radiometric_config"), dict) else None
if module_contract:
# Não troca o schema do module_params. Apenas cria perfis internos para a UI.
data.setdefault("schema", MODULE_PARAMS_SCHEMA)
else:
data.setdefault("schema", PROFILE_SCHEMA)
2026-05-06 13:24:17 +00:00
data.setdefault("saved_at", now_str())
# Se já existe um radiometric_config na raiz, ele é a fonte da verdade.
if root_cfg:
root_cfg = sanitize_radiometric_config(root_cfg)
active = "three_reference_patches_mode" if str(root_cfg.get("metering_mode", "")).lower() == "reference_patches" else "global_scene_mode"
data["active_profile"] = active
data.setdefault("global_scene_mode", default_profile_global())
data.setdefault("three_reference_patches_mode", default_profile_patches())
data[active]["radiometric_config"] = root_cfg
else:
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", deep_clone(DEFAULT_PATCH_NORMALIZATION))
# Migração: injeta chaves novas nos dois perfis sem sobrescrever ROIs existentes.
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, deep_clone(v))
if profile_name == "three_reference_patches_mode":
data[profile_name]["radiometric_config"] = sanitize_radiometric_config(cfg)
update_root_radiometric_config(data)
2026-05-06 13:24:17 +00:00
return data
def save_config(path: str, data: dict):
ensure_dir(os.path.dirname(path) or ".")
module_contract = is_module_params_contract(data)
2026-05-06 13:24:17 +00:00
# Atualiza radiometric_config root a partir do perfil ativo, usando o contrato runtime atual.
update_root_radiometric_config(data)
data["radiometric_config"] = sanitize_radiometric_config(data.get("radiometric_config", {}))
data["patch_normalization"] = data.get("patch_normalization") or deep_clone(DEFAULT_PATCH_NORMALIZATION)
data["saved_at"] = now_str()
if module_contract:
# Salva limpo no contrato multispec_module_params_v3, sem wrappers internos da UI.
out = dict(data)
out["schema"] = MODULE_PARAMS_SCHEMA
out.pop("active_profile", None)
out.pop("global_scene_mode", None)
out.pop("three_reference_patches_mode", None)
else:
out = dict(data)
out["schema"] = PROFILE_SCHEMA
out["active_profile"] = get_active_profile_name(data)
update_root_radiometric_config(out)
out["radiometric_config"] = sanitize_radiometric_config(out.get("radiometric_config", {}))
2026-05-06 13:24:17 +00:00
with open(path, "w", encoding="utf-8") as f:
json.dump(out, f, ensure_ascii=False, indent=2)
2026-05-06 13:24:17 +00:00
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}
2026-05-06 13:24:17 +00:00
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):
2026-05-06 13:24:17 +00:00
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)
2026-05-06 13:24:17 +00:00
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 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):
2026-05-06 13:24:17 +00:00
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"],
)
2026-05-06 13:24:17 +00:00
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)
2026-05-06 13:24:17 +00:00
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
2026-05-06 13:24:17 +00:00
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 {}
2026-05-06 13:24:17 +00:00
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,
)
2026-05-06 13:24:17 +00:00
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)
2026-05-06 13:24:17 +00:00
# ============================================================
# 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
2026-05-06 13:24:17 +00:00
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)
2026-05-06 13:24:17 +00:00
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,
):
2026-05-06 13:24:17 +00:00
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.
# ------------------------------------------------------------
2026-05-06 13:24:17 +00:00
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]
2026-05-06 13:24:17 +00:00
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)
2026-05-06 13:24:17 +00:00
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)
2026-05-06 13:24:17 +00:00
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)
2026-05-06 13:24:17 +00:00
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)
2026-05-06 13:24:17 +00:00
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)
2026-05-06 13:24:17 +00:00
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')}",
2026-05-06 13:24:17 +00:00
"",
"Arraste no painel da camera editada para definir/atualizar ROI.",
"PATCHES agora suportam N ROIs por cor e por camera.",
2026-05-06 13:24:17 +00:00
"",
]
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))
2026-05-06 13:24:17 +00:00
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
2026-05-06 13:24:17 +00:00
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.")
2026-05-06 13:24:17 +00:00
sel_patch = get_patch_by_type(data, selected_target)
2026-05-06 13:24:17 +00:00
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))
2026-05-06 13:24:17 +00:00
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",
2026-05-06 13:24:17 +00:00
])
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):
2026-05-06 13:24:17 +00:00
lines = []
for role in ROLES:
2026-05-06 13:24:17 +00:00
_, 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 += ",..."
2026-05-06 13:24:17 +00:00
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}%"
2026-05-06 13:24:17 +00:00
)
lines.append(f" p50_each=[{mini}]")
2026-05-06 13:24:17 +00:00
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 = {}
2026-05-06 13:24:17 +00:00
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
2026-05-06 13:24:17 +00:00
# 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):
2026-05-06 13:24:17 +00:00
return
lx, ly = local_from_rect(edit_rect, x, y)
2026-05-06 13:24:17 +00:00
if event == cv2.EVENT_LBUTTONDOWN:
dragging = True
drag_role = edit_role
2026-05-06 13:24:17 +00:00
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:
2026-05-06 13:24:17 +00:00
return
_, _, x1, y1 = edit_rect
x0r, y0r, _, _ = edit_rect
2026-05-06 13:24:17 +00:00
w = x1 - x0r
h = y1 - y0r
roi_pct = px_to_pct(rect, w, h)
role_to_save = normalize_role(drag_role or edit_role)
2026-05-06 13:24:17 +00:00
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}"
2026-05-06 13:24:17 +00:00
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}"
)
2026-05-06 13:24:17 +00:00
drag_role = None
2026-05-06 13:24:17 +00:00
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:
2026-05-06 16:27:38 +00:00
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}")
2026-05-06 13:24:17 +00:00
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
2026-05-06 13:24:17 +00:00
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,
2026-05-06 13:24:17 +00:00
drag_rect_local=drag_rect_local,
drag_role=drag_role,
2026-05-06 13:24:17 +00:00
panel_rects=panel_rects,
preview_scale=args.preview_scale,
visual_previews=visual_previews_last,
meta=raw_meta_last,
beauty_preview=beauty_preview,
2026-05-06 13:24:17 +00:00
)
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']}"
2026-05-06 13:24:17 +00:00
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)
2026-05-06 13:24:17 +00:00
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)
2026-05-06 13:24:17 +00:00
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)
2026-05-06 13:24:17 +00:00
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")):
# Restaura somente a parte radiométrica, preservando o restante do module_params quando existir.
module_contract = is_module_params_contract(data)
preserved = dict(data) if module_contract else {}
if module_contract:
preserved["radiometric_config"] = sanitize_radiometric_config(DEFAULT_RADIOMETRIC_CONFIG)
preserved["patch_normalization"] = deep_clone(DEFAULT_PATCH_NORMALIZATION)
preserved["active_profile"] = "three_reference_patches_mode"
preserved["global_scene_mode"] = default_profile_global()
preserved["three_reference_patches_mode"] = default_profile_patches()
data = preserved
else:
data = {
"schema": PROFILE_SCHEMA,
"saved_at": now_str(),
"active_profile": "three_reference_patches_mode",
"global_scene_mode": default_profile_global(),
"three_reference_patches_mode": default_profile_patches(),
"patch_normalization": deep_clone(DEFAULT_PATCH_NORMALIZATION),
}
update_root_radiometric_config(data)
2026-05-06 13:24:17 +00:00
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()
2026-05-06 13:24:17 +00:00
finally:
cv2.destroyAllWindows()
print("Fim da parametrizacao radiometrica.")
2026-05-06 13:24:17 +00:00
if __name__ == "__main__":
main()