#!/usr/bin/env python3 """Audita a normalizacao radiometrica do OAK-FFC-3 usando os meta.json. O calculo replica o contrato do RawProcessorCore: factor_actual = exposure_time_us * (sensitivity_iso / iso_base) factor_ref = ref_exposure_time_us * (ref_iso / iso_base) raw_scale = factor_ref / factor_actual applied_scale = clip(raw_scale, scale_min, scale_max) O script nao abre os .bin e, portanto, mede o risco de clamp do fator, nao a saturacao real dos pixels. Ele gera um CSV por frame e um resumo JSON. """ from __future__ import annotations import argparse import csv import json import math import sys from collections import defaultdict from pathlib import Path from typing import Any, Iterable ROLES = ("rgb", "re", "nir") ROLE_FALLBACK = {"CAM_A": "rgb", "CAM_B": "re", "CAM_C": "nir"} PERCENTILES = (1, 5, 10, 25, 50, 75, 90, 95, 99) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Audita exposure/ISO e fatores da radiometric_normalization." ) parser.add_argument( "--meta-root", type=Path, default=Path("dataset/brutas/group"), help="Raiz contendo os grupos e suas pastas metas (padrao: dataset/brutas/group).", ) parser.add_argument( "--module-params", type=Path, default=Path("calibration/module_params.json"), help="JSON com radiometric_normalization.", ) parser.add_argument( "--pattern", default="*.json", help="Padrao dos metadados dentro das pastas metas (padrao: *.json).", ) parser.add_argument( "--out-dir", type=Path, default=Path("radiometric_audit"), help="Pasta dos relatorios.", ) parser.add_argument( "--recursive-anywhere", action="store_true", help="Procura JSON em qualquer subpasta; por padrao prioriza /metas/.", ) return parser.parse_args() def load_json(path: Path) -> dict[str, Any]: with path.open("r", encoding="utf-8-sig") as handle: data = json.load(handle) if not isinstance(data, dict): raise ValueError("raiz JSON nao e objeto") return data def finite_float(value: Any, default: float | None = None) -> float | None: try: result = float(value) except (TypeError, ValueError): return default return result if math.isfinite(result) else default def percentile(values: list[float], q: float) -> float | None: if not values: return None ordered = sorted(values) if len(ordered) == 1: return float(ordered[0]) position = (len(ordered) - 1) * float(q) / 100.0 lo = int(math.floor(position)) hi = int(math.ceil(position)) if lo == hi: return float(ordered[lo]) fraction = position - lo return float(ordered[lo] * (1.0 - fraction) + ordered[hi] * fraction) def describe(values: Iterable[float]) -> dict[str, Any]: vals = [float(v) for v in values if math.isfinite(float(v))] if not vals: return {"count": 0} mean = sum(vals) / len(vals) variance = sum((v - mean) ** 2 for v in vals) / len(vals) result: dict[str, Any] = { "count": len(vals), "min": min(vals), "max": max(vals), "mean": mean, "std": math.sqrt(max(variance, 0.0)), } result.update({f"p{q:02d}": percentile(vals, q) for q in PERCENTILES}) return result def get_nested_stream_meta(meta: dict[str, Any]) -> dict[str, Any]: candidates = [ meta.get("stream_meta"), meta.get("meta"), meta, ] for candidate in candidates: if not isinstance(candidate, dict): continue nested = candidate.get("stream_meta") if isinstance(nested, dict) and isinstance(nested.get("frame_controls"), dict): return nested if isinstance(candidate.get("frame_controls"), dict): return candidate return {} def role_map_from_stream(stream_meta: dict[str, Any]) -> dict[str, str]: mapping: dict[str, str] = {} camera_info = stream_meta.get("camera_info", {}) if isinstance(camera_info, dict): for cam_id, info in camera_info.items(): if not isinstance(info, dict): continue role = str(info.get("role", "")).strip().lower() if role in ROLES: mapping[str(cam_id).upper()] = role for cam_id, role in ROLE_FALLBACK.items(): mapping.setdefault(cam_id, role) return mapping def infer_group(meta_path: Path, root: Path) -> str: try: relative = meta_path.relative_to(root) except ValueError: return "unknown" parts = relative.parts if "metas" in parts: index = parts.index("metas") if index > 0: return parts[index - 1] return parts[0] if len(parts) > 1 else "root" def discover_meta_files(root: Path, pattern: str, anywhere: bool) -> list[Path]: if root.is_file(): return [root] if not root.is_dir(): raise FileNotFoundError(f"Raiz de metadados nao encontrada: {root}") if anywhere: return sorted(p for p in root.rglob(pattern) if p.is_file()) files = sorted(p for p in root.glob(f"*/metas/{pattern}") if p.is_file()) if files: return files return sorted(p for p in root.rglob(pattern) if p.is_file()) def read_contract(module_params: dict[str, Any]) -> dict[str, Any]: cfg = module_params.get("radiometric_normalization", {}) if not isinstance(cfg, dict): raise ValueError("radiometric_normalization ausente ou invalido") iso_base = finite_float(cfg.get("iso_base"), 100.0) or 100.0 references = cfg.get("reference_controls", {}) limits = cfg.get("scale_limits", {}) if not isinstance(references, dict): references = {} if not isinstance(limits, dict): limits = {} roles: dict[str, Any] = {} for role in ROLES: ref = references.get(role, {}) if not isinstance(ref, dict): ref = {} exp = finite_float(ref.get("exposure_time_us"), 0.0) or 0.0 iso = finite_float(ref.get("sensitivity_iso"), iso_base) or iso_base reference_factor = exp * (iso / iso_base) role_limits = limits.get(role) if not isinstance(role_limits, dict): role_limits = limits.get("default", {}) if not isinstance(role_limits, dict): role_limits = {} scale_min = finite_float(role_limits.get("min"), 0.15) or 0.15 scale_max = finite_float(role_limits.get("max"), 6.0) or 6.0 if scale_min <= 0: scale_min = 0.001 if scale_max < scale_min: scale_max = scale_min roles[role] = { "reference_exposure_time_us": exp, "reference_sensitivity_iso": iso, "reference_factor": reference_factor, "scale_min": scale_min, "scale_max": scale_max, } return { "enabled": bool(cfg.get("enabled", False)), "method": cfg.get("method"), "iso_base": iso_base, "roles": roles, } def extract_role_controls(meta: dict[str, Any]) -> tuple[dict[str, dict[str, Any]], dict[str, Any]]: stream = get_nested_stream_meta(meta) controls = stream.get("frame_controls", {}) if not isinstance(controls, dict): controls = {} mapping = role_map_from_stream(stream) by_role: dict[str, dict[str, Any]] = {} for cam_id, control in controls.items(): if not isinstance(control, dict): continue role = mapping.get(str(cam_id).upper(), "") if role in ROLES: by_role[role] = control info = { "frame_id": stream.get("frame_id", meta.get("frame_id")), "sync_ok": stream.get("sync_ok", meta.get("sync_ok")), "sync_dt_ms": stream.get("sync_dt_ms", meta.get("sync_dt_ms")), "timestamp": meta.get("ts", stream.get("ts")), } return by_role, info def compute_role_row(control: dict[str, Any], contract: dict[str, Any]) -> dict[str, Any] | None: exp = finite_float(control.get("exposure_time_us")) iso = finite_float(control.get("sensitivity_iso")) gain_est = finite_float(control.get("analogue_gain_est")) gain = finite_float(control.get("analogue_gain")) if exp is None or exp <= 0: return None iso_base = float(contract["iso_base"]) if iso is not None: gain_factor = iso / iso_base gain_source = "sensitivity_iso" elif gain_est is not None: gain_factor = gain_est gain_source = "analogue_gain_est" elif gain is not None: gain_factor = gain gain_source = "analogue_gain" else: gain_factor = 1.0 gain_source = "unity_fallback" if not math.isfinite(gain_factor) or gain_factor <= 0: gain_factor = 1.0 gain_source = "unity_invalid_fallback" actual_factor = exp * gain_factor reference_factor = float(contract["reference_factor"]) raw_scale = reference_factor / actual_factor if actual_factor > 0 else math.nan scale_min = float(contract["scale_min"]) scale_max = float(contract["scale_max"]) applied_scale = min(max(raw_scale, scale_min), scale_max) tolerance = max(1e-9, 1e-9 * max(abs(raw_scale), abs(scale_max), 1.0)) return { "exposure_time_us": exp, "sensitivity_iso": iso, "analogue_gain_est": gain_est, "gain_factor_used": gain_factor, "gain_source": gain_source, "actual_factor": actual_factor, "reference_factor": reference_factor, "raw_scale": raw_scale, "applied_scale": applied_scale, "scale_min": scale_min, "scale_max": scale_max, "clipped_low": raw_scale < scale_min - tolerance, "clipped_high": raw_scale > scale_max + tolerance, "at_low_limit": abs(applied_scale - scale_min) <= tolerance, "at_high_limit": abs(applied_scale - scale_max) <= tolerance, } def empty_accumulator() -> dict[str, list[float] | int]: return { "exposure_time_us": [], "sensitivity_iso": [], "gain_factor_used": [], "actual_factor": [], "raw_scale": [], "applied_scale": [], "clipped_low": 0, "clipped_high": 0, } def add_to_accumulator(acc: dict[str, Any], row: dict[str, Any]) -> None: for key in ( "exposure_time_us", "sensitivity_iso", "gain_factor_used", "actual_factor", "raw_scale", "applied_scale", ): value = row.get(key) if value is not None and math.isfinite(float(value)): acc[key].append(float(value)) acc["clipped_low"] += int(bool(row.get("clipped_low"))) acc["clipped_high"] += int(bool(row.get("clipped_high"))) def summarize_accumulator(acc: dict[str, Any], contract: dict[str, Any]) -> dict[str, Any]: actual = list(acc["actual_factor"]) count = len(actual) clipped_low = int(acc["clipped_low"]) clipped_high = int(acc["clipped_high"]) scale_min = float(contract["scale_min"]) scale_max = float(contract["scale_max"]) p01 = percentile(actual, 1) p50 = percentile(actual, 50) p99 = percentile(actual, 99) feasible_low = scale_min * p99 if p99 is not None else None feasible_high = scale_max * p01 if p01 is not None else None return { "count": count, "contract": contract, "exposure_time_us": describe(acc["exposure_time_us"]), "sensitivity_iso": describe(acc["sensitivity_iso"]), "gain_factor_used": describe(acc["gain_factor_used"]), "actual_factor": describe(actual), "raw_scale": describe(acc["raw_scale"]), "applied_scale": describe(acc["applied_scale"]), "clipping": { "low_count": clipped_low, "low_pct": 100.0 * clipped_low / count if count else 0.0, "high_count": clipped_high, "high_pct": 100.0 * clipped_high / count if count else 0.0, "any_count": clipped_low + clipped_high, "any_pct": 100.0 * (clipped_low + clipped_high) / count if count else 0.0, }, "recommendation": { "median_actual_factor": p50, "suggested_reference_exposure_us_at_iso100": p50, "central_98pct_feasible_reference_factor_min": feasible_low, "central_98pct_feasible_reference_factor_max": feasible_high, "central_98pct_has_feasible_interval": ( feasible_low is not None and feasible_high is not None and feasible_low <= feasible_high ), "note": ( "A mediana coloca a escala mediana em 1.0. Nao aplique automaticamente: " "valide tambem a saturacao dos pixels e preserve a calibracao cruzada entre bandas." ), }, } def json_safe(value: Any) -> Any: if isinstance(value, dict): return {str(k): json_safe(v) for k, v in value.items()} if isinstance(value, list): return [json_safe(v) for v in value] if isinstance(value, float) and not math.isfinite(value): return None return value def main() -> int: args = parse_args() module_params = load_json(args.module_params) contract = read_contract(module_params) meta_files = discover_meta_files(args.meta_root, args.pattern, args.recursive_anywhere) if not meta_files: print(f"[ERRO] Nenhum meta encontrado em {args.meta_root}", file=sys.stderr) return 2 args.out_dir.mkdir(parents=True, exist_ok=True) rows: list[dict[str, Any]] = [] skipped: list[dict[str, Any]] = [] global_acc = {role: empty_accumulator() for role in ROLES} group_acc: dict[str, dict[str, dict[str, Any]]] = defaultdict( lambda: {role: empty_accumulator() for role in ROLES} ) samples_complete = 0 samples_with_valid_controls = 0 samples_any_clipped = 0 for meta_path in meta_files: group = infer_group(meta_path, args.meta_root) try: meta = load_json(meta_path) controls_by_role, info = extract_role_controls(meta) except Exception as exc: skipped.append({"path": str(meta_path), "reason": f"json_error:{exc}"}) continue if not controls_by_role: skipped.append({"path": str(meta_path), "reason": "missing_frame_controls"}) continue samples_with_valid_controls += 1 complete = all(role in controls_by_role for role in ROLES) samples_complete += int(complete) frame_any_clipped = False for role in ROLES: control = controls_by_role.get(role) if control is None: skipped.append({"path": str(meta_path), "reason": f"missing_role:{role}"}) continue role_row = compute_role_row(control, {"iso_base": contract["iso_base"], **contract["roles"][role]}) if role_row is None: skipped.append({"path": str(meta_path), "reason": f"invalid_controls:{role}"}) continue frame_any_clipped = frame_any_clipped or bool( role_row["clipped_low"] or role_row["clipped_high"] ) add_to_accumulator(global_acc[role], role_row) add_to_accumulator(group_acc[group][role], role_row) rows.append( { "group": group, "meta_path": str(meta_path), "timestamp": info.get("timestamp"), "frame_id": info.get("frame_id"), "sync_ok": info.get("sync_ok"), "sync_dt_ms": info.get("sync_dt_ms"), "role": role, **role_row, } ) samples_any_clipped += int(frame_any_clipped) report = { "meta_root": str(args.meta_root.resolve()), "module_params": str(args.module_params.resolve()), "radiometric_normalization_enabled": contract["enabled"], "method": contract["method"], "iso_base": contract["iso_base"], "files_discovered": len(meta_files), "files_with_any_valid_controls": len({row["meta_path"] for row in rows}), "complete_rgb_re_nir_samples": samples_complete, "samples_with_valid_controls": samples_with_valid_controls, "samples_with_any_clipped_role": samples_any_clipped, "samples_with_any_clipped_role_pct": ( 100.0 * samples_any_clipped / samples_with_valid_controls if samples_with_valid_controls else 0.0 ), "skipped_records": len(skipped), "global": { role: summarize_accumulator(global_acc[role], contract["roles"][role]) for role in ROLES }, "groups": { group: { role: summarize_accumulator(acc_by_role[role], contract["roles"][role]) for role in ROLES } for group, acc_by_role in sorted(group_acc.items()) }, } rows_path = args.out_dir / "radiometric_audit_rows.csv" summary_path = args.out_dir / "radiometric_audit_summary.json" skipped_path = args.out_dir / "radiometric_audit_skipped.csv" fieldnames = [ "group", "meta_path", "timestamp", "frame_id", "sync_ok", "sync_dt_ms", "role", "exposure_time_us", "sensitivity_iso", "analogue_gain_est", "gain_factor_used", "gain_source", "actual_factor", "reference_factor", "raw_scale", "applied_scale", "scale_min", "scale_max", "clipped_low", "clipped_high", "at_low_limit", "at_high_limit", ] with rows_path.open("w", newline="", encoding="utf-8-sig") as handle: writer = csv.DictWriter(handle, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() writer.writerows(rows) with skipped_path.open("w", newline="", encoding="utf-8-sig") as handle: writer = csv.DictWriter(handle, fieldnames=["path", "reason"]) writer.writeheader() writer.writerows(skipped) with summary_path.open("w", encoding="utf-8") as handle: json.dump(json_safe(report), handle, ensure_ascii=False, indent=2) print("=" * 64) print("Auditoria da normalizacao radiometrica") print(f"Metas encontrados : {len(meta_files)}") print(f"Amostras completas: {samples_complete}") print(f"Com algum clamp : {samples_any_clipped}") print("-" * 64) for role in ROLES: summary = report["global"][role] scale = summary["applied_scale"] clipping = summary["clipping"] rec = summary["recommendation"] print( f"{role.upper():>3} | n={summary['count']:5d} " f"scale_med={scale.get('p50', 0):7.3f} " f"scale_p05={scale.get('p05', 0):7.3f} " f"scale_p95={scale.get('p95', 0):7.3f} " f"clamp_low={clipping['low_pct']:6.2f}% " f"clamp_high={clipping['high_pct']:6.2f}% " f"ref_atual={summary['contract']['reference_factor']:.1f} " f"ref_mediana={float(rec['median_actual_factor'] or 0):.1f}" ) print("-" * 64) print(f"Rows : {rows_path}") print(f"Summary : {summary_path}") print(f"Skipped : {skipped_path}") print("=" * 64) return 0 if __name__ == "__main__": raise SystemExit(main())