diff --git a/Python/OAK/datasets/oak-fcc-3/core/radiometric_controller.py b/Python/OAK/datasets/oak-fcc-3/core/radiometric_controller.py index ffdfdf4d0..cd9d8e509 100644 --- a/Python/OAK/datasets/oak-fcc-3/core/radiometric_controller.py +++ b/Python/OAK/datasets/oak-fcc-3/core/radiometric_controller.py @@ -82,6 +82,15 @@ class RadiometricController: self.reference_patches = cfg.get("reference_patches", []) or [] self.patch_aggregation = str(cfg.get("patch_aggregation", "weighted_mean")).lower() + # Novo contrato de ROIs dinâmicas por patch/cor/câmera. + # Compatível com o formato antigo: roi_pct_by_role e roi_pct continuam + # servindo como fallback quando a lista dinâmica não existir. + self.patch_roi_contract = str(cfg.get("patch_roi_contract", "legacy_single_roi")).lower() + self.patch_roi_reduce_method = str(cfg.get("patch_roi_reduce_method", "median_valid_rois")).lower() + self.patch_roi_outlier_reject = bool(cfg.get("patch_roi_outlier_reject", True)) + self.patch_roi_max_p50_delta = float(cfg.get("patch_roi_max_p50_delta", 0.12)) + self.patch_roi_min_valid_rois = int(cfg.get("patch_roi_min_valid_rois", 1)) + self.patch_control_mode = str(cfg.get("patch_control_mode", "gray_primary")).lower() self.patch_require_order = bool(cfg.get("patch_require_order", True)) @@ -231,20 +240,82 @@ class RadiometricController: return self._safe_roi_pct(self.global_roi_pct) - def _get_patch_roi_pct_for_role(self, patch: dict, role: str): - role = self._normalize_role(role) + def _get_patch_roi_items_for_role(self, patch: dict, role: str) -> list[dict]: + """ + Retorna uma lista normalizada de ROIs para um patch/cor em uma câmera. + Contrato novo: + patch["roi_list_by_role"][role] = [ + {"name": "gray_rgb_01", "enabled": True, "roi_pct": {...}}, + ... + ] + + Fallbacks legados: + patch["roi_pct_by_role"][role] + patch["roi_pct"] + """ + role = self._normalize_role(role) + items: list[dict] = [] + + list_by_role = patch.get("roi_list_by_role", {}) + raw_items = [] + if isinstance(list_by_role, dict): + role_items = list_by_role.get(role) + if isinstance(role_items, list): + raw_items = role_items + + for idx, item in enumerate(raw_items, start=1): + if not isinstance(item, dict): + continue + if item.get("enabled", True) is False: + continue + roi = item.get("roi_pct") + if not isinstance(roi, dict) or not roi: + continue + items.append({ + "name": str(item.get("name") or f"{patch.get('type', 'patch')}_{role}_{idx:02d}"), + "enabled": True, + "roi_pct": self._safe_roi_pct(roi), + "roi_source": "roi_list_by_role", + "index": int(idx - 1), + }) + + if items: + return items + + # Fallback 1: contrato antigo por câmera. by_role = patch.get("roi_pct_by_role", {}) if isinstance(by_role, dict): roi = by_role.get(role) if isinstance(roi, dict) and roi: - return self._safe_roi_pct(roi), "roi_pct_by_role" + return [{ + "name": f"{patch.get('type', 'patch')}_{role}_legacy_by_role", + "enabled": True, + "roi_pct": self._safe_roi_pct(roi), + "roi_source": "roi_pct_by_role", + "index": 0, + }] + # Fallback 2: contrato antigo global do patch. legacy = patch.get("roi_pct") if isinstance(legacy, dict) and legacy: - return self._safe_roi_pct(legacy), "roi_pct" + return [{ + "name": f"{patch.get('type', 'patch')}_{role}_legacy", + "enabled": True, + "roi_pct": self._safe_roi_pct(legacy), + "roi_source": "roi_pct", + "index": 0, + }] - return None, "missing" + return [] + + def _get_patch_roi_pct_for_role(self, patch: dict, role: str): + """Compatibilidade para código legado: retorna a primeira ROI ativa.""" + items = self._get_patch_roi_items_for_role(patch, role) + if not items: + return None, "missing" + item = items[0] + return item["roi_pct"], item.get("roi_source", "roi_list_by_role") def get_patch_target_for_role(self, patch, role, fallback): by_role = patch.get("target_value_by_role", {}) @@ -516,20 +587,10 @@ class RadiometricController: if role not in roles and "all" not in roles: continue - roi_pct, roi_source = self._get_patch_roi_pct_for_role(patch, role) - - if not isinstance(roi_pct, dict): + roi_items = self._get_patch_roi_items_for_role(patch, role) + if not roi_items: continue - x0 = float(roi_pct.get("x0", 0.0)) - y0 = float(roi_pct.get("y0", 0.0)) - x1 = float(roi_pct.get("x1", 1.0)) - y1 = float(roi_pct.get("y1", 1.0)) - - roi = self._roi_pct_to_pixels(h, w, x0, y0, x1, y1) - arr = self._crop_array(img_gray, roi) - stats = self.compute_stats(arr) - target = self.get_patch_target_for_role( patch=patch, role=role, @@ -538,16 +599,59 @@ class RadiometricController: if target is not None: target = float(target) + roi_results = [] + for roi_item in roi_items: + roi_pct = roi_item.get("roi_pct") + if not isinstance(roi_pct, dict): + continue + + x0 = float(roi_pct.get("x0", 0.0)) + y0 = float(roi_pct.get("y0", 0.0)) + x1 = float(roi_pct.get("x1", 1.0)) + y1 = float(roi_pct.get("y1", 1.0)) + + roi = self._roi_pct_to_pixels(h, w, x0, y0, x1, y1) + arr = self._crop_array(img_gray, roi) + stats = self.compute_stats(arr) + + roi_results.append({ + "name": roi_item.get("name"), + "enabled": bool(roi_item.get("enabled", True)), + "role": role, + "roi": list(roi), + "roi_pct": {"x0": x0, "y0": y0, "x1": x1, "y1": y1}, + "roi_source": roi_item.get("roi_source", "roi_list_by_role"), + "index": int(roi_item.get("index", len(roi_results))), + "stats": stats, + }) + + if not roi_results: + continue + + patch_stats, roi_quality = self.aggregate_roi_metrics( + roi_results=roi_results, + patch_type=str(patch.get("type", "reference")).lower(), + ) + + first_roi = next( + (r for r in roi_results if r.get("stats", {}).get("valid")), + roi_results[0], + ) + patch_results.append({ "name": patch.get("name", f"patch_{len(patch_results) + 1}"), "type": patch.get("type", "reference"), "role": role, - "roi": list(roi), - "roi_pct": {"x0": x0, "y0": y0, "x1": x1, "y1": y1}, - "roi_source": roi_source, + "roi": first_roi.get("roi", []), + "roi_pct": first_roi.get("roi_pct", {}), + "roi_source": "roi_list_by_role" if len(roi_results) > 1 else first_roi.get("roi_source", "roi_list_by_role"), + "roi_count": int(len(roi_results)), + "roi_valid_count": int(roi_quality.get("valid_count", 0)), + "roi_results": roi_results, + "roi_quality": roi_quality, "weight": float(patch.get("weight", 1.0)), "target_value": target, - "stats": stats, + "stats": patch_stats, }) if not patch_results: @@ -561,6 +665,120 @@ class RadiometricController: return metrics + def aggregate_roi_metrics(self, roi_results: list[dict], patch_type: str = "reference") -> tuple[dict, dict]: + """ + Agrega várias ROIs da mesma cor/câmera em uma métrica única. + + O controle usa uma estatística robusta, normalmente mediana dos p50. + As guardas de proteção usam p95/saturação máximos para não ignorar + uma ROI branca estourada, mesmo quando outra ROI está boa. + """ + valid = [r for r in roi_results if r.get("stats", {}).get("valid")] + warnings = [] + + if not valid: + 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, + }, { + "valid": False, + "warnings": ["no_valid_roi"], + "roi_count": len(roi_results), + "valid_count": 0, + "used_count": 0, + "rejected_count": 0, + "p50_values": [], + } + + p50_values = np.array([float(r["stats"].get("p50", 0.0)) for r in valid], dtype=np.float32) + median_p50 = float(np.median(p50_values)) + spread_p50 = float(np.max(p50_values) - np.min(p50_values)) if len(p50_values) > 1 else 0.0 + + used = valid + rejected = [] + + if self.patch_roi_outlier_reject and len(valid) >= 3: + kept = [] + for r in valid: + p50 = float(r["stats"].get("p50", 0.0)) + if abs(p50 - median_p50) <= self.patch_roi_max_p50_delta: + kept.append(r) + else: + rejected.append(r) + if kept: + used = kept + if rejected: + warnings.append(f"roi_outliers_rejected:{len(rejected)}") + elif self.patch_roi_outlier_reject and len(valid) == 2 and spread_p50 > self.patch_roi_max_p50_delta: + warnings.append(f"roi_p50_spread_high:{spread_p50:.3f}") + + if len(used) < self.patch_roi_min_valid_rois: + warnings.append(f"valid_roi_count_low:{len(used)}<{self.patch_roi_min_valid_rois}") + + def arr(key: str) -> np.ndarray: + return np.array([float(r["stats"].get(key, 0.0)) for r in used], dtype=np.float32) + + means = arr("mean") + stds = arr("std") + p05s = arr("p05") + p50s = arr("p50") + p95s = arr("p95") + sats = arr("sat_pct") + darks = arr("dark_pct") + pixels = int(sum(int(r["stats"].get("pixels", 0)) for r in used)) + + reduce_method = self.patch_roi_reduce_method + if reduce_method not in ("median_valid_rois", "mean_valid_rois"): + reduce_method = "median_valid_rois" + + reducer = np.mean if reduce_method == "mean_valid_rois" else np.median + + # Para brilho/controle usamos redução robusta; para guardas de risco, + # mantemos máximos de p95/saturação. Assim um cartão branco saturado + # ainda força redução de exposição. + stats = { + "valid": True, + "pixels": pixels, + "mean": float(reducer(means)), + "std": float(reducer(stds)), + "p05": float(reducer(p05s)), + "p50": float(reducer(p50s)), + "p95": float(np.max(p95s)), + "sat_pct": float(np.max(sats)), + "dark_pct": float(reducer(darks)), + "roi_reduce_method": reduce_method, + "roi_count": int(len(roi_results)), + "roi_valid_count": int(len(valid)), + "roi_used_count": int(len(used)), + "roi_rejected_count": int(len(rejected)), + "roi_p50_values": [float(v) for v in p50_values.tolist()], + "roi_p50_spread": spread_p50, + } + + quality = { + "valid": bool(len(used) >= self.patch_roi_min_valid_rois), + "warnings": warnings, + "roi_count": int(len(roi_results)), + "valid_count": int(len(valid)), + "used_count": int(len(used)), + "rejected_count": int(len(rejected)), + "p50_values": [float(v) for v in p50_values.tolist()], + "p50_spread": spread_p50, + "reduce_method": reduce_method, + "used_names": [str(r.get("name")) for r in used], + "rejected_names": [str(r.get("name")) for r in rejected], + "patch_type": str(patch_type).lower(), + } + + return stats, quality + def aggregate_patch_metrics(self, patch_results: list[dict]) -> dict: valid = [p for p in patch_results if p["stats"].get("valid")] @@ -588,6 +806,14 @@ class RadiometricController: white = self._find_patch_result(valid, "white") warnings = [] + roi_quality_by_type = {} + + for p in valid: + ptype = str(p.get("type", "reference")).lower() + rq = p.get("roi_quality", {}) or {} + roi_quality_by_type[ptype] = rq + for warn in rq.get("warnings", []) or []: + warnings.append(f"{ptype}_{warn}") # Stats gerais de proteção. p95s = np.array([p["stats"]["p95"] for p in valid], dtype=np.float32) @@ -741,6 +967,7 @@ class RadiometricController: "black_target": black_target, "gray_target": gray_target, "white_target": white_target, + "roi_quality_by_type": roi_quality_by_type, }, } @@ -1187,6 +1414,7 @@ class RadiometricController: f" [PATCH] role={p.get('role', role)} " f"type={p.get('type')} " f"roi_source={p.get('roi_source')} " + f"roi_count={p.get('roi_valid_count', p.get('roi_count', 1))}/{p.get('roi_count', 1)} " f"roi_pct={p.get('roi_pct')} " f"p50={st.get('p50', 0):.3f} " f"p95={st.get('p95', 0):.3f} " diff --git a/Python/OAK/datasets/oak-fcc-3/core/radiometric_controller_bkp.py b/Python/OAK/datasets/oak-fcc-3/core/radiometric_controller_bkp.py index 8e70c59df..ffdfdf4d0 100644 --- a/Python/OAK/datasets/oak-fcc-3/core/radiometric_controller_bkp.py +++ b/Python/OAK/datasets/oak-fcc-3/core/radiometric_controller_bkp.py @@ -4,6 +4,20 @@ import numpy as np class RadiometricController: + """ + Controlador radiométrico para o módulo RGB/RE/NIR. + + Modos principais: + metering_mode: + - "global": usa uma região grande da cena, robusta por percentis. + - "reference_patches": usa patches/ROIs conhecidos, por exemplo branco/cinza/preto. + - "legacy_patch": compatível com o comportamento antigo: strip_y/patch_x. + + spectral_control_mode: + - "independent": controla rgb, re e nir separadamente. + - "shared": controla rgb separado e aplica uma decisão conjunta para re/nir. + """ + def __init__( self, client, @@ -14,62 +28,148 @@ class RadiometricController: strip_y1_pct=1.0, patch_x0_pct=0.35, patch_x1_pct=0.75, - target_mean=0.70, - deadband=0.03, - alpha=0.20, + target_mean=0.55, + deadband=0.04, + alpha=0.18, exp_min_us=100, exp_max_us=80000, gain_min=1.0, - gain_max=8.0, - exp_step_gain=0.65, + gain_max=4.0, + exp_step_gain=0.55, prefer_exposure=True, verbose=False, ): self.client = client - cfg = self._load_config_json(config_json_path) - interval_s = cfg.get("interval_s", interval_s) - strip_y0_pct = cfg.get("strip_y0_pct", strip_y0_pct) - strip_y1_pct = cfg.get("strip_y1_pct", strip_y1_pct) - patch_x0_pct = cfg.get("patch_x0_pct", patch_x0_pct) - patch_x1_pct = cfg.get("patch_x1_pct", patch_x1_pct) - target_mean = cfg.get("target_mean", target_mean) - deadband = cfg.get("deadband", deadband) - alpha = cfg.get("alpha", alpha) - exp_min_us = cfg.get("exp_min_us", exp_min_us) - exp_max_us = cfg.get("exp_max_us", exp_max_us) - gain_min = cfg.get("gain_min", gain_min) - gain_max = cfg.get("gain_max", gain_max) - exp_step_gain = cfg.get("exp_step_gain", exp_step_gain) - prefer_exposure = cfg.get("prefer_exposure", prefer_exposure) - verbose = cfg.get("verbose", verbose) - exp_apply_threshold_us = cfg.get("exp_apply_threshold_us", 50) - gain_apply_threshold = cfg.get("gain_apply_threshold", 0.02) + self.enabled = bool(cfg.get("enabled", enabled)) + self.interval_s = float(cfg.get("interval_s", interval_s)) + self.verbose = bool(cfg.get("verbose", verbose)) - self.enabled = bool(enabled) - self.interval_s = float(interval_s) + self.metering_mode = str(cfg.get("metering_mode", "global")).lower() + self.spectral_control_mode = str(cfg.get("spectral_control_mode", "shared")).lower() - self.strip_y0_pct = float(strip_y0_pct) - self.strip_y1_pct = float(strip_y1_pct) - self.patch_x0_pct = float(patch_x0_pct) - self.patch_x1_pct = float(patch_x1_pct) + if self.metering_mode not in ("global", "reference_patches", "legacy_patch"): + self.metering_mode = "global" - self.target_mean = float(target_mean) - self.deadband = float(deadband) - self.alpha = float(alpha) + if self.spectral_control_mode not in ("shared", "independent"): + self.spectral_control_mode = "shared" - self.exp_min_us = int(exp_min_us) - self.exp_max_us = int(exp_max_us) - self.gain_min = float(gain_min) - self.gain_max = float(gain_max) + self.strip_y0_pct = float(cfg.get("strip_y0_pct", strip_y0_pct)) + self.strip_y1_pct = float(cfg.get("strip_y1_pct", strip_y1_pct)) + self.patch_x0_pct = float(cfg.get("patch_x0_pct", patch_x0_pct)) + self.patch_x1_pct = float(cfg.get("patch_x1_pct", patch_x1_pct)) - self.exp_step_gain = float(exp_step_gain) - self.prefer_exposure = bool(prefer_exposure) - self.verbose = bool(verbose) + global_roi = cfg.get("global_roi_pct", {}) or {} + self.global_roi_pct = self._safe_roi_pct( + global_roi, + fallback={"x0": 0.08, "y0": 0.08, "x1": 0.92, "y1": 0.92}, + ) - self.exp_apply_threshold_us = int(exp_apply_threshold_us) - self.gain_apply_threshold = float(gain_apply_threshold) + raw_global_by_role = cfg.get("global_roi_pct_by_role", {}) or {} + self.global_roi_pct_by_role = {} + + for role in self.ROLES: + roi = raw_global_by_role.get(role) + if isinstance(roi, dict) and roi: + self.global_roi_pct_by_role[role] = self._safe_roi_pct( + roi, + fallback=self.global_roi_pct, + ) + else: + self.global_roi_pct_by_role[role] = dict(self.global_roi_pct) + + self.reference_patches = cfg.get("reference_patches", []) or [] + self.patch_aggregation = str(cfg.get("patch_aggregation", "weighted_mean")).lower() + + self.patch_control_mode = str(cfg.get("patch_control_mode", "gray_primary")).lower() + + self.patch_require_order = bool(cfg.get("patch_require_order", True)) + self.patch_min_separation = float(cfg.get("patch_min_separation", 0.08)) + + self.patch_white_sat_limit_pct = float(cfg.get("patch_white_sat_limit_pct", 0.50)) + self.patch_white_p95_limit = float(cfg.get("patch_white_p95_limit", 0.90)) + + self.patch_black_dark_limit_pct = float(cfg.get("patch_black_dark_limit_pct", 80.0)) + self.patch_black_max_p50 = float(cfg.get("patch_black_max_p50", 0.20)) + + self.patch_gray_min_p50 = float(cfg.get("patch_gray_min_p50", 0.08)) + self.patch_gray_max_p50 = float(cfg.get("patch_gray_max_p50", 0.85)) + + self.control_metric = str(cfg.get("control_metric", "p50")).lower() + self.target_value = float(cfg.get("target_value", cfg.get("target_mean", target_mean))) + self.target_mean = self.target_value + self.deadband = float(cfg.get("deadband", deadband)) + self.alpha = float(cfg.get("alpha", alpha)) + + self.p95_limit = float(cfg.get("p95_limit", 0.92)) + self.saturation_limit_pct = float(cfg.get("saturation_limit_pct", 0.50)) + self.dark_limit_pct = float(cfg.get("dark_limit_pct", 35.0)) + + self.reduce_fast_factor = float(cfg.get("reduce_fast_factor", 0.82)) + self.factor_min = float(cfg.get("factor_min", 0.72)) + self.factor_max = float(cfg.get("factor_max", 1.28)) + + self.saturation_hard_pct = float(cfg.get("saturation_hard_pct", 20.0)) + self.saturation_extreme_pct = float(cfg.get("saturation_extreme_pct", 60.0)) + + self.gain_return_enabled = bool(cfg.get("gain_return_enabled", True)) + self.gain_return_factor = float(cfg.get("gain_return_factor", 0.60)) + self.gain_reduce_on_saturation = bool(cfg.get("gain_reduce_on_saturation", True)) + + self.exp_high_ratio_for_gain = float(cfg.get("exp_high_ratio_for_gain", 0.85)) + self.exp_low_ratio_for_gain_return = float(cfg.get("exp_low_ratio_for_gain_return", 0.65)) + + self.gain_increase_required_cycles = int(cfg.get("gain_increase_required_cycles", 5)) + self.gain_decrease_required_cycles = int(cfg.get("gain_decrease_required_cycles", 2)) + + self.gain_step_up = float(cfg.get("gain_step_up", 0.25)) + self.gain_step_down = float(cfg.get("gain_step_down", 0.50)) + + self.gain_hard_reset_on_saturation = bool(cfg.get("gain_hard_reset_on_saturation", False)) + + self._underexposed_cycles = { + "rgb": 0, + "re": 0, + "nir": 0, + "spectral_shared": 0, + } + + self._overexposed_cycles = { + "rgb": 0, + "re": 0, + "nir": 0, + "spectral_shared": 0, + } + + self.control_strategy = str(cfg.get("control_strategy", "ratio")).lower() + + self.ratio_alpha = float(cfg.get("ratio_alpha", 0.55)) + self.ratio_min = float(cfg.get("ratio_min", 0.55)) + self.ratio_max = float(cfg.get("ratio_max", 1.85)) + + self.ready_required_cycles = int(cfg.get("ready_required_cycles", 3)) + self._ready_cycles = { + "rgb": 0, + "re": 0, + "nir": 0, + "spectral_shared": 0, + } + + self.exp_min_us = int(cfg.get("exp_min_us", exp_min_us)) + self.exp_max_us = int(cfg.get("exp_max_us", exp_max_us)) + self.gain_min = float(cfg.get("gain_min", gain_min)) + self.gain_max = float(cfg.get("gain_max", gain_max)) + self.role_limits = cfg.get("role_limits", {}) or {} + + self.exp_step_gain = float(cfg.get("exp_step_gain", exp_step_gain)) + self.prefer_exposure = bool(cfg.get("prefer_exposure", prefer_exposure)) + + self.exp_apply_threshold_us = int(cfg.get("exp_apply_threshold_us", 80)) + self.gain_apply_threshold = float(cfg.get("gain_apply_threshold", 0.05)) + + self.apply_same_spectral_to_both = bool(cfg.get("apply_same_spectral_to_both", True)) + self.spectral_roles = tuple(cfg.get("spectral_roles", ["re", "nir"])) self.last_update_ts = 0.0 self.last_result = {} @@ -79,6 +179,7 @@ class RadiometricController: "nir": {"exp": 15000, "gain": 1.0}, "re": {"exp": 15000, "gain": 1.0}, } + self._ae_disabled = set() self._last_applied = { "rgb": {"exp": None, "gain": None}, @@ -89,220 +190,964 @@ class RadiometricController: def _load_config_json(self, path): if not path: return {} - try: with open(path, "r", encoding="utf-8") as f: data = json.load(f) except Exception: return {} - cfg = data.get("radiometric_config", {}) return cfg if isinstance(cfg, dict) else {} + ROLES = ("rgb", "re", "nir") + + @classmethod + def _normalize_role(cls, role: str) -> str: + role = str(role or "").lower() + return role if role in cls.ROLES else "rgb" + + @staticmethod + def _safe_roi_pct(roi_pct, fallback=None) -> dict: + if fallback is None: + fallback = {"x0": 0.08, "y0": 0.08, "x1": 0.92, "y1": 0.92} + + if not isinstance(roi_pct, dict): + roi_pct = fallback + + return { + "x0": float(roi_pct.get("x0", fallback.get("x0", 0.08))), + "y0": float(roi_pct.get("y0", fallback.get("y0", 0.08))), + "x1": float(roi_pct.get("x1", fallback.get("x1", 0.92))), + "y1": float(roi_pct.get("y1", fallback.get("y1", 0.92))), + } + + def _get_global_roi_pct_for_role(self, role: str) -> dict: + role = self._normalize_role(role) + + by_role = getattr(self, "global_roi_pct_by_role", {}) or {} + if isinstance(by_role, dict): + roi = by_role.get(role) + if isinstance(roi, dict) and roi: + return self._safe_roi_pct(roi, fallback=self.global_roi_pct) + + return self._safe_roi_pct(self.global_roi_pct) + + def _get_patch_roi_pct_for_role(self, patch: dict, role: str): + role = self._normalize_role(role) + + by_role = patch.get("roi_pct_by_role", {}) + if isinstance(by_role, dict): + roi = by_role.get(role) + if isinstance(roi, dict) and roi: + return self._safe_roi_pct(roi), "roi_pct_by_role" + + legacy = patch.get("roi_pct") + if isinstance(legacy, dict) and legacy: + return self._safe_roi_pct(legacy), "roi_pct" + + return None, "missing" + + def get_patch_target_for_role(self, patch, role, fallback): + by_role = patch.get("target_value_by_role", {}) + if isinstance(by_role, dict) and role in by_role: + return float(by_role[role]) + return float(patch.get("target_value", fallback)) + def sync_from_camera_controls(self, camera_controls: dict | None): if not isinstance(camera_controls, dict): return - for role, ctrl in camera_controls.items(): - if role not in self.state: + role = str(role).lower() + if role not in self.state or not isinstance(ctrl, dict): continue - - exp = ctrl.get("exposure_time_us") - gain = ctrl.get("analogue_gain") - + src = ctrl + if "requested" in ctrl and isinstance(ctrl["requested"], dict): + src = ctrl["requested"] + exp = src.get("exposure_time_us") + gain = src.get("analogue_gain") if exp is not None: self.state[role]["exp"] = int(exp) - if gain is not None: self.state[role]["gain"] = float(gain) + def sync_from_actual_camera_controls(self): + for role in ("rgb", "re", "nir"): + try: + ctrl = self.client.svc.get_camera_controls(role=role) + exp = ctrl.get("exposure_time_us") + gain = ctrl.get("analogue_gain") + if exp is not None: + self.state[role]["exp"] = int(exp) + if gain is not None: + self.state[role]["gain"] = float(gain) + except Exception: + pass + def update(self, decoded: dict, meta: dict | None = None): if not self.enabled: return None - now = time.perf_counter() if now - self.last_update_ts < self.interval_s: return None - self.last_update_ts = now results = {} + rgb_result = self._update_single_role(decoded, "rgb") + if rgb_result is not None: + results["rgb"] = rgb_result - for role in ("rgb", "nir", "re"): - cam_id = self._resolve_cam_id(decoded, role) - if cam_id is None: - continue - - img = decoded[cam_id].get("image") - if img is None: - continue - - metrics = self.measure_reference_patch(img) - decision = self.compute_control(role, metrics) - apply_resp = self.apply_control(role, decision) - - results[role] = { - "cam_id": cam_id, - "metrics": metrics, - "decision": decision, - "apply": apply_resp, - } + if self.spectral_control_mode == "independent": + for role in self.spectral_roles: + result = self._update_single_role(decoded, role) + if result is not None: + results[role] = result + else: + result = self._update_spectral_shared(decoded) + if result is not None: + results["spectral_shared"] = result self.last_result = results return results + def _update_single_role(self, decoded: dict, role: str): + cam_id = self._resolve_cam_id(decoded, role) + if cam_id is None: + return None + img = decoded[cam_id].get("image") + if img is None: + return None + metrics = self.measure_image(img, role=role) + decision = self.compute_control(role, metrics) + apply_resp = self.apply_control(role, decision) + result = { + "mode": "single_role", + "cam_id": cam_id, + "role": role, + "metering_mode": self.metering_mode, + "metrics": metrics, + "decision": decision, + "apply": apply_resp, + } + self._print_metrics_debug(role, result) + + return result + + def _update_spectral_shared(self, decoded: dict): + role_items = {} + for role in self.spectral_roles: + cam_id = self._resolve_cam_id(decoded, role) + if cam_id is None: + continue + img = decoded[cam_id].get("image") + if img is None: + continue + role_items[role] = { + "cam_id": cam_id, + "metrics": self.measure_image(img, role=role), + } + if not role_items: + return None + + shared_metrics = self.aggregate_spectral_metrics(role_items) + state_role = "re" if "re" in role_items else list(role_items.keys())[0] + decision = self.compute_control(state_role, shared_metrics, virtual_role="spectral_shared") + + apply_resp = {} + if self.apply_same_spectral_to_both: + for role in role_items.keys(): + apply_resp[role] = self.apply_control(role, decision) + else: + apply_resp[state_role] = self.apply_control(state_role, decision) + + result = { + "mode": "shared_spectral", + "roles": role_items, + "metering_mode": self.metering_mode, + "metrics": shared_metrics, + "decision": decision, + "apply": apply_resp, + } + self._print_metrics_debug("spectral_shared", result) + + return result + + def _update_ready_state( + self, + log_role: str, + action: str, + error: float, + p95: float, + sat_pct: float, + ) -> tuple[bool, int]: + key = str(log_role).lower() + + is_ready_now = ( + action == "hold" + and abs(float(error)) <= self.deadband + and float(p95) <= self.p95_limit + and float(sat_pct) <= self.saturation_limit_pct + ) + + if is_ready_now: + self._ready_cycles[key] = self._ready_cycles.get(key, 0) + 1 + else: + self._ready_cycles[key] = 0 + + cycles = self._ready_cycles.get(key, 0) + return cycles >= self.ready_required_cycles, cycles + + def _update_exposure_pressure_state( + self, + log_role: str, + error: float, + p95: float, + sat_pct: float, + ) -> tuple[int, int]: + key = str(log_role).lower() + + under = ( + error > self.deadband + and p95 < self.p95_limit + and sat_pct <= self.saturation_limit_pct + ) + + over = ( + error < -self.deadband + or p95 > self.p95_limit + or sat_pct > self.saturation_limit_pct + ) + + if under: + self._underexposed_cycles[key] = self._underexposed_cycles.get(key, 0) + 1 + else: + self._underexposed_cycles[key] = 0 + + if over: + self._overexposed_cycles[key] = self._overexposed_cycles.get(key, 0) + 1 + else: + self._overexposed_cycles[key] = 0 + + return ( + self._underexposed_cycles.get(key, 0), + self._overexposed_cycles.get(key, 0), + ) + def _resolve_cam_id(self, decoded, role): role = str(role).lower() - for cam_id, data in decoded.items(): if str(data.get("role", "")).lower() == role: return cam_id - return None - def measure_reference_patch(self, img01: np.ndarray) -> dict: + def measure_image(self, img01: np.ndarray, role: str) -> dict: + role = self._normalize_role(role) + gray = self.to_luma_or_gray(img01) + + if self.metering_mode == "reference_patches": + return self.measure_reference_patches(gray, role=role) + + if self.metering_mode == "legacy_patch": + return self.measure_legacy_patch(gray) + + return self.measure_global(gray, role=role) + + @staticmethod + def to_luma_or_gray(img01: np.ndarray) -> np.ndarray: if img01.ndim == 3: - # RGB: usa luminância simples - img_gray = ( + return ( 0.299 * img01[:, :, 0] + 0.587 * img01[:, :, 1] + 0.114 * img01[:, :, 2] ).astype(np.float32) - else: - img_gray = img01.astype(np.float32) + return img01.astype(np.float32) + + def measure_global(self, img_gray: np.ndarray, role: str = "rgb") -> dict: + role = self._normalize_role(role) h, w = img_gray.shape[:2] + roi_pct = self._get_global_roi_pct_for_role(role) - y0 = int(h * self.strip_y0_pct) - y1 = int(h * self.strip_y1_pct) - x0 = int(w * self.patch_x0_pct) - x1 = int(w * self.patch_x1_pct) + roi = self._roi_pct_to_pixels( + h, w, + roi_pct["x0"], + roi_pct["y0"], + roi_pct["x1"], + roi_pct["y1"], + ) - y0 = max(0, min(h - 1, y0)) - y1 = max(y0 + 1, min(h, y1)) - x0 = max(0, min(w - 1, x0)) - x1 = max(x0 + 1, min(w, x1)) + arr = self._crop_array(img_gray, roi) + stats = self.compute_stats(arr) - patch = img_gray[y0:y1, x0:x1] - arr = patch.reshape(-1) + stats["roi"] = list(roi) + stats["roi_pct"] = dict(roi_pct) + stats["roi_source"] = "global_roi_pct_by_role" + stats["role"] = role + stats["source"] = "global" + + return stats + + def measure_legacy_patch(self, img_gray: np.ndarray) -> dict: + h, w = img_gray.shape[:2] + roi = self._roi_pct_to_pixels( + h, w, + self.patch_x0_pct, + self.strip_y0_pct, + self.patch_x1_pct, + self.strip_y1_pct, + ) + arr = self._crop_array(img_gray, roi) + stats = self.compute_stats(arr) + stats["roi"] = list(roi) + stats["source"] = "legacy_patch" + return stats + + def measure_reference_patches(self, img_gray: np.ndarray, role: str) -> dict: + role = self._normalize_role(role) + + h, w = img_gray.shape[:2] + patch_results = [] + + for patch in self.reference_patches: + if not isinstance(patch, dict): + continue + + roles = patch.get("roles", ["rgb", "re", "nir", "all"]) + roles = [str(r).lower() for r in roles] + + if role not in roles and "all" not in roles: + continue + + roi_pct, roi_source = self._get_patch_roi_pct_for_role(patch, role) + + if not isinstance(roi_pct, dict): + continue + + x0 = float(roi_pct.get("x0", 0.0)) + y0 = float(roi_pct.get("y0", 0.0)) + x1 = float(roi_pct.get("x1", 1.0)) + y1 = float(roi_pct.get("y1", 1.0)) + + roi = self._roi_pct_to_pixels(h, w, x0, y0, x1, y1) + arr = self._crop_array(img_gray, roi) + stats = self.compute_stats(arr) + + target = self.get_patch_target_for_role( + patch=patch, + role=role, + fallback=self.target_value, + ) + if target is not None: + target = float(target) + + patch_results.append({ + "name": patch.get("name", f"patch_{len(patch_results) + 1}"), + "type": patch.get("type", "reference"), + "role": role, + "roi": list(roi), + "roi_pct": {"x0": x0, "y0": y0, "x1": x1, "y1": y1}, + "roi_source": roi_source, + "weight": float(patch.get("weight", 1.0)), + "target_value": target, + "stats": stats, + }) + + if not patch_results: + stats = self.measure_global(img_gray, role=role) + stats["source"] = "reference_patches_fallback_global" + stats["patches"] = [] + return stats + + metrics = self.aggregate_patch_metrics(patch_results) + metrics["role"] = role + + return metrics + + def aggregate_patch_metrics(self, patch_results: list[dict]) -> dict: + valid = [p for p in patch_results if p["stats"].get("valid")] + + if not valid: + return { + "valid": False, + "source": "reference_patches", + "patches": patch_results, + "mean": 0.0, + "p50": 0.0, + "p95": 0.0, + "sat_pct": 0.0, + "dark_pct": 0.0, + "control_value": 0.0, + "target_value": self.target_value, + "weighted_error": 0.0, + "patch_quality": { + "valid": False, + "warnings": ["no_valid_patches"], + }, + } + + black = self._find_patch_result(valid, "black") + gray = self._find_patch_result(valid, "gray") + white = self._find_patch_result(valid, "white") + + warnings = [] + + # Stats gerais de proteção. + p95s = np.array([p["stats"]["p95"] for p in valid], dtype=np.float32) + sats = np.array([p["stats"]["sat_pct"] for p in valid], dtype=np.float32) + darks = np.array([p["stats"]["dark_pct"] for p in valid], dtype=np.float32) + means = np.array([p["stats"]["mean"] for p in valid], dtype=np.float32) + p50s = np.array([p["stats"]["p50"] for p in valid], dtype=np.float32) + + p95_max = float(np.max(p95s)) + sat_max = float(np.max(sats)) + dark_mean = float(np.mean(darks)) + mean_mean = float(np.mean(means)) + p50_mean = float(np.mean(p50s)) + + # Valores por patch, quando existem. + black_p50 = float(black["stats"]["p50"]) if black else None + gray_p50 = float(gray["stats"]["p50"]) if gray else None + white_p50 = float(white["stats"]["p50"]) if white else None + + black_target = float(black.get("target_value", 0.06)) if black else 0.06 + gray_target = float(gray.get("target_value", self.target_value)) if gray else self.target_value + white_target = float(white.get("target_value", 0.80)) if white else 0.80 + + # ============================================================ + # Validações de coerência dos cartões + # ============================================================ + + if gray is None: + warnings.append("missing_gray_patch") + + if self.patch_require_order and black and gray and white: + if not (black_p50 < gray_p50 < white_p50): + warnings.append( + f"patch_order_invalid: black={black_p50:.3f}, gray={gray_p50:.3f}, white={white_p50:.3f}" + ) + + if (gray_p50 - black_p50) < self.patch_min_separation: + warnings.append( + f"black_gray_separation_low: diff={gray_p50 - black_p50:.3f}" + ) + + if (white_p50 - gray_p50) < self.patch_min_separation: + warnings.append( + f"gray_white_separation_low: diff={white_p50 - gray_p50:.3f}" + ) + + if white: + white_sat = float(white["stats"]["sat_pct"]) + white_p95 = float(white["stats"]["p95"]) + if white_sat > self.patch_white_sat_limit_pct: + warnings.append(f"white_patch_saturated: sat={white_sat:.2f}%") + if white_p95 > self.patch_white_p95_limit: + warnings.append(f"white_patch_p95_high: p95={white_p95:.3f}") + + if black: + black_dark = float(black["stats"]["dark_pct"]) + if black_dark > self.patch_black_dark_limit_pct: + warnings.append(f"black_patch_too_dark: dark={black_dark:.1f}%") + if black_p50 > self.patch_black_max_p50: + warnings.append(f"black_patch_too_bright: p50={black_p50:.3f}") + + if gray: + if gray_p50 < self.patch_gray_min_p50: + warnings.append(f"gray_patch_too_dark: p50={gray_p50:.3f}") + if gray_p50 > self.patch_gray_max_p50: + warnings.append(f"gray_patch_too_bright: p50={gray_p50:.3f}") + + # ============================================================ + # Modo recomendado: cinza como controle principal + # ============================================================ + if self.patch_control_mode == "gray_primary" and gray is not None: + control_value = gray_p50 + target_value = gray_target + weighted_error = target_value - control_value + + control_source = "gray_primary" + + else: + # Fallback: média ponderada original, mas preservando guardas. + weights = np.array([max(0.0, p.get("weight", 1.0)) for p in valid], dtype=np.float32) + + if float(weights.sum()) <= 1e-9: + weights = np.ones(len(valid), dtype=np.float32) + + weights = weights / weights.sum() + + patch_errors = [] + control_values = [] + + for p in valid: + target = p.get("target_value") + if target is None: + target = self.target_value + + value = p["stats"].get( + self.control_metric, + p["stats"].get("p50", p["stats"].get("mean", 0.0)) + ) + + control_values.append(float(value)) + patch_errors.append(float(target) - float(value)) + + weighted_error = float(np.sum(np.array(patch_errors, dtype=np.float32) * weights)) + control_value = float(np.sum(np.array(control_values, dtype=np.float32) * weights)) + target_value = self.target_value + control_source = "weighted_patches" + + # ============================================================ + # Guardas de saturação e faixa útil + # ============================================================ + # Se o branco saturou, queremos que o compute_control reduza exposição, + # mesmo que o cinza esteja aparentemente bom. + if white: + white_sat = float(white["stats"]["sat_pct"]) + white_p95 = float(white["stats"]["p95"]) + + sat_max = max(sat_max, white_sat) + p95_max = max(p95_max, white_p95) + + # Se o cinza está ausente, a métrica ainda pode funcionar por fallback, + # mas marcamos warning para debug. + quality_valid = gray is not None and len(warnings) == 0 return { - "valid": arr.size > 0, - "mean": float(arr.mean()) if arr.size else 0.0, - "p05": float(np.percentile(arr, 5)) if arr.size else 0.0, - "p95": float(np.percentile(arr, 95)) if arr.size else 0.0, - "sat_pct": float((arr >= 0.98).mean() * 100.0) if arr.size else 0.0, - "dark_pct": float((arr <= 0.02).mean() * 100.0) if arr.size else 0.0, - "roi": [x0, y0, x1, y1], + "valid": True, + "source": "reference_patches", + "patches": patch_results, + + # Métricas agregadas informativas. + "mean": mean_mean, + "p50": p50_mean, + "p95": p95_max, + "sat_pct": sat_max, + "dark_pct": dark_mean, + + # Métricas usadas pelo controle. + "control_metric": self.control_metric, + "control_value": float(control_value), + "target_value": float(target_value), + "weighted_error": float(weighted_error), + + # Debug/qualidade. + "patch_control_mode": self.patch_control_mode, + "control_source": control_source, + "patch_quality": { + "valid": bool(quality_valid), + "warnings": warnings, + "black_p50": black_p50, + "gray_p50": gray_p50, + "white_p50": white_p50, + "black_target": black_target, + "gray_target": gray_target, + "white_target": white_target, + }, } - def compute_control(self, role: str, metrics: dict) -> dict: - st = self.state.setdefault(role, {"exp": 15000, "gain": 1.0}) + def aggregate_spectral_metrics(self, role_items: dict) -> dict: + valid_items = {role: item for role, item in role_items.items() if item["metrics"].get("valid")} + if not valid_items: + return { + "valid": False, + "source": "spectral_shared", + "roles": role_items, + "mean": 0.0, + "p50": 0.0, + "p95": 0.0, + "sat_pct": 0.0, + "dark_pct": 0.0, + "control_value": 0.0, + "target_value": self.target_value, + } + metrics_list = [item["metrics"] for item in valid_items.values()] + p95 = max(float(m.get("p95", 0.0)) for m in metrics_list) + sat_pct = max(float(m.get("sat_pct", 0.0)) for m in metrics_list) + mean = float(np.mean([float(m.get("mean", 0.0)) for m in metrics_list])) + p50 = float(np.mean([float(m.get("p50", m.get("mean", 0.0))) for m in metrics_list])) + dark_pct = float(np.mean([float(m.get("dark_pct", 0.0)) for m in metrics_list])) + control_values = [ + float(m.get("control_value", m.get(self.control_metric, m.get("p50", m.get("mean", 0.0))))) + for m in metrics_list + ] + + target_values = [ + float(m.get("target_value", self.target_value)) + for m in metrics_list + ] + + errors = [ + float(m.get("weighted_error", target - value)) + for m, target, value in zip(metrics_list, target_values, control_values) + ] + + control_value = float(np.mean(control_values)) + target_value = float(np.mean(target_values)) + weighted_error = float(np.mean(errors)) + + return { + "valid": True, + "source": "spectral_shared", + "roles": role_items, + "mean": mean, + "p50": p50, + "p95": p95, + "sat_pct": sat_pct, + "dark_pct": dark_pct, + "control_metric": self.control_metric, + "control_value": control_value, + "target_value": target_value, + "weighted_error": weighted_error, + "control_values_by_role": { + role: float(item["metrics"].get("control_value", item["metrics"].get("p50", 0.0))) + for role, item in valid_items.items() + }, + "targets_by_role": { + role: float(item["metrics"].get("target_value", self.target_value)) + for role, item in valid_items.items() + }, + "patch_quality_by_role": { + role: item["metrics"].get("patch_quality", {}) + for role, item in valid_items.items() + }, + } + + def compute_stats(self, arr: np.ndarray) -> dict: + arr = np.asarray(arr, dtype=np.float32).reshape(-1) + if arr.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(arr.size), + "mean": float(arr.mean()), + "std": float(arr.std()), + "p05": float(np.percentile(arr, 5)), + "p50": float(np.percentile(arr, 50)), + "p95": float(np.percentile(arr, 95)), + "sat_pct": float((arr >= 0.98).mean() * 100.0), + "dark_pct": float((arr <= 0.02).mean() * 100.0), + } + + @staticmethod + def _crop_array(img: np.ndarray, roi: tuple[int, int, int, int]) -> np.ndarray: + x0, y0, x1, y1 = roi + return img[y0:y1, x0:x1].reshape(-1) + + @staticmethod + def _roi_pct_to_pixels(h: int, w: int, x0_pct: float, y0_pct: float, x1_pct: float, y1_pct: float): + x0 = int(w * x0_pct) + x1 = int(w * x1_pct) + y0 = int(h * y0_pct) + y1 = int(h * y1_pct) + x0 = max(0, min(w - 1, x0)) + x1 = max(x0 + 1, min(w, x1)) + y0 = max(0, min(h - 1, y0)) + y1 = max(y0 + 1, min(h, y1)) + return x0, y0, x1, y1 + + @staticmethod + def _find_patch_result(patch_results: list[dict], patch_type: str): + patch_type = str(patch_type).lower() + for p in patch_results: + if str(p.get("type", "")).lower() == patch_type: + return p + return None + + def compute_control(self, role: str, metrics: dict, virtual_role: str | None = None) -> dict: + state_role = str(role).lower() + log_role = virtual_role or state_role + + st = self.state.setdefault(state_role, {"exp": 15000, "gain": 1.0}) old_exp = int(st["exp"]) old_gain = float(st["gain"]) + limits = self._limits_for_role(state_role) if not metrics.get("valid"): + ready, ready_cycles = self._update_ready_state( + log_role=log_role, + action="hold", + error=999.0, + p95=1.0, + sat_pct=100.0, + ) return { + "role": log_role, + "state_role": state_role, "action": "hold", - "reason": "patch inválido", + "reason": "métrica inválida", "old_exp": old_exp, "new_exp": old_exp, "old_gain": old_gain, "new_gain": old_gain, + "ready": ready, + "ready_cycles": ready_cycles, + "ready_required_cycles": self.ready_required_cycles, } - mean = float(metrics["mean"]) - p95 = float(metrics["p95"]) - sat_pct = float(metrics["sat_pct"]) - error = self.target_mean - mean + control_value = float(metrics.get( + "control_value", + metrics.get(self.control_metric, metrics.get("p50", metrics.get("mean", 0.0))) + )) + target = float(metrics.get("target_value", self.target_value)) + error = float(metrics.get("weighted_error", target - control_value)) + p95 = float(metrics.get("p95", 0.0)) + sat_pct = float(metrics.get("sat_pct", 0.0)) + + under_cycles, over_cycles = self._update_exposure_pressure_state( + log_role=log_role, + error=error, + p95=p95, + sat_pct=sat_pct, + ) + + exp_min = int(limits["exp_min_us"]) + exp_max = int(limits["exp_max_us"]) + gain_min = float(limits["gain_min"]) + gain_max = float(limits["gain_max"]) new_exp = old_exp new_gain = old_gain action = "hold" reason = "dentro da faixa morta" - # Proteção contra saturação - if sat_pct > 1.0 or p95 > 0.96: - desired_exp = max(self.exp_min_us, int(old_exp * 0.85)) - new_exp = self._smooth_int(old_exp, desired_exp) - action = "decrease_exposure" - reason = f"saturação detectada: sat={sat_pct:.2f}% p95={p95:.3f}" + ratio = None + factor = 1.0 + gain_policy = "hold" + # ============================================================ + # 1) Proteção forte contra saturação / p95 alto + # ============================================================ + if sat_pct > self.saturation_limit_pct or p95 > self.p95_limit: + if sat_pct >= self.saturation_extreme_pct: + exp_factor = 0.45 + elif sat_pct >= self.saturation_hard_pct: + exp_factor = 0.32 + elif sat_pct > self.saturation_limit_pct: + exp_factor = 0.55 + else: + exp_factor = self.reduce_fast_factor + + new_exp = int(self._clamp(old_exp * exp_factor, exp_min, exp_max)) + + if self.gain_reduce_on_saturation and old_gain > gain_min: + if self.gain_hard_reset_on_saturation and sat_pct >= self.saturation_extreme_pct: + new_gain = gain_min + gain_policy = "hard_reset_gain_on_extreme_saturation" + else: + desired_gain = old_gain - self.gain_step_down + new_gain = float(self._clamp(desired_gain, gain_min, gain_max)) + gain_policy = "decrease_gain_step_on_saturation" + else: + new_gain = old_gain + gain_policy = "hold_gain" + + action = "decrease_exposure" + reason = ( + f"saturação/p95 alto: sat={sat_pct:.2f}% p95={p95:.3f} " + f"exp_factor={exp_factor:.3f} gain_policy={gain_policy}" + ) + + # ============================================================ + # 2) Fora da faixa morta: controle por ratio/linear + # ============================================================ elif abs(error) > self.deadband: - factor = 1.0 + self.exp_step_gain * error - factor = max(0.70, min(1.35, factor)) + if self.control_strategy == "ratio": + safe_value = max(control_value, 1e-6) + ratio = target / safe_value + ratio = self._clamp(ratio, self.ratio_min, self.ratio_max) + factor = 1.0 + self.ratio_alpha * (ratio - 1.0) + else: + factor = 1.0 + self.exp_step_gain * error + factor = max(self.factor_min, min(self.factor_max, factor)) if self.prefer_exposure: - desired_exp = int(old_exp * factor) - desired_exp = self._clamp(desired_exp, self.exp_min_us, self.exp_max_us) - new_exp = self._smooth_int(old_exp, desired_exp) + # ---------------------------------------------------- + # 2A) Cena escura: subir exposição primeiro. + # Só subir ganho se exposição já estiver perto do máximo. + # ---------------------------------------------------- + if error > 0: + desired_exp = int(self._clamp(old_exp * factor, exp_min, exp_max)) + new_exp = desired_exp + new_gain = old_gain + gain_policy = "hold_gain_prefer_exposure" - # Se exposição bateu limite e ainda precisa clarear/escurecer, mexe no ganho - if desired_exp in (self.exp_min_us, self.exp_max_us): - desired_gain = old_gain * factor - desired_gain = self._clamp(desired_gain, self.gain_min, self.gain_max) - new_gain = self._smooth_float(old_gain, desired_gain) + exp_high_threshold = int(exp_max * self.exp_high_ratio_for_gain) + + if ( + desired_exp >= exp_high_threshold + and under_cycles >= self.gain_increase_required_cycles + ): + # Sobe ganho devagar, em degrau fixo. + desired_gain = old_gain + self.gain_step_up + new_gain = float(self._clamp(desired_gain, gain_min, gain_max)) + gain_policy = f"increase_gain_slow_under_cycles_{under_cycles}" + else: + new_gain = old_gain + gain_policy = f"hold_gain_under_cycles_{under_cycles}" + + action = "increase_exposure" + reason = ( + f"subindo exposição por {self.control_metric}: " + f"value={control_value:.3f} target={target:.3f} " + f"error={error:.3f} factor={factor:.3f} gain_policy={gain_policy}" + ) + + # ---------------------------------------------------- + # 2B) Cena clara: se ganho está acima do mínimo, + # reduzir ganho primeiro ou junto. + # ---------------------------------------------------- + else: + desired_exp = int(self._clamp(old_exp * factor, exp_min, exp_max)) + new_exp = desired_exp + + if ( + self.gain_return_enabled + and old_gain > gain_min + and over_cycles >= self.gain_decrease_required_cycles + ): + desired_gain = old_gain - self.gain_step_down + new_gain = float(self._clamp(desired_gain, gain_min, gain_max)) + gain_policy = f"return_gain_step_over_cycles_{over_cycles}" + else: + new_gain = old_gain + gain_policy = f"hold_gain_over_cycles_{over_cycles}" + + action = "decrease_exposure" + reason = ( + f"reduzindo brilho por {self.control_metric}: " + f"value={control_value:.3f} target={target:.3f} " + f"error={error:.3f} factor={factor:.3f} gain_policy={gain_policy}" + ) - action = "increase_exposure" if error > 0 else "decrease_exposure" - reason = f"corrigindo erro radiométrico: error={error:.3f}" else: desired_gain = old_gain * factor - desired_gain = self._clamp(desired_gain, self.gain_min, self.gain_max) - new_gain = self._smooth_float(old_gain, desired_gain) + new_gain = float(self._clamp(desired_gain, gain_min, gain_max)) action = "increase_gain" if error > 0 else "decrease_gain" - reason = f"corrigindo ganho: error={error:.3f}" + gain_policy = "direct_gain_control" + reason = ( + f"corrigindo ganho por {self.control_metric}: " + f"value={control_value:.3f} target={target:.3f} " + f"error={error:.3f} factor={factor:.3f}" + ) - new_exp = int(self._clamp(new_exp, self.exp_min_us, self.exp_max_us)) - new_gain = float(self._clamp(new_gain, self.gain_min, self.gain_max)) + # ============================================================ + # 3) Dentro da faixa morta: opcionalmente devolver ganho + # se ganho alto não é mais necessário. + # ============================================================ + else: + if self.gain_return_enabled and old_gain > gain_min: + exp_low_threshold = int(exp_max * self.exp_low_ratio_for_gain_return) + + if old_exp < exp_low_threshold: + new_gain = float(self._clamp(old_gain * self.gain_return_factor, gain_min, gain_max)) + gain_policy = "return_gain_while_ready" + action = "decrease_gain" + reason = ( + f"dentro da faixa, devolvendo ganho: " + f"value={control_value:.3f} target={target:.3f} " + f"gain={old_gain:.2f}->{new_gain:.2f}" + ) + else: + gain_policy = "hold_gain_high_exp" + else: + gain_policy = "hold_gain" + + new_exp = int(self._clamp(new_exp, exp_min, exp_max)) + new_gain = float(self._clamp(new_gain, gain_min, gain_max)) + + ready, ready_cycles = self._update_ready_state( + log_role=log_role, + action=action, + error=error, + p95=p95, + sat_pct=sat_pct, + ) return { + "role": log_role, + "state_role": state_role, "action": action, "reason": reason, - "mean": mean, - "target_mean": self.target_mean, + "metering_mode": self.metering_mode, + "spectral_control_mode": self.spectral_control_mode, + "control_metric": self.control_metric, + "control_value": control_value, + "target_value": target, "error": error, + "p95": p95, + "sat_pct": sat_pct, + #"metrics_source": metrics.get("source"), + #"control_source": metrics.get("control_source"), + #"patch_quality": metrics.get("patch_quality"), + #"patches": metrics.get("patches"), + #"control_values_by_role": metrics.get("control_values_by_role"), + #"targets_by_role": metrics.get("targets_by_role"), + #"patch_quality_by_role": metrics.get("patch_quality_by_role"), "old_exp": old_exp, "new_exp": new_exp, "old_gain": old_gain, "new_gain": new_gain, + "limits": limits, + "ready": ready, + "ready_cycles": ready_cycles, + "ready_required_cycles": self.ready_required_cycles, + "control_strategy": self.control_strategy, + "ratio": ratio, + "factor": float(factor), + "gain_policy": gain_policy, } def apply_control(self, role: str, decision: dict): + role = str(role).lower() new_exp = int(decision["new_exp"]) new_gain = float(decision["new_gain"]) - self.state[role]["exp"] = new_exp self.state[role]["gain"] = new_gain - responses = {} last = self._last_applied.setdefault(role, {"exp": None, "gain": None}) - try: if role not in self._ae_disabled: responses["ae"] = self.client.svc.set_ae_enable(role=role, enable=False) - if role == "rgb": responses["awb"] = self.client.svc.set_awb_enable(role=role, enable=False) - self._ae_disabled.add(role) - if last["exp"] is None or abs(new_exp - last["exp"]) >= self.exp_apply_threshold_us: responses["exposure"] = self.client.svc.set_exposure_time(role=role, exposure_time_us=new_exp) last["exp"] = new_exp - if last["gain"] is None or abs(new_gain - last["gain"]) >= self.gain_apply_threshold: responses["gain"] = self.client.svc.set_analogue_gain(role=role, analogue_gain=new_gain) last["gain"] = new_gain - except Exception as e: responses["error"] = str(e) - if self.verbose: - print(f"[RAD] {role}: {json.dumps(decision, ensure_ascii=False)} | apply={responses}") - + print( + f"[RAD_APPLY] role={role} " + f"action={decision.get('action')} " + f"exp={decision.get('old_exp')}->{decision.get('new_exp')} " + f"gain={decision.get('old_gain'):.2f}->{decision.get('new_gain'):.2f} " + f"ok={'error' not in responses}" + ) return responses + def _limits_for_role(self, role: str) -> dict: + role_cfg = self.role_limits.get(role, {}) or {} + return { + "exp_min_us": int(role_cfg.get("exp_min_us", self.exp_min_us)), + "exp_max_us": int(role_cfg.get("exp_max_us", self.exp_max_us)), + "gain_min": float(role_cfg.get("gain_min", self.gain_min)), + "gain_max": float(role_cfg.get("gain_max", self.gain_max)), + } + def _smooth_int(self, old, desired): return int(round((1.0 - self.alpha) * old + self.alpha * desired)) @@ -311,4 +1156,68 @@ class RadiometricController: @staticmethod def _clamp(v, lo, hi): - return max(lo, min(hi, v)) \ No newline at end of file + return max(lo, min(hi, v)) + + def _print_metrics_debug(self, role: str, result: dict): + if not self.verbose: + return + + metrics = result.get("metrics", {}) + decision = result.get("decision", {}) + + print( + f"[RAD_METRICS] role={role} " + f"mode={result.get('mode')} " + f"metering={result.get('metering_mode')} " + f"action={decision.get('action')} " + f"exp={decision.get('old_exp')}->{decision.get('new_exp')} " + f"gain={decision.get('old_gain')}->{decision.get('new_gain')} " + f"control={decision.get('control_value'):.3f} " + f"target={decision.get('target_value'):.3f} " + f"p95={decision.get('p95'):.3f} " + f"sat={decision.get('sat_pct'):.2f}%" + ) + + # Caso normal: rgb individual + patches = metrics.get("patches", []) + if patches: + for p in patches: + st = p.get("stats", {}) + print( + f" [PATCH] role={p.get('role', role)} " + f"type={p.get('type')} " + f"roi_source={p.get('roi_source')} " + f"roi_pct={p.get('roi_pct')} " + f"p50={st.get('p50', 0):.3f} " + f"p95={st.get('p95', 0):.3f} " + f"sat={st.get('sat_pct', 0):.2f}% " + f"dark={st.get('dark_pct', 0):.1f}%" + ) + + # Caso spectral_shared: RE/NIR agregados + roles = metrics.get("roles", {}) + if roles: + for r, item in roles.items(): + m = item.get("metrics", {}) + print( + f" [ROLE_METRICS] role={r} " + f"cam_id={item.get('cam_id')} " + f"control={m.get('control_value', 0):.3f} " + f"target={m.get('target_value', 0):.3f} " + f"p95={m.get('p95', 0):.3f} " + f"sat={m.get('sat_pct', 0):.2f}% " + f"warnings={m.get('patch_quality', {}).get('warnings', [])}" + ) + + for p in m.get("patches", []): + st = p.get("stats", {}) + print( + f" [PATCH] role={p.get('role', r)} " + f"type={p.get('type')} " + f"roi_source={p.get('roi_source')} " + f"roi_pct={p.get('roi_pct')} " + f"p50={st.get('p50', 0):.3f} " + f"p95={st.get('p95', 0):.3f} " + f"sat={st.get('sat_pct', 0):.2f}% " + f"dark={st.get('dark_pct', 0):.1f}%" + ) diff --git a/Python/OAK/datasets/oak-fcc-3/utils/radiometric_config_tool.py b/Python/OAK/datasets/oak-fcc-3/utils/radiometric_config_tool.py index 445eaba38..0f5d6ea78 100644 --- a/Python/OAK/datasets/oak-fcc-3/utils/radiometric_config_tool.py +++ b/Python/OAK/datasets/oak-fcc-3/utils/radiometric_config_tool.py @@ -302,6 +302,24 @@ def default_profile_global(): } +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() cfg.update({ @@ -309,7 +327,6 @@ def default_profile_patches(): "spectral_control_mode": "shared", "deadband": 0.035, - "metering_mode": "reference_patches", "patch_control_mode": "gray_primary", "patch_require_order": True, "patch_min_separation": 0.08, @@ -323,34 +340,16 @@ def default_profile_patches(): "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": [ - { - "name": "black_reference", - "type": "black", - "roles": ["rgb", "re", "nir"], - "target_value": 0.06, - "weight": 0.25, - "roi_pct": {}, - "roi_pct_by_role": {"rgb": {}, "re": {}, "nir": {}} - }, - { - "name": "gray_reference", - "type": "gray", - "roles": ["rgb", "re", "nir"], - "target_value": 0.40, - "weight": 1.0, - "roi_pct": {}, - "roi_pct_by_role": {"rgb": {}, "re": {}, "nir": {}} - }, - { - "name": "white_reference", - "type": "white", - "roles": ["rgb", "re", "nir"], - "target_value": 0.78, - "weight": 0.7, - "roi_pct": {}, - "roi_pct_by_role": {"rgb": {}, "re": {}, "nir": {}} - } + make_default_patch("black", 0.06, 0.25), + make_default_patch("gray", 0.40, 1.0), + make_default_patch("white", 0.78, 0.7), ], }) @@ -358,7 +357,6 @@ def default_profile_patches(): "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"): @@ -514,36 +512,122 @@ def get_patches(data: dict): ) -def ensure_patch_roi_by_role(patch: dict): - legacy = patch.get("roi_pct", {}) +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: - if role not in by_role or not by_role[role]: - by_role[role] = clone_roi(legacy) if legacy else {} + 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 {} - return by_role + 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 get_patch_roi_for_role(data: dict, patch_type: str, role: str): - role = normalize_role(role) - patch = get_patch_by_type(data, patch_type) - if not patch: - return {} - - by_role = ensure_patch_roi_by_role(patch) - return by_role.get(role, {}) or patch.get("roi_pct", {}) or {} - - -def set_patch_roi_for_role(data: dict, patch_type: str, role: str, roi_pct: dict): +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", @@ -556,37 +640,114 @@ def set_patch_roi_for_role(data: dict, patch_type: str, role: str, roi_pct: dict ) patch_type = str(patch_type).lower() - role = normalize_role(role) - 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) - patch = None for p in patches: if str(p.get("type", "")).lower() == patch_type: - patch = p - break + ensure_patch_roi_lists_by_role(p) + return p - if patch is None: - patch = { - "name": f"{patch_type}_reference", - "type": patch_type, - "roles": ROLES[:], - "target_value": target, - "weight": weight, - "roi_pct": {}, - "roi_pct_by_role": {}, - } - patches.append(patch) + patch = make_default_patch(patch_type, target, weight) + patches.append(patch) + ensure_patch_roi_lists_by_role(patch) + return patch - by_role = ensure_patch_roi_by_role(patch) - by_role[role] = roi_pct - # Compatibilidade com formato antigo. - # Mantém roi_pct como RGB, para scripts antigos não quebrarem. - patch["roi_pct"] = by_role.get("rgb", roi_pct) +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()) @@ -624,7 +785,7 @@ def draw_roi_on_panel(panel, roi_pct, label, color, thickness=2): 0.55, color, 1, cv2.LINE_AA) -def draw_all_rois(panel, data, selected_target, mode, panel_role, edit_role): +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) @@ -640,13 +801,22 @@ def draw_all_rois(panel, data, selected_target, mode, panel_role, edit_role): for p in get_patches(data): typ = str(p.get("type", "")).lower() color = PATCH_COLORS.get(typ, (0, 255, 255)) - roi = get_patch_roi_for_role(data, typ, panel_role) + entries = get_patch_roi_entries_for_role(data, typ, panel_role, enabled_only=False) - label = f"{typ.upper()}/{panel_role.upper()}" - selected = typ == selected_target and is_edit_panel - thickness = 3 if selected else 2 + for idx, entry in enumerate(entries): + roi = entry.get("roi_pct", {}) + if not roi: + continue - draw_roi_on_panel(panel, roi, label, color, thickness) + 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( @@ -655,6 +825,7 @@ def build_board( mode, selected_target, edit_role, + selected_roi_index, drag_rect_local, drag_role, panel_rects, @@ -717,9 +888,9 @@ def build_board( 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) - draw_all_rois(re_panel, data, selected_target, mode, "re", edit_role) - draw_all_rois(nir_panel, data, selected_target, mode, "nir", edit_role) + 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 @@ -760,8 +931,8 @@ def build_board( board = np.vstack([top, bottom]) x0, y0, x1, y1 = panel_rects["data"] - lines = build_data_lines(decoded, data, mode, selected_target, edit_role, base_w, base_h) - overlay_hud(board, lines, x=x0 + 16, y=y0 + 28, font_scale=0.53, line_step=21) + 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( @@ -773,19 +944,19 @@ def build_board( return board -def build_data_lines(decoded, data, mode, selected_target, edit_role, base_w, base_h): +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_edicao={mode.upper()} | camera_editada={edit_role.upper()} | active={active_profile}", + 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"interval={active_cfg.get('interval_s')}s | ratio_alpha={active_cfg.get('ratio_alpha')} | ready={active_cfg.get('ready_required_cycles')}", + f"roi_contract={active_cfg.get('patch_roi_contract', 'legacy_single_roi')}", "", - "Arraste no painel da camera editada para definir a ROI.", - "Cada camera salva sua propria ROI: RGB / RE / NIR.", + "Arraste no painel da camera editada para definir/atualizar ROI.", + "PATCHES agora suportam N ROIs por cor e por camera.", "", ] @@ -801,39 +972,73 @@ def build_data_lines(decoded, data, mode, selected_target, edit_role, base_w, ba 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("ROIs do patch selecionado:") + 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: - roi_pct = get_patch_roi_for_role(data, selected_target, role) + 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()}: roi={roi_pct}") + 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("") 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 do patch {selected_target.upper()}:") + 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 = alterna GLOBAL / 3 PATCHES", - "C = alterna camera RGB / RE / NIR", - "V = alterna preview bonito / bruto", - "1/2/3 = BLACK / GRAY / WHITE", - "S = alterna spectral shared/independent", - "P ou SPACE = salva JSON", - "R = restaura defaults | Q/Esc = sai", + "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 = [] @@ -843,27 +1048,55 @@ def stats_lines_for_mode(data, decoded, mode: str, patch_type: str | None = 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) - else: - roi_pct = get_patch_roi_for_role(data, patch_type, role) - - if not roi_pct: - lines.append(f"{role.upper()}: sem ROI") + 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 - h, w = img.shape[:2] - roi = pct_to_px(roi_pct, w, h) - st = compute_stats(img, roi) + 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()}: p50={st['p50']:.3f} p95={st['p95']:.3f} " - f"sat={st['sat_pct']:.2f}% dark={st['dark_pct']:.1f}%" + 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 @@ -904,6 +1137,7 @@ def main(): mode = "global" selected_target = "gray" + selected_roi_index = 0 edit_role = "rgb" drag_role = None beauty_preview = True @@ -923,7 +1157,7 @@ def main(): 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 + 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: @@ -970,8 +1204,13 @@ def main(): set_global_roi_for_role(data, role_to_save, roi_pct) last_msg = f"GLOBAL ROI {role_to_save.upper()} atualizada: {roi_pct}" else: - set_patch_roi_for_role(data, selected_target, role_to_save, roi_pct) - last_msg = f"{selected_target.upper()} ROI {role_to_save.upper()} atualizada: {roi_pct}" + 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() @@ -1018,6 +1257,7 @@ def main(): 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, @@ -1052,12 +1292,14 @@ def main(): 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" @@ -1066,6 +1308,7 @@ def main(): 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" @@ -1074,6 +1317,7 @@ def main(): 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" @@ -1131,9 +1375,44 @@ def main(): 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}"