import json import argparse import os from copy import deepcopy from datetime import datetime # ============================================================ # Helpers # ============================================================ def now_str(): return datetime.now().strftime("%Y-%m-%d %H:%M:%S") def load_json(path, required=True): if not path or not os.path.isfile(path): if required: raise FileNotFoundError(f"Arquivo não encontrado: {path}") return {} with open(path, "r", encoding="utf-8") as f: return json.load(f) def save_json(path, data): out_dir = os.path.dirname(os.path.abspath(path)) if out_dir: os.makedirs(out_dir, exist_ok=True) with open(path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) f.write("\n") def rel_or_abs(path): """ Mantém o caminho como veio, mas normaliza separadores. Isso evita quebrar projetos Windows/Linux e deixa o module_params legível. """ if path is None: return None return str(path).replace("\\", "/") def deep_merge(base, update, *, skip_none=True): """ Merge recursivo seguro. - dict + dict: combina recursivamente. - listas/escalares: valor novo substitui o antigo. - None: por padrão NÃO apaga valor antigo, para evitar perder calibração quando um arquivo fonte não conhece determinada chave. """ if not isinstance(base, dict): base = {} out = deepcopy(base) if not isinstance(update, dict): return out for key, value in update.items(): if value is None and skip_none: continue if isinstance(value, dict) and isinstance(out.get(key), dict): out[key] = deep_merge(out[key], value, skip_none=skip_none) else: out[key] = deepcopy(value) return out def first_dict(*values): for value in values: if isinstance(value, dict): return value return None # ============================================================ # Defaults coerentes com RawProcessorCore + module_params atual # ============================================================ DEFAULT_RGB_PROCESSING = { "mode": "bayer_planes", } DEFAULT_PATCH_NORMALIZATION = { "enabled": True, "apply_when_metering_mode": "reference_patches", "apply_stage": "after_fusion", "method": "gray_scale_with_white_guard", "space": "multispec_tensor", "targets_by_patch_channel": { "black": { "R": 0.06, "G": 0.06, "B": 0.06, "RE": 0.06, "NIR": 0.06, }, "gray": { "R": 0.34, "G": 0.34, "B": 0.34, "RE": 0.24, "NIR": 0.30, }, "white": { "R": 0.78, "G": 0.78, "B": 0.78, "RE": 0.78, "NIR": 0.78, }, }, "white_guard_max": 0.92, "white_guard_max_by_channel": { "R": 0.92, "G": 0.92, "B": 0.92, "RE": 0.88, "NIR": 0.88, }, "scale_min": 0.35, "scale_max": 2.5, "clip_output": True, "require_valid_gray": True, "use_black_for_offset": False, "save_patch_stats": True, "rgb_saturation_guard_enabled": True, "rgb_saturation_guard_mode": "fade_strength", "rgb_saturation_soft_start": 0.88, "rgb_saturation_hard": 0.97, "rgb_saturation_threshold": 0.97, } DEFAULT_FLATFIELD_RUNTIME = { "strength": 0.35, "strength_by_channel": { "R": 0.9, "G": 0.9, "B": 0.9, "RE": 0.25, "NIR": 0.25, }, "gain_min_runtime": 0.75, "gain_max_runtime": 1.35, "runtime_smooth_ksize": 81, "saturation_guard_enabled": True, "saturation_guard_mode": "fade_strength", "saturation_guard_threshold": 0.97, "saturation_guard_soft_start": 0.88, "saturation_guard_hard": 0.97, } DEFAULT_RADIOMETRIC_NORMALIZATION = { "enabled": False, "method": "exposure_gain_reference", "apply_stage": "after_dark_before_flat_gain", "reference_controls": { "rgb": {"exposure_time_us": 3000, "analogue_gain": 1.0}, "re": {"exposure_time_us": 7000, "analogue_gain": 1.0}, "nir": {"exposure_time_us": 7000, "analogue_gain": 1.0}, }, "clip_output": True, } DEFAULT_RADIOMETRIC_CONFIG = { "enabled": True, "interval_s": 0.25, "verbose": True, "metering_mode": "reference_patches", "spectral_control_mode": "shared", "control_metric": "p50", "target_value": 0.5, "deadband": 0.055, "p95_limit": 0.975, "saturation_limit_pct": 5.0, "alpha": 0.18, "exp_step_gain": 0.55, "prefer_exposure": True, "exp_min_us": 100, "exp_max_us": 80000, "gain_min": 1.0, "gain_max": 4.0, "exp_apply_threshold_us": 15, "gain_apply_threshold": 0.05, "apply_same_spectral_to_both": True, "spectral_roles": ["re", "nir"], "dark_limit_pct": 35.0, "control_strategy": "ratio", "ratio_alpha": 0.42, "ratio_min": 0.72, "ratio_max": 1.38, "reduce_fast_factor": 0.8, "factor_min": 0.62, "factor_max": 1.42, "gain_return_enabled": True, "gain_reduce_on_saturation": True, "gain_increase_required_cycles": 3, "gain_decrease_required_cycles": 1, "gain_step_up": 0.3, "gain_step_down": 0.5, "gain_hard_reset_on_saturation": False, "exp_high_ratio_for_gain": 0.95, "exp_low_ratio_for_gain_return": 0.75, "role_limits": { "rgb": {"exp_min_us": 100, "exp_max_us": 80000, "gain_min": 1.0, "gain_max": 2.0}, "re": {"exp_min_us": 100, "exp_max_us": 3500, "gain_min": 1.0, "gain_max": 2.0}, "nir": {"exp_min_us": 100, "exp_max_us": 3500, "gain_min": 1.0, "gain_max": 2.0}, }, "ready_required_cycles": 3, "patch_control_mode": "gray_primary", "patch_require_order": True, "patch_min_separation": 0.08, "patch_white_sat_limit_pct": 5.0, "patch_white_p95_limit": 0.985, "patch_black_dark_limit_pct": 80.0, "patch_black_max_p50": 0.2, "patch_gray_min_p50": 0.08, "patch_gray_max_p50": 0.85, "patch_roi_contract": "multi_roi_by_role_v1", "patch_roi_reduce_method": "median_valid_rois", "patch_roi_outlier_reject": True, "patch_roi_max_p50_delta": 0.12, "global_saturation_guard_enabled": True, "global_guard_roi_pct": {"x0": 0.05, "y0": 0.05, "x1": 0.95, "y1": 0.76}, "global_guard_sat_threshold": 0.985, "global_guard_near_sat_threshold": 0.94, "global_guard_sat_pct_soft": 0.50, "global_guard_sat_pct_hard": 1.5, "global_guard_sat_pct_extreme": 5.0, "global_guard_blob_pct_soft": 0.20, "global_guard_blob_pct_hard": 0.80, "global_guard_blob_pct_extreme": 2.2, "global_guard_min_blob_px": 48, "global_guard_downsample_max_side": 320, "global_guard_reduce_factor_soft": 0.96, "global_guard_reduce_factor_hard": 0.82, "global_guard_reduce_factor_extreme": 0.62, "sun_guard_enabled": True, "sun_guard_p99_threshold": 0.96, "sun_guard_near_sat_pct_threshold": 2.0, "sun_guard_freeze_increase_cycles": 1, "sun_guard_allow_decrease": True, "guard_force_apply_enabled": True, "guard_force_apply_soft": False, "guard_force_apply_hard": True, "guard_force_apply_extreme": True, "guard_force_apply_on_patch_saturation": True, "guard_freeze_cycles_soft": 1, "guard_freeze_cycles_hard": 2, "guard_freeze_cycles_extreme": 3, "guard_reapply_min_exp_on_emergency": True, "guard_min_exp_margin_us": 80, "patch_two_roi_soften_risk": True, "patch_two_roi_white_risk_percentile": 75, "patch_two_roi_other_risk_percentile": 50, "patch_white_single_roi_saturation_reject": True, "patch_white_roi_reject_sat_pct": 5.0, "patch_white_roi_reject_p95": 0.995, } def default_module_template(): return { "schema": "multispec_module_params_v3", "saved_at": now_str(), "frame_type": "RAW_BRUTO", "capture_mode_requested": "AUTO", "capture_mode_effective": "AUTO", "raw_policy": "allow_single", "sensor_width": 1280, "sensor_height": 800, "bayer_pattern": "BGGR", "rgb_processing": deepcopy(DEFAULT_RGB_PROCESSING), "camera_settings": {}, "fusion_config": { "alignment_mode": "manual_affine", "baseline_mm": 75.0, "reference_camera": "rgb", "manual_offsets": { "re": {"dx": 0, "dy": 0, "theta_deg": 0.0}, "nir": {"dx": 0, "dy": 0, "theta_deg": 0.0}, }, "homographies": { "re_to_rgb": None, "nir_to_rgb": None, }, "crop_valid_common": True, "resize_after_crop": True, "target_size": None, }, "radiometric_config": deepcopy(DEFAULT_RADIOMETRIC_CONFIG), "radiometric_normalization": deepcopy(DEFAULT_RADIOMETRIC_NORMALIZATION), "patch_normalization": deepcopy(DEFAULT_PATCH_NORMALIZATION), "rgb_calibration": { "enabled": False, "gains": {"R": 1.0, "G": 1.0, "B": 1.0}, }, "flatfield_config": deep_merge( { "enabled": False, "reason": "flatfield não informado ou arquivo inexistente", "subtract_dark": True, "apply_before_fusion": True, "apply_after_decode": True, "apply_space": "native_camera_space", "map_type": "gain", "channels": ["R", "G", "B", "RE", "NIR"], "channel_maps": {}, "clip_output": True, }, DEFAULT_FLATFIELD_RUNTIME, ), } # ============================================================ # Builders / normalizers # ============================================================ def build_flatfield_config(flatfield_json_path, flatfield_data, previous_flatfield_config=None): """ Espera o JSON gerado pelo flatfield_calibration_tool_v2.py. Importante: usa previous_flatfield_config como base para preservar knobs runtime que não existem no arquivo de calibração do flat-field, como strength, gain_min_runtime, runtime_smooth_ksize e saturation_guard_*. """ base = deep_merge( deep_merge({}, previous_flatfield_config or {}), DEFAULT_FLATFIELD_RUNTIME, ) if not isinstance(flatfield_data, dict): return deep_merge(base, { "enabled": False, "reason": "flatfield_json ausente ou inválido", }) outputs = flatfield_data.get("outputs", {}) or {} maps = flatfield_data.get("maps", {}) or {} npz_path = outputs.get("npz") if not npz_path: base_name, _ = os.path.splitext(flatfield_json_path) npz_path = base_name + ".npz" channels = flatfield_data.get("channels") or base.get("channels") or ["R", "G", "B", "RE", "NIR"] channel_maps = {} previous_channel_maps = base.get("channel_maps", {}) or {} for ch in channels: m = maps.get(ch, {}) or {} prev = previous_channel_maps.get(ch, {}) or {} channel_maps[ch] = deep_merge(prev, { "gain_key": m.get("gain_key", f"gain_{ch}"), "flat_norm_key": m.get("flat_norm_key", f"flat_norm_{ch}"), "white_median_key": m.get("white_median_key", f"white_median_{ch}"), "dark_median_key": m.get("dark_median_key", f"dark_median_{ch}"), "shape": m.get("shape"), "gain_min": m.get("gain_min"), "gain_max": m.get("gain_max"), "gain_mean": m.get("gain_mean"), "gain_std": m.get("gain_std"), }) generated = { "enabled": True, "subtract_dark": True, "schema": flatfield_data.get("schema", "multispec_flatfield_v1"), "created_at": flatfield_data.get("created_at"), "json_file": rel_or_abs(flatfield_json_path), "npz_file": rel_or_abs(npz_path), "apply_before_fusion": True, "apply_after_decode": True, "apply_space": "native_camera_space", "map_type": "gain", "formula": "channel_corrected = max(channel_linear - dark, 0) * gain_map", "channels": channels, "channel_maps": channel_maps, "exp_gain_correct_during_flat_capture": bool(flatfield_data.get("exp_gain_correct", False)), "smooth_ksize": flatfield_data.get("smooth_ksize"), "min_gain": flatfield_data.get("min_gain"), "max_gain": flatfield_data.get("max_gain"), "notes": flatfield_data.get("notes", ""), } return deep_merge(base, generated) def pick_radiometric_config(radiometric_data: dict, selected_profile: str | None = None): if not isinstance(radiometric_data, dict): return None root_cfg = radiometric_data.get("radiometric_config") if isinstance(root_cfg, dict): return root_cfg active_profile = radiometric_data.get("active_profile") if active_profile in ("global_scene_mode", "three_reference_patches_mode"): cfg = radiometric_data.get(active_profile, {}).get("radiometric_config") if isinstance(cfg, dict): return cfg if selected_profile: cfg = radiometric_data.get(selected_profile, {}).get("radiometric_config") if isinstance(cfg, dict): return cfg return None def normalize_patch_normalization_contract(base_patch_config, incoming_patch_config=None): """ Garante o contrato atual do RawProcessorCore. - Sempre tem targets_by_patch_channel. - Preserva white_guard_max_by_channel. - Preserva rgb_saturation_guard_*. - Remove a chave legada targets, porque ela não é usada pelo core atual. """ cfg = deep_merge(DEFAULT_PATCH_NORMALIZATION, base_patch_config or {}) cfg = deep_merge(cfg, incoming_patch_config or {}) legacy_targets = cfg.pop("targets", None) if isinstance(legacy_targets, dict) and "targets_by_patch_channel" not in cfg: # Fallback conservador. Na prática, com DEFAULT_PATCH_NORMALIZATION acima, # normalmente não entra aqui. Mantido só para arquivos muito antigos. t = deepcopy(DEFAULT_PATCH_NORMALIZATION["targets_by_patch_channel"]) for patch_type in ("black", "gray", "white"): if patch_type in legacy_targets: scalar = legacy_targets.get(patch_type) try: scalar = float(scalar) for ch in ("R", "G", "B", "RE", "NIR"): t[patch_type][ch] = scalar except Exception: pass cfg["targets_by_patch_channel"] = t cfg = deep_merge(DEFAULT_PATCH_NORMALIZATION, cfg) return cfg def normalize_radiometric_config(base_rad_config, incoming_rad_config=None): cfg = deep_merge(DEFAULT_RADIOMETRIC_CONFIG, base_rad_config or {}) cfg = deep_merge(cfg, incoming_rad_config or {}) return cfg def normalize_radiometric_normalization(base_config, incoming_config=None): cfg = deep_merge(DEFAULT_RADIOMETRIC_NORMALIZATION, base_config or {}) cfg = deep_merge(cfg, incoming_config or {}) cfg["enabled"] = bool(cfg.get("enabled", False)) return cfg def build_fusion_config(fusion_data, previous_fusion_config=None): base = previous_fusion_config or {} generated = { "alignment_mode": fusion_data.get("alignment_mode"), "baseline_mm": fusion_data.get("baseline_mm"), "reference_camera": fusion_data.get("reference_camera"), "manual_offsets": fusion_data.get("manual_offsets"), "homographies": fusion_data.get("homographies"), "crop_valid_common": fusion_data.get("crop_valid_common"), "resize_after_crop": fusion_data.get("resize_after_crop"), "target_size": fusion_data.get("target_size"), } cfg = deep_merge(default_module_template()["fusion_config"], base) cfg = deep_merge(cfg, generated) return cfg def load_base_module(args): """ Carrega defaults do module_params atual. Prioridade: 1. --base_module_json, se informado. 2. --out, se já existir. 3. template interno coerente com o contrato atual. """ candidates = [] if args.base_module_json: candidates.append(args.base_module_json) if args.out: candidates.append(args.out) for path in candidates: if path and os.path.isfile(path): print(f"[INFO] Usando module_params base: {path}") return deep_merge(default_module_template(), load_json(path, required=True)) print("[WARN] Nenhum module_params base encontrado. Usando defaults internos.") return default_module_template() # ============================================================ # Main # ============================================================ def main(): parser = argparse.ArgumentParser( description="Monta o module_params.json preservando o contrato atual do RawProcessorCore.", formatter_class=argparse.ArgumentDefaultsHelpFormatter, ) parser.add_argument("--camera_json", default="calibration/sensor_calibration.json") parser.add_argument("--fusion_json", default="calibration/manual_offsets.json") parser.add_argument("--radiometric_json", default="calibration/radiometric_config.json") parser.add_argument( "--radiometric_profile", default="global_scene_mode", choices=["global_scene_mode", "three_reference_patches_mode"], ) parser.add_argument("--flatfield_json", default="calibration/flatfield_maps_v1.json") parser.add_argument("--disable_flatfield", action="store_true") parser.add_argument("--base_module_json", default=None, help="module_params atual usado como defaults antes de sobrescrever") parser.add_argument("--out", default="calibration/module_params.json") args = parser.parse_args() module_base = load_base_module(args) cam_data = load_json(args.camera_json, required=True) fusion_data = load_json(args.fusion_json, required=True) radiometric_data = load_json(args.radiometric_json, required=False) if args.radiometric_json else {} # ========================= # MODULE PARAMS FINAL # ========================= module_params = deepcopy(module_base) module_params["schema"] = "multispec_module_params_v3" module_params["saved_at"] = now_str() # ========================= # ROOT / CAMERA # ========================= root_updates = { "frame_type": cam_data.get("frame_type", fusion_data.get("frame_type")), "capture_mode_requested": cam_data.get("capture_mode_requested"), "capture_mode_effective": cam_data.get("capture_mode_effective"), "raw_policy": cam_data.get("raw_policy"), "sensor_width": cam_data.get("sensor_width", fusion_data.get("sensor_width")), "sensor_height": cam_data.get("sensor_height", fusion_data.get("sensor_height")), "bayer_pattern": cam_data.get("bayer_pattern", fusion_data.get("bayer_pattern")), } module_params = deep_merge(module_params, root_updates) module_params["rgb_processing"] = deep_merge( deep_merge(DEFAULT_RGB_PROCESSING, module_base.get("rgb_processing", {})), cam_data.get("rgb_processing") if isinstance(cam_data.get("rgb_processing"), dict) else {}, ) camera_settings = cam_data.get("camera_settings") if not isinstance(camera_settings, dict): camera_settings = module_base.get("camera_settings") if not isinstance(camera_settings, dict): raise RuntimeError("camera_json sem camera_settings válido e sem fallback no module_params base") module_params["camera_settings"] = camera_settings module_params["rgb_calibration"] = deep_merge( module_base.get("rgb_calibration", {}), cam_data.get("rgb_calibration") if isinstance(cam_data.get("rgb_calibration"), dict) else {}, ) # ========================= # FUSION # ========================= module_params["fusion_config"] = build_fusion_config( fusion_data, previous_fusion_config=module_base.get("fusion_config", {}), ) # ========================= # RADIOMETRIC # ========================= incoming_rad = pick_radiometric_config(radiometric_data, selected_profile=args.radiometric_profile) if not isinstance(incoming_rad, dict): incoming_rad = cam_data.get("radiometric_config") if isinstance(cam_data.get("radiometric_config"), dict) else {} module_params["radiometric_config"] = normalize_radiometric_config( module_base.get("radiometric_config", {}), incoming_rad, ) incoming_rad_norm = first_dict( radiometric_data.get("radiometric_normalization") if isinstance(radiometric_data, dict) else None, cam_data.get("radiometric_normalization") if isinstance(cam_data, dict) else None, ) or {} module_params["radiometric_normalization"] = normalize_radiometric_normalization( module_base.get("radiometric_normalization", {}), incoming_rad_norm, ) incoming_patch_norm = first_dict( radiometric_data.get("patch_normalization") if isinstance(radiometric_data, dict) else None, cam_data.get("patch_normalization") if isinstance(cam_data, dict) else None, ) or {} module_params["patch_normalization"] = normalize_patch_normalization_contract( module_base.get("patch_normalization", {}), incoming_patch_norm, ) # ========================= # FLATFIELD # ========================= previous_flatfield = module_base.get("flatfield_config", {}) or {} if args.disable_flatfield: module_params["flatfield_config"] = deep_merge(previous_flatfield, { "enabled": False, "reason": "desabilitado via --disable_flatfield", }) elif args.flatfield_json and os.path.isfile(args.flatfield_json): flatfield_data = load_json(args.flatfield_json, required=True) module_params["flatfield_config"] = build_flatfield_config( args.flatfield_json, flatfield_data, previous_flatfield_config=previous_flatfield, ) else: # Não achou novo flatfield: preserva o anterior se já existia. module_params["flatfield_config"] = deep_merge(previous_flatfield, DEFAULT_FLATFIELD_RUNTIME) if not module_params["flatfield_config"].get("enabled", False): module_params["flatfield_config"]["reason"] = "flatfield não informado ou arquivo inexistente" # ========================= # Save # ========================= save_json(args.out, module_params) print(f"[OK] module_params gerado em: {args.out}") print("[OK] contrato preservado: rgb_processing, patch_normalization, radiometric_config e knobs runtime do flatfield") flat_cfg = module_params.get("flatfield_config", {}) or {} if flat_cfg.get("enabled"): print(f"[OK] flatfield habilitado: {flat_cfg.get('npz_file')}") else: print(f"[WARN] flatfield desabilitado: {flat_cfg.get('reason')}") if __name__ == "__main__": main()