796 lines
26 KiB
Python
796 lines
26 KiB
Python
|
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 validate_module_ready(status: dict, raw_policy: str):
|
||
|
|
if not status.get("ok", True):
|
||
|
|
raise RuntimeError(f"Status inválido retornado pelo módulo: {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 default_profile_global():
|
||
|
|
return {
|
||
|
|
"radiometric_config": {
|
||
|
|
"enabled": True,
|
||
|
|
"interval_s": 0.5,
|
||
|
|
"verbose": True,
|
||
|
|
|
||
|
|
"metering_mode": "global",
|
||
|
|
"spectral_control_mode": "shared",
|
||
|
|
|
||
|
|
"global_roi_pct": {
|
||
|
|
"x0": 0.08,
|
||
|
|
"y0": 0.08,
|
||
|
|
"x1": 0.92,
|
||
|
|
"y1": 0.92,
|
||
|
|
},
|
||
|
|
|
||
|
|
"control_metric": "p50",
|
||
|
|
"target_value": 0.40,
|
||
|
|
"deadband": 0.04,
|
||
|
|
|
||
|
|
"p95_limit": 0.94,
|
||
|
|
"saturation_limit_pct": 1.0,
|
||
|
|
"dark_limit_pct": 35.0,
|
||
|
|
|
||
|
|
"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,
|
||
|
|
|
||
|
|
"role_limits": {
|
||
|
|
"rgb": {"exp_min_us": 100, "exp_max_us": 80000, "gain_min": 1.0, "gain_max": 4.0},
|
||
|
|
"re": {"exp_min_us": 100, "exp_max_us": 80000, "gain_min": 1.0, "gain_max": 3.0},
|
||
|
|
"nir": {"exp_min_us": 100, "exp_max_us": 80000, "gain_min": 1.0, "gain_max": 3.0},
|
||
|
|
},
|
||
|
|
|
||
|
|
"exp_apply_threshold_us": 80,
|
||
|
|
"gain_apply_threshold": 0.05,
|
||
|
|
|
||
|
|
"apply_same_spectral_to_both": True,
|
||
|
|
"spectral_roles": ["re", "nir"],
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def default_profile_patches():
|
||
|
|
return {
|
||
|
|
"radiometric_config": {
|
||
|
|
"enabled": True,
|
||
|
|
"interval_s": 0.5,
|
||
|
|
"verbose": True,
|
||
|
|
|
||
|
|
"metering_mode": "reference_patches",
|
||
|
|
"spectral_control_mode": "shared",
|
||
|
|
|
||
|
|
"control_metric": "p50",
|
||
|
|
"target_value": 0.40,
|
||
|
|
"deadband": 0.035,
|
||
|
|
|
||
|
|
"p95_limit": 0.94,
|
||
|
|
"saturation_limit_pct": 1.0,
|
||
|
|
"dark_limit_pct": 35.0,
|
||
|
|
|
||
|
|
"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,
|
||
|
|
|
||
|
|
"role_limits": {
|
||
|
|
"rgb": {"exp_min_us": 100, "exp_max_us": 80000, "gain_min": 1.0, "gain_max": 4.0},
|
||
|
|
"re": {"exp_min_us": 100, "exp_max_us": 80000, "gain_min": 1.0, "gain_max": 3.0},
|
||
|
|
"nir": {"exp_min_us": 100, "exp_max_us": 80000, "gain_min": 1.0, "gain_max": 3.0},
|
||
|
|
},
|
||
|
|
|
||
|
|
"reference_patches": [
|
||
|
|
{
|
||
|
|
"name": "black_reference",
|
||
|
|
"type": "black",
|
||
|
|
"roles": ["rgb", "re", "nir"],
|
||
|
|
"roi_pct": {"x0": 0.05, "y0": 0.92, "x1": 0.18, "y1": 0.99},
|
||
|
|
"target_value": 0.08,
|
||
|
|
"weight": 0.7,
|
||
|
|
},
|
||
|
|
{
|
||
|
|
"name": "gray_reference",
|
||
|
|
"type": "gray",
|
||
|
|
"roles": ["rgb", "re", "nir"],
|
||
|
|
"roi_pct": {"x0": 0.35, "y0": 0.92, "x1": 0.55, "y1": 0.99},
|
||
|
|
"target_value": 0.40,
|
||
|
|
"weight": 1.0,
|
||
|
|
},
|
||
|
|
{
|
||
|
|
"name": "white_reference",
|
||
|
|
"type": "white",
|
||
|
|
"roles": ["rgb", "re", "nir"],
|
||
|
|
"roi_pct": {"x0": 0.75, "y0": 0.92, "x1": 0.95, "y1": 0.99},
|
||
|
|
"target_value": 0.82,
|
||
|
|
"weight": 0.8,
|
||
|
|
},
|
||
|
|
],
|
||
|
|
|
||
|
|
"exp_apply_threshold_us": 80,
|
||
|
|
"gain_apply_threshold": 0.05,
|
||
|
|
|
||
|
|
"apply_same_spectral_to_both": True,
|
||
|
|
"spectral_roles": ["re", "nir"],
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
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_v1")
|
||
|
|
data.setdefault("saved_at", now_str())
|
||
|
|
data.setdefault("global_scene_mode", default_profile_global())
|
||
|
|
data.setdefault("three_reference_patches_mode", default_profile_patches())
|
||
|
|
|
||
|
|
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_v1"
|
||
|
|
data["saved_at"] = now_str()
|
||
|
|
|
||
|
|
with open(path, "w", encoding="utf-8") as f:
|
||
|
|
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
|
|
|
||
|
|
|
||
|
|
def get_global_roi(data: dict):
|
||
|
|
return (
|
||
|
|
data.get("global_scene_mode", {})
|
||
|
|
.get("radiometric_config", {})
|
||
|
|
.get("global_roi_pct", {"x0": 0.08, "y0": 0.08, "x1": 0.92, "y1": 0.92})
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def set_global_roi(data: dict, roi_pct: dict):
|
||
|
|
data.setdefault("global_scene_mode", default_profile_global())
|
||
|
|
data["global_scene_mode"].setdefault("radiometric_config", default_profile_global()["radiometric_config"])
|
||
|
|
data["global_scene_mode"]["radiometric_config"]["global_roi_pct"] = roi_pct
|
||
|
|
|
||
|
|
|
||
|
|
def get_patches(data: dict):
|
||
|
|
return (
|
||
|
|
data.get("three_reference_patches_mode", {})
|
||
|
|
.get("radiometric_config", {})
|
||
|
|
.get("reference_patches", [])
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def set_patch_roi(data: dict, patch_type: str, roi_pct: dict):
|
||
|
|
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()
|
||
|
|
for p in patches:
|
||
|
|
if str(p.get("type", "")).lower() == patch_type:
|
||
|
|
p["roi_pct"] = roi_pct
|
||
|
|
return
|
||
|
|
|
||
|
|
# Fallback se não existir.
|
||
|
|
target = {"black": 0.08, "gray": 0.40, "white": 0.82}.get(patch_type, 0.40)
|
||
|
|
weight = {"black": 0.7, "gray": 1.0, "white": 0.8}.get(patch_type, 1.0)
|
||
|
|
|
||
|
|
patches.append({
|
||
|
|
"name": f"{patch_type}_reference",
|
||
|
|
"type": patch_type,
|
||
|
|
"roles": ["rgb", "re", "nir"],
|
||
|
|
"roi_pct": roi_pct,
|
||
|
|
"target_value": target,
|
||
|
|
"weight": weight,
|
||
|
|
})
|
||
|
|
|
||
|
|
|
||
|
|
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)
|
||
|
|
|
||
|
|
|
||
|
|
# ============================================================
|
||
|
|
# 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):
|
||
|
|
if mode == "global":
|
||
|
|
roi = get_global_roi(data)
|
||
|
|
draw_roi_on_panel(panel, roi, "GLOBAL", PATCH_COLORS["global"], 2)
|
||
|
|
|
||
|
|
else:
|
||
|
|
for p in get_patches(data):
|
||
|
|
typ = str(p.get("type", "")).lower()
|
||
|
|
color = PATCH_COLORS.get(typ, (0, 255, 255))
|
||
|
|
label = typ.upper()
|
||
|
|
thickness = 3 if typ == selected_target else 2
|
||
|
|
draw_roi_on_panel(panel, p.get("roi_pct"), label, color, thickness)
|
||
|
|
|
||
|
|
|
||
|
|
def build_board(decoded, data, mode, selected_target, drag_rect_local, panel_rects, preview_scale=1.0):
|
||
|
|
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")
|
||
|
|
|
||
|
|
if rgb01 is not None:
|
||
|
|
rgb_panel = to_bgr_u8_from_rgb01(rgb01)
|
||
|
|
base_h, base_w = rgb01.shape[:2]
|
||
|
|
else:
|
||
|
|
base_h, base_w = 800, 1280
|
||
|
|
rgb_panel = np.zeros((base_h, base_w, 3), dtype=np.uint8)
|
||
|
|
overlay_hud(rgb_panel, ["RGB", "sem frame"])
|
||
|
|
|
||
|
|
re01 = resize_if_needed(re01, (base_h, base_w)) if re01 is not None else None
|
||
|
|
nir01 = resize_if_needed(nir01, (base_h, base_w)) if nir01 is not None else None
|
||
|
|
|
||
|
|
re_panel = gray_to_bgr_u8(re01) if re01 is not None else np.zeros_like(rgb_panel)
|
||
|
|
nir_panel = gray_to_bgr_u8(nir01) if nir01 is not None else np.zeros_like(rgb_panel)
|
||
|
|
|
||
|
|
for p in (rgb_panel, re_panel, nir_panel):
|
||
|
|
draw_all_rois(p, data, selected_target, mode)
|
||
|
|
|
||
|
|
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))
|
||
|
|
for p in (rgb_panel, re_panel, nir_panel):
|
||
|
|
cv2.rectangle(p, (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, base_w, base_h)
|
||
|
|
overlay_hud(board, lines, x=x0 + 16, y=y0 + 28, font_scale=0.53, line_step=21)
|
||
|
|
|
||
|
|
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, base_w, base_h):
|
||
|
|
shared = (
|
||
|
|
data.get("global_scene_mode", {})
|
||
|
|
.get("radiometric_config", {})
|
||
|
|
.get("spectral_control_mode", "shared")
|
||
|
|
)
|
||
|
|
|
||
|
|
lines = [
|
||
|
|
"RADIOMETRIC CONFIG TOOL",
|
||
|
|
f"modo_edição={mode.upper()} | spectral={shared}",
|
||
|
|
"",
|
||
|
|
"Arraste com o mouse no painel RGB para definir ROI.",
|
||
|
|
"A ROI é salva em percentuais e aplicada aos 3 sensores.",
|
||
|
|
"",
|
||
|
|
]
|
||
|
|
|
||
|
|
if mode == "global":
|
||
|
|
roi_pct = get_global_roi(data)
|
||
|
|
lines.append(f"GLOBAL ROI: {roi_pct}")
|
||
|
|
lines.extend(stats_lines_for_roi(decoded, roi_pct, base_w, base_h))
|
||
|
|
else:
|
||
|
|
lines.append(f"PATCH selecionado: {selected_target.upper()}")
|
||
|
|
for p in get_patches(data):
|
||
|
|
typ = str(p.get("type", "")).lower()
|
||
|
|
roi_pct = p.get("roi_pct", {})
|
||
|
|
lines.append(
|
||
|
|
f"{typ}: target={float(p.get('target_value', 0.0)):.2f} "
|
||
|
|
f"weight={float(p.get('weight', 1.0)):.2f}"
|
||
|
|
)
|
||
|
|
lines.append(f" roi={roi_pct}")
|
||
|
|
|
||
|
|
sel_patch = None
|
||
|
|
for p in get_patches(data):
|
||
|
|
if str(p.get("type", "")).lower() == selected_target:
|
||
|
|
sel_patch = p
|
||
|
|
break
|
||
|
|
|
||
|
|
if sel_patch:
|
||
|
|
lines.append("")
|
||
|
|
lines.append(f"Stats do patch {selected_target.upper()}:")
|
||
|
|
lines.extend(stats_lines_for_roi(decoded, sel_patch.get("roi_pct", {}), base_w, base_h))
|
||
|
|
|
||
|
|
lines.extend([
|
||
|
|
"",
|
||
|
|
"M = alterna GLOBAL / 3 PATCHES",
|
||
|
|
"1/2/3 = BLACK / GRAY / WHITE",
|
||
|
|
"S = alterna spectral shared/independent",
|
||
|
|
"P ou SPACE = salva JSON",
|
||
|
|
"R = restaura defaults | Q/Esc = sai",
|
||
|
|
])
|
||
|
|
|
||
|
|
return lines
|
||
|
|
|
||
|
|
|
||
|
|
def stats_lines_for_roi(decoded, roi_pct, base_w, base_h):
|
||
|
|
if not roi_pct:
|
||
|
|
return ["sem ROI"]
|
||
|
|
|
||
|
|
lines = []
|
||
|
|
for role in ("rgb", "re", "nir"):
|
||
|
|
_, img = get_image_by_role(decoded, role)
|
||
|
|
if img is None:
|
||
|
|
lines.append(f"{role.upper()}: sem frame")
|
||
|
|
continue
|
||
|
|
|
||
|
|
h, w = img.shape[:2]
|
||
|
|
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}%"
|
||
|
|
)
|
||
|
|
|
||
|
|
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"
|
||
|
|
|
||
|
|
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
|
||
|
|
|
||
|
|
# 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)
|
||
|
|
|
||
|
|
rgb_rect = panel_rects.get("rgb")
|
||
|
|
if not rect_inside(rgb_rect, x, y):
|
||
|
|
return
|
||
|
|
|
||
|
|
lx, ly = local_from_rect(rgb_rect, x, y)
|
||
|
|
|
||
|
|
if event == cv2.EVENT_LBUTTONDOWN:
|
||
|
|
dragging = True
|
||
|
|
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 RGB.
|
||
|
|
if rgb_rect is None:
|
||
|
|
return
|
||
|
|
|
||
|
|
_, _, x1, y1 = rgb_rect
|
||
|
|
x0r, y0r, _, _ = rgb_rect
|
||
|
|
w = x1 - x0r
|
||
|
|
h = y1 - y0r
|
||
|
|
|
||
|
|
roi_pct = px_to_pct(rect, w, h)
|
||
|
|
|
||
|
|
if mode == "global":
|
||
|
|
set_global_roi(data, roi_pct)
|
||
|
|
last_msg = f"GLOBAL ROI atualizada: {roi_pct}"
|
||
|
|
else:
|
||
|
|
set_patch_roi(data, selected_target, roi_pct)
|
||
|
|
last_msg = f"{selected_target.upper()} ROI atualizada: {roi_pct}"
|
||
|
|
|
||
|
|
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, update_radiometry=False)
|
||
|
|
|
||
|
|
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
|
||
|
|
|
||
|
|
if decoded_last:
|
||
|
|
board = build_board(
|
||
|
|
decoded=decoded_last,
|
||
|
|
data=data,
|
||
|
|
mode=mode,
|
||
|
|
selected_target=selected_target,
|
||
|
|
drag_rect_local=drag_rect_local,
|
||
|
|
panel_rects=panel_rects,
|
||
|
|
preview_scale=args.preview_scale,
|
||
|
|
)
|
||
|
|
|
||
|
|
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"
|
||
|
|
last_msg = f"Modo -> {mode}"
|
||
|
|
last_msg_t = time.time()
|
||
|
|
|
||
|
|
elif k == ord("1"):
|
||
|
|
mode = "patches"
|
||
|
|
selected_target = "black"
|
||
|
|
last_msg = "Selecionado: BLACK"
|
||
|
|
last_msg_t = time.time()
|
||
|
|
|
||
|
|
elif k == ord("2"):
|
||
|
|
mode = "patches"
|
||
|
|
selected_target = "gray"
|
||
|
|
last_msg = "Selecionado: GRAY"
|
||
|
|
last_msg_t = time.time()
|
||
|
|
|
||
|
|
elif k == ord("3"):
|
||
|
|
mode = "patches"
|
||
|
|
selected_target = "white"
|
||
|
|
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_v1",
|
||
|
|
"saved_at": now_str(),
|
||
|
|
"global_scene_mode": default_profile_global(),
|
||
|
|
"three_reference_patches_mode": default_profile_patches(),
|
||
|
|
}
|
||
|
|
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}")
|
||
|
|
|
||
|
|
finally:
|
||
|
|
cv2.destroyAllWindows()
|
||
|
|
print("Fim da parametrização radiométrica.")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|