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

338 lines
12 KiB
Python

import json
import argparse
import os
from datetime import datetime
def now_str():
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def load_json(path):
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
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 build_flatfield_config(flatfield_json_path, flatfield_data):
"""
Espera o JSON gerado pelo flatfield_calibration_tool_v2.py.
Estrutura esperada:
schema: multispec_flatfield_v1
outputs.npz: calibration/flatfield_maps_v1.npz
channels: ["R", "G", "B", "RE", "NIR"]
maps.CH.gain_key / flat_norm_key / ...
"""
if not isinstance(flatfield_data, dict):
return {
"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:
# Fallback: tenta deduzir pelo nome do json.
base, _ = os.path.splitext(flatfield_json_path)
npz_path = base + ".npz"
channels = flatfield_data.get("channels") or ["R", "G", "B", "RE", "NIR"]
channel_maps = {}
for ch in channels:
m = maps.get(ch, {}) or {}
channel_maps[ch] = {
"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"),
}
return {
"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", ""),
}
def pick_radiometric_config(radiometric_data: dict, selected_profile: str | None = None):
if not isinstance(radiometric_data, dict):
return None
# 1) Novo contrato: usa radiometric_config da raiz se existir.
root_cfg = radiometric_data.get("radiometric_config")
if isinstance(root_cfg, dict):
return root_cfg
# 2) Usa active_profile se existir.
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
# 3) Fallback explícito por argumento.
if selected_profile:
cfg = radiometric_data.get(selected_profile, {}).get("radiometric_config")
if isinstance(cfg, dict):
return cfg
return None
def main():
parser = argparse.ArgumentParser(
description="Monta o module_params.json unificando calibração de câmera, fusão, radiometria e flat-field.",
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("--out", default="calibration/module_params.json")
args = parser.parse_args()
cam_data = load_json(args.camera_json)
fusion_data = load_json(args.fusion_json)
radiometric_data = {}
if args.radiometric_json:
radiometric_data = load_json(args.radiometric_json)
flatfield_data = {}
flatfield_config = {
"enabled": False,
"reason": "flatfield não informado ou arquivo inexistente",
}
if not args.disable_flatfield and args.flatfield_json and os.path.isfile(args.flatfield_json):
flatfield_data = load_json(args.flatfield_json)
flatfield_config = build_flatfield_config(args.flatfield_json, flatfield_data)
elif args.disable_flatfield:
flatfield_config = {
"enabled": False,
"reason": "desabilitado via --disable_flatfield",
}
# =========================
# CAMERA SETTINGS
# =========================
camera_settings = cam_data.get("camera_settings")
if not isinstance(camera_settings, dict):
raise RuntimeError("camera_json sem camera_settings válido")
# =========================
# RGB CALIBRATION
# =========================
rgb_calibration = cam_data.get("rgb_calibration")
if not isinstance(rgb_calibration, dict):
rgb_calibration = {
"enabled": False,
"gains": {
"R": 1.0,
"G": 1.0,
"B": 1.0,
}
}
# =========================
# RADIOMETRIC
# =========================
radiometric_config = pick_radiometric_config(radiometric_data, selected_profile=args.radiometric_profile)
if not isinstance(radiometric_config, dict):
radiometric_config = cam_data.get("radiometric_config")
if not isinstance(radiometric_config, dict):
radiometric_config = {
"enabled": True,
"interval_s": 0.20,
"verbose": True,
"metering_mode": "global",
"spectral_control_mode": "shared",
"control_metric": "p50",
"target_value": 0.40,
"deadband": 0.04,
"p95_limit": 0.90,
"saturation_limit_pct": 0.50,
"saturation_hard_pct": 10.0,
"saturation_extreme_pct": 50.0,
"dark_limit_pct": 35.0,
"control_strategy": "ratio",
"ratio_alpha": 0.55,
"ratio_min": 0.55,
"ratio_max": 1.85,
"reduce_fast_factor": 0.70,
"gain_return_enabled": True,
"gain_return_factor": 0.50,
"gain_reduce_on_saturation": True,
"gain_hard_reset_on_saturation": False,
"gain_increase_required_cycles": 5,
"gain_decrease_required_cycles": 2,
"gain_step_up": 0.20,
"gain_step_down": 0.50,
"exp_high_ratio_for_gain": 0.95,
"exp_low_ratio_for_gain_return": 0.75,
"prefer_exposure": True,
"exp_min_us": 100,
"exp_max_us": 80000,
"gain_min": 1.0,
"gain_max": 4.0,
"role_limits": {
"rgb": {"exp_min_us": 100, "exp_max_us": 80000, "gain_min": 1.0, "gain_max": 4.0},
"re": {"exp_min_us": 100, "exp_max_us": 80000, "gain_min": 1.0, "gain_max": 3.0},
"nir": {"exp_min_us": 100, "exp_max_us": 80000, "gain_min": 1.0, "gain_max": 3.0},
},
"exp_apply_threshold_us": 40,
"gain_apply_threshold": 0.03,
"ready_required_cycles": 3,
"apply_same_spectral_to_both": True,
"spectral_roles": ["re", "nir"],
}
radiometric_normalization_config = radiometric_data.get("radiometric_normalization")
if not isinstance(radiometric_normalization_config, dict):
radiometric_normalization_config = cam_data.get("radiometric_normalization")
if not isinstance(radiometric_normalization_config, dict):
radiometric_normalization_config = {
"enabled": True,
"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,
}
patch_normalization_config = radiometric_data.get("patch_normalization")
if not isinstance(patch_normalization_config, dict):
patch_normalization_config = cam_data.get("patch_normalization")
if not isinstance(patch_normalization_config, dict):
patch_normalization_config = {
"enabled": False,
"apply_when_metering_mode": "reference_patches",
"apply_stage": "after_fusion",
"method": "gray_scale_with_white_guard",
"space": "multispec_tensor",
"targets": {
"black": 0.06,
"gray": 0.40,
"white": 0.78,
},
"white_guard_max": 0.92,
"scale_min": 0.35,
"scale_max": 2.50,
"clip_output": True,
"require_valid_gray": True,
"use_black_for_offset": False,
"save_patch_stats": True,
}
# =========================
# FUSION CONFIG
# =========================
fusion_config = {
"alignment_mode": fusion_data.get("alignment_mode", "manual_affine"),
"baseline_mm": fusion_data.get("baseline_mm", 75.0),
"reference_camera": fusion_data.get("reference_camera", "rgb"),
"manual_offsets": fusion_data.get("manual_offsets", {}),
"homographies": fusion_data.get("homographies", {}),
"crop_valid_common": fusion_data.get("crop_valid_common", True),
"resize_after_crop": fusion_data.get("resize_after_crop", True),
"target_size": fusion_data.get("target_size", None),
}
# =========================
# MODULE PARAMS FINAL
# =========================
module_params = {
"schema": "multispec_module_params_v3",
"saved_at": now_str(),
"frame_type": cam_data.get("frame_type", fusion_data.get("frame_type", "RAW_BRUTO")),
"capture_mode_requested": cam_data.get("capture_mode_requested", "AUTO"),
"capture_mode_effective": cam_data.get("capture_mode_effective", "AUTO"),
"raw_policy": cam_data.get("raw_policy", "allow_single"),
"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", "GBRG")),
"camera_settings": camera_settings,
"fusion_config": fusion_config,
"radiometric_config": radiometric_config,
"radiometric_normalization": radiometric_normalization_config,
"patch_normalization": patch_normalization_config,
"rgb_calibration": rgb_calibration,
"flatfield_config": flatfield_config,
}
with open(args.out, "w", encoding="utf-8") as f:
json.dump(module_params, f, ensure_ascii=False, indent=2)
print(f"[OK] module_params gerado em: {args.out}")
if flatfield_config.get("enabled"):
print(f"[OK] flatfield habilitado: {flatfield_config.get('npz_file')}")
else:
print(f"[WARN] flatfield desabilitado: {flatfield_config.get('reason')}")
if __name__ == "__main__":
main()