From f0f175d7519b51ad9f5d254277ab1bf9f56b2ba7 Mon Sep 17 00:00:00 2001 From: Diego Freitas Date: Tue, 5 May 2026 14:39:58 -0300 Subject: [PATCH] ajustes no modulo multiespectral oak-fcc-3 --- Python/OAK/datasets/oak-fcc-3/_0_capture.py | 54 +- .../oak-fcc-3/calibration/manual_offsets.json | 27 +- .../oak-fcc-3/calibration/module_params.json | 66 +- .../calibration/sensor_calibration.json | 222 +----- Python/OAK/datasets/oak-fcc-3/config.json | 10 +- .../oak-fcc-3/core/oak_fcc3_client.py | 30 +- .../oak-fcc-3/core/raw_processor_core.py | 28 +- .../oak-fcc-3/core/raw_processor_preview.py | 11 +- .../oak-fcc-3/tests/test_oak_fcc3_client.py | 22 +- .../oak-fcc-3/utils/build_module_params.py | 36 +- .../oak-fcc-3/utils/check_saved_files.py | 37 +- .../utils/manual_fusion_calibrator.py | 583 ++++++++------- .../utils/sensor_calibration_tool.py | 683 +++++++++++------- 13 files changed, 984 insertions(+), 825 deletions(-) diff --git a/Python/OAK/datasets/oak-fcc-3/_0_capture.py b/Python/OAK/datasets/oak-fcc-3/_0_capture.py index 9a26f8748..3fbccd191 100644 --- a/Python/OAK/datasets/oak-fcc-3/_0_capture.py +++ b/Python/OAK/datasets/oak-fcc-3/_0_capture.py @@ -144,13 +144,12 @@ def main(): parser.add_argument("--height", type=int, default=RAW_SIZE[1], help="Altura óptica da câmera.") parser.add_argument("--interval", type=float, default=1.0, help="Intervalo em segundos para auto-save quando ligado.") parser.add_argument("--preview_upscale", type=int, default=2, help="Fator de upscale visual do preview.") - parser.add_argument("--bayer", default="GBRG", choices=["GBRG", "GRBG", "RGGB", "BGGR"], help="Padrão Bayer das câmeras.") + parser.add_argument("--bayer", default="RGGB", choices=["GBRG", "GRBG", "RGGB", "BGGR"], help="Padrão Bayer das câmeras.") parser.add_argument("--output_dtype", default="float32", choices=["uint8", "uint16", "float32"], help="Dtype do payload processado no Pi.") parser.add_argument("--frame_type", default="RAW_BRUTO", choices=["RAW_BRUTO", "RGB", "MULTISPEC"], help="Tipo de payload pedido ao Pi.") parser.add_argument("--capture_mode", default="AUTO", choices=["AUTO", "SINGLE", "DOUBLE", "TRIPLE"], help="Modo de captura desejado no módulo.") parser.add_argument("--raw_policy", default="allow_single", choices=["allow_single", "require_triple"], help="Quando frame_type=RAW_BRUTO, define se o script aceita 1 câmera ou exige 3.") parser.add_argument("--module_calibration_json", default=MODULE_PARAMS, help="JSON salvo pelo calibrador de sensores com parâmetros fixos por câmera.") - parser.add_argument("--radiometric_ae", action="store_true", help="Liga controle automatico de exposicao radiometrico") args = parser.parse_args() @@ -179,6 +178,8 @@ def main(): print(f"RAW policy : {args.raw_policy}") print("============================================") + beauty_preview = False + radiometric_ae = True auto_save = False last_auto_t = 0.0 preview_upscale = args.preview_upscale @@ -195,7 +196,7 @@ def main(): last_msg = "" last_msg_t = 0.0 - window_name = "Dataset Capture (C/SPACE=save | A=auto-save | M=preview scale | Q=quit)" + window_name = "Dataset Capture (C/SPACE=save | A=auto-save | M=preview | R=rad | Q=quit)" cv2.namedWindow(window_name, cv2.WINDOW_NORMAL) last_frame_id = -1 @@ -215,8 +216,8 @@ def main(): output_dtype=args.output_dtype, capture_mode=effective_capture_mode, raw_policy=args.raw_policy, - #module_calibration_json=args.module_calibration_json, - radiometric_enabled=args.radiometric_ae, + module_calibration_json=args.module_calibration_json, + radiometric_enabled=radiometric_ae, ) as cam: while True: t0 = time.time() @@ -227,7 +228,7 @@ def main(): try: frame_type = meta.get("frame_type", "RAW_BRUTO") dtype_str = meta.get("dtype") or meta.get("output_dtype", "uint8") - preview_source_id = "cam2" + preview_source_id = "rgb" if frame_type == "RAW_BRUTO": if isinstance(frame, dict): @@ -235,16 +236,22 @@ def main(): preview_bgr, raw3_preview, preview_source_id = cam.build_preview_from_raw_payload(frame=frame, meta=meta) + if beauty_preview: + rgb_preview = None + previews = cam.build_visual_preview_from_raw(frame, meta) + camera_info = meta.get("camera_info", {}) or {} + for cam_id, img in previews.items(): + role = camera_info.get(cam_id, {}).get("role") + if role == "rgb": + rgb_preview = img + preview_source_id = cam_id + break + if rgb_preview is not None: + preview_bgr = rgb_preview + last_packed_raw = None last_packed_raw_by_camera = {cam_id: arr.copy() for cam_id, arr in packed_by_camera.items()} last_payload_float = raw3_preview.copy() - - else: - preview_bgr, raw3_preview, preview_source_id = cam.build_preview_from_raw_payload(frame=frame, meta=meta) - - last_packed_raw = frame.copy() - last_packed_raw_by_camera = None - last_payload_float = raw3_preview.copy() elif frame_type == "RGB": rgb_chw = frame @@ -334,9 +341,9 @@ def main(): st = rad.state line_rad = ( f"RAD | " - f"RGB(exp={st['cam2']['exp']}, g={st['cam2']['gain']:.2f}) | " - f"RE(exp={st['cam0']['exp']}, g={st['cam0']['gain']:.2f}) | " - f"NIR(exp={st['cam1']['exp']}, g={st['cam1']['gain']:.2f})" + f"RGB(exp={st['rgb']['exp']}, g={st['rgb']['gain']:.2f}) | " + f"RE(exp={st['re']['exp']}, g={st['re']['gain']:.2f}) | " + f"NIR(exp={st['nir']['exp']}, g={st['nir']['gain']:.2f})" ) else: line_rad = "RAD | OFF" @@ -349,7 +356,7 @@ def main(): f"codec={meta.get('codec_name', meta.get('codec_family', '-'))} | comp={meta.get('dt_comp', 0):.4f}s | send={meta.get('dt_send_payload_prev', 0):.4f}s", f"CAM_PARAMS={os.path.basename(args.module_calibration_json)} | controles fixos aplicados", line_rad, - "Keys: C/SPACE=save | A=auto-save | M=preview | Q/Esc=quit" + "Keys: C/SPACE=save | A=auto-save | M=preview | R=rad | Q/Esc=quit" ] overlay_hud(preview_show, lines, base_h=raw_h) @@ -425,10 +432,19 @@ def main(): last_msg_t = time.time() elif k in (ord("m"), ord("M")): - preview_upscale = 0 if preview_upscale else args.preview_upscale - last_msg = f"Preview UPSCALE -> {preview_upscale}" + #preview_upscale = 0 if preview_upscale else args.preview_upscale + beauty_preview = False if beauty_preview else True + last_msg = f"Preview Beauty -> {beauty_preview}" last_msg_t = time.time() + elif k in (ord("r"), ord("R")): + radiometric_ae = False if radiometric_ae else True + rad = getattr(cam, "radiometric_controller", None) + rad.enabled = radiometric_ae + last_msg = f"RAD -> {radiometric_ae}" + last_msg_t = time.time() + + elif k in (ord("c"), ord("C"), 32): if can_save: frame_type_save = last_meta_stream.get("frame_type") diff --git a/Python/OAK/datasets/oak-fcc-3/calibration/manual_offsets.json b/Python/OAK/datasets/oak-fcc-3/calibration/manual_offsets.json index f18715d6b..ee901a0e2 100644 --- a/Python/OAK/datasets/oak-fcc-3/calibration/manual_offsets.json +++ b/Python/OAK/datasets/oak-fcc-3/calibration/manual_offsets.json @@ -1,34 +1,31 @@ { - "schema": "manual_multispec_offsets_v1", - "saved_at": "2026-05-04 20:50:02", - "pi_host": "192.168.105.6", - "pc_host": "192.168.105.5", - "stream_port": 6001, + "schema": "manual_multispec_offsets_v2", + "saved_at": "2026-05-05 12:39:55", "frame_type": "RAW_BRUTO", "capture_mode_requested": "AUTO", "capture_mode_effective": "AUTO", "raw_policy": "allow_single", "sensor_width": 640, "sensor_height": 480, - "bayer_pattern": "GBRG", - "reference_camera": "cam2", + "bayer_pattern": "RGGB", + "reference_camera": "rgb", "baseline_mm": 75.0, "alignment_mode": "manual_affine", "manual_offsets": { - "cam0": { - "dx": -28, - "dy": 9, + "re": { + "dx": -8, + "dy": 39, "theta_deg": 0.0 }, - "cam1": { - "dx": -4, - "dy": 31, + "nir": { + "dx": -2, + "dy": 18, "theta_deg": 0.0 } }, "homographies": { - "cam0_to_cam2": null, - "cam1_to_cam2": null + "re_to_rgb": null, + "nir_to_rgb": null }, "notes": "" } \ No newline at end of file diff --git a/Python/OAK/datasets/oak-fcc-3/calibration/module_params.json b/Python/OAK/datasets/oak-fcc-3/calibration/module_params.json index 4ea95cc57..ec8c27d67 100644 --- a/Python/OAK/datasets/oak-fcc-3/calibration/module_params.json +++ b/Python/OAK/datasets/oak-fcc-3/calibration/module_params.json @@ -1,58 +1,58 @@ { - "schema": "multispec_module_params_v1", - "saved_at": "2026-05-05 08:03:18", + "schema": "multispec_module_params_v2", + "saved_at": "2026-05-05 13:57:20", "frame_type": "RAW_BRUTO", "capture_mode_requested": "AUTO", "capture_mode_effective": "AUTO", "raw_policy": "allow_single", "sensor_width": 640, "sensor_height": 480, - "bayer_pattern": "GBRG", + "bayer_pattern": "RGGB", "camera_settings": { - "cam0": { + "rgb": { "ae_enable": false, "awb_enable": false, - "exposure_time_us": 15000, - "analogue_gain": 1.0, - "colour_gains": null - }, - "cam1": { - "ae_enable": false, - "awb_enable": false, - "exposure_time_us": 15000, - "analogue_gain": 1.0, - "colour_gains": null - }, - "cam2": { - "ae_enable": true, - "awb_enable": true, - "exposure_time_us": 15000, - "analogue_gain": 1.0, + "exposure_time_us": 20000, + "analogue_gain": 1.2100000000000002, "colour_gains": [ 1.0, 1.0 ] + }, + "re": { + "ae_enable": false, + "awb_enable": false, + "exposure_time_us": 20000, + "analogue_gain": 1.0, + "colour_gains": null + }, + "nir": { + "ae_enable": false, + "awb_enable": false, + "exposure_time_us": 20000, + "analogue_gain": 1.0, + "colour_gains": null } }, "fusion_config": { "alignment_mode": "manual_affine", "baseline_mm": 75.0, - "reference_camera": "cam2", + "reference_camera": "rgb", "manual_offsets": { - "cam0": { - "dx": -28, - "dy": 9, + "re": { + "dx": -8, + "dy": 39, "theta_deg": 0.0 }, - "cam1": { - "dx": -4, - "dy": 31, + "nir": { + "dx": -2, + "dy": 18, "theta_deg": 0.0 } }, "homographies": { - "cam0_to_cam2": null, - "cam1_to_cam2": null + "re_to_rgb": null, + "nir_to_rgb": null }, "crop_valid_common": true, "resize_after_crop": true, @@ -74,5 +74,13 @@ "verbose": true, "exp_apply_threshold_us": 50, "gain_apply_threshold": 0.02 + }, + "rgb_calibration": { + "enabled": true, + "gains": { + "R": 1.3000000000000003, + "G": 1.0, + "B": 1.5500000000000005 + } } } \ No newline at end of file diff --git a/Python/OAK/datasets/oak-fcc-3/calibration/sensor_calibration.json b/Python/OAK/datasets/oak-fcc-3/calibration/sensor_calibration.json index 1579e5b97..2459010a2 100644 --- a/Python/OAK/datasets/oak-fcc-3/calibration/sensor_calibration.json +++ b/Python/OAK/datasets/oak-fcc-3/calibration/sensor_calibration.json @@ -1,211 +1,51 @@ { - "schema": "multispec_camera_params_v1", - "saved_at": "2026-05-04 19:29:24", - "pi_host": "192.168.105.6", - "pc_host": "192.168.105.5", - "stream_port": 6001, + "schema": "multispec_camera_params_v2", + "saved_at": "2026-05-05 13:53:08", "frame_type": "RAW_BRUTO", "capture_mode_requested": "AUTO", "capture_mode_effective": "AUTO", "raw_policy": "allow_single", "sensor_width": 640, "sensor_height": 480, - "bayer_pattern": "GBRG", + "bayer_pattern": "RGGB", "camera_settings": { - "cam0": { + "rgb": { "ae_enable": false, "awb_enable": false, - "exposure_time_us": 15000, - "analogue_gain": 1.0, - "colour_gains": null - }, - "cam1": { - "ae_enable": false, - "awb_enable": false, - "exposure_time_us": 15000, - "analogue_gain": 1.0, - "colour_gains": null - }, - "cam2": { - "ae_enable": true, - "awb_enable": true, - "exposure_time_us": 15000, - "analogue_gain": 1.0, + "exposure_time_us": 20000, + "analogue_gain": 1.2100000000000002, "colour_gains": [ 1.0, 1.0 ] + }, + "re": { + "ae_enable": false, + "awb_enable": false, + "exposure_time_us": 20000, + "analogue_gain": 1.0, + "colour_gains": null + }, + "nir": { + "ae_enable": false, + "awb_enable": false, + "exposure_time_us": 20000, + "analogue_gain": 1.0, + "colour_gains": null + } + }, + "rgb_calibration": { + "enabled": true, + "gains": { + "R": 1.3000000000000003, + "G": 1.0, + "B": 1.5500000000000005 } }, "rois": { - "cam2": [ - { - "name": "mesa", - "type": "polygon", - "points": [ - [ - 429, - 305 - ], - [ - 446, - 195 - ], - [ - 512, - 199 - ], - [ - 512, - 309 - ] - ], - "color": [ - 0, - 255, - 255 - ] - }, - { - "name": "teto", - "type": "polygon", - "points": [ - [ - 148, - 345 - ], - [ - 153, - 269 - ], - [ - 216, - 265 - ], - [ - 221, - 345 - ] - ], - "color": [ - 0, - 255, - 0 - ] - } - ], - "cam0": [ - { - "name": "mesa", - "type": "polygon", - "points": [ - [ - 417, - 235 - ], - [ - 433, - 158 - ], - [ - 485, - 162 - ], - [ - 479, - 237 - ] - ], - "color": [ - 0, - 255, - 255 - ] - }, - { - "name": "tet", - "type": "polygon", - "points": [ - [ - 218, - 306 - ], - [ - 219, - 230 - ], - [ - 274, - 225 - ], - [ - 276, - 305 - ] - ], - "color": [ - 0, - 255, - 0 - ] - } - ], - "cam1": [ - { - "name": "mesa", - "type": "polygon", - "points": [ - [ - 468, - 227 - ], - [ - 480, - 123 - ], - [ - 556, - 137 - ], - [ - 549, - 233 - ] - ], - "color": [ - 0, - 255, - 255 - ] - }, - { - "name": "teto", - "type": "polygon", - "points": [ - [ - 139, - 333 - ], - [ - 157, - 238 - ], - [ - 237, - 258 - ], - [ - 222, - 351 - ] - ], - "color": [ - 0, - 255, - 0 - ] - } - ] + "rgb": [], + "re": [], + "nir": [] }, "snapshots": [], "notes": "", diff --git a/Python/OAK/datasets/oak-fcc-3/config.json b/Python/OAK/datasets/oak-fcc-3/config.json index 7c3a6d69a..ea741c0f3 100644 --- a/Python/OAK/datasets/oak-fcc-3/config.json +++ b/Python/OAK/datasets/oak-fcc-3/config.json @@ -1,20 +1,20 @@ { "camera": "oak-fcc-3", "modelo": "segformer_b1", - "model_name": "pulv_mit", + "model_name": "oak_mit", "dual_head": false, "main_class_name": "cana", "es_classes": "", "model_to_use": "geral", - "raw_size": [640, 480], - "resolucao": [1024, 800], + "raw_size": [1280, 800], + "resolucao": [1024, 640], "roi_inicio": 0.0, "roi_tamanho": 1.0, "shaves": 3, - "channels": 4, + "channels": 5, "use_ndvi": false, "backbone": "nvidia/mit-b1", "fusion_mode": "stacked", - "stats_source_tag": "stacked_raw4", + "stats_source_tag": "stacked_raw5", "module_params_json": "calibration/module_params.json" } \ No newline at end of file diff --git a/Python/OAK/datasets/oak-fcc-3/core/oak_fcc3_client.py b/Python/OAK/datasets/oak-fcc-3/core/oak_fcc3_client.py index ef5d6daf6..62288157f 100644 --- a/Python/OAK/datasets/oak-fcc-3/core/oak_fcc3_client.py +++ b/Python/OAK/datasets/oak-fcc-3/core/oak_fcc3_client.py @@ -14,7 +14,7 @@ class OakFcc3Client: self, width=640, height=400, - bayer="GBRG", + bayer="RGGB", fps=30, frame_type="RAW_BRUTO", output_dtype="uint8", @@ -196,6 +196,32 @@ class OakFcc3Client: return frame, meta, decoded + def get_next_tensor_preview(self, timeout=2.0): + frame, meta, decoded = self.get_next_decoded(timeout=timeout) + + frame_type = str(meta.get("frame_type", self.frame_type)).upper() + + if frame_type == "RGB": + rgb_hwc = np.transpose(frame[:3], (1, 2, 0)) + preview = self._rgb01_to_bgr(rgb_hwc) + return {"rgb_tensor": preview}, meta + + if frame_type == "MULTISPEC": + rgb_hwc = np.transpose(frame[:3], (1, 2, 0)) + re01 = frame[3] + nir01 = frame[4] + + return { + "rgb_tensor": self._rgb01_to_bgr(rgb_hwc), + "re_tensor": self._gray01_to_bgr(re01), + "nir_tensor": self._gray01_to_bgr(nir01), + }, meta + + else: + return self.build_visual_preview_from_raw(frame, meta), meta + + raise RuntimeError(f"frame_type não suportado para preview: {frame_type}") + def get_next_preview(self, timeout=2.0): raw_frame, raw_meta = self.get_next_raw_frame(timeout=timeout) @@ -259,7 +285,7 @@ class OakFcc3Client: return np.ascontiguousarray(tensor.astype(np.float32, copy=False)) def build_multispec_tensor(self, decoded, meta=None): - tensor = self.core.fuse_multispec_cameras( + tensor = self.build_infer_tensor_from_decoded( decoded=decoded, meta=meta, channels_expected=5, diff --git a/Python/OAK/datasets/oak-fcc-3/core/raw_processor_core.py b/Python/OAK/datasets/oak-fcc-3/core/raw_processor_core.py index c1abfdb73..bfd80ab58 100644 --- a/Python/OAK/datasets/oak-fcc-3/core/raw_processor_core.py +++ b/Python/OAK/datasets/oak-fcc-3/core/raw_processor_core.py @@ -26,6 +26,14 @@ class RawProcessorCore: "resize_after_crop": True, "target_size": None, } + self.rgb_calibration = { + "enabled": False, + "gains": { + "R": 1.0, + "G": 1.0, + "B": 1.0 + } + } if calibration_json_path: self.load_fusion_config_json(calibration_json_path) @@ -103,10 +111,10 @@ class RawProcessorCore: b = raw16[1::2, 0::2] g2 = raw16[1::2, 1::2] elif p == "RGGB": - r = raw16[0::2, 0::2] + b = raw16[0::2, 0::2] g1 = raw16[0::2, 1::2] g2 = raw16[1::2, 0::2] - b = raw16[1::2, 1::2] + r = raw16[1::2, 1::2] elif p == "BGGR": b = raw16[0::2, 0::2] g1 = raw16[0::2, 1::2] @@ -131,6 +139,13 @@ class RawProcessorCore: g = ((ch["G1"].astype(np.float32) + ch["G2"].astype(np.float32)) * 0.5) / max_val b = ch["B"].astype(np.float32) / max_val + rgb_cal = getattr(self, "rgb_calibration", {}) or {} + if rgb_cal.get("enabled", False): + gains = rgb_cal.get("gains", {}) or {} + r *= float(gains.get("R", 1.0)) + g *= float(gains.get("G", 1.0)) + b *= float(gains.get("B", 1.0)) + chw = np.stack([r, g, b], axis=0).astype(np.float32) chw = np.clip(chw, 0.0, 1.0) @@ -823,11 +838,14 @@ class RawProcessorCore: data = json.load(f) fusion = data.get("fusion_config") - if not isinstance(fusion, dict): + if isinstance(fusion, dict): + self.fusion_config = self._merge_fusion_config(self.fusion_config, fusion) + else: print("[WARN] JSON sem fusion_config. Mantendo config padrão.") - return - self.fusion_config = self._merge_fusion_config(self.fusion_config, fusion) + rgb_cal = data.get("rgb_calibration") + if isinstance(rgb_cal, dict): + self.rgb_calibration = self._merge_fusion_config(self.rgb_calibration, rgb_cal) def _merge_fusion_config(self, default_cfg: dict, loaded_cfg: dict) -> dict: cfg = json.loads(json.dumps(default_cfg)) diff --git a/Python/OAK/datasets/oak-fcc-3/core/raw_processor_preview.py b/Python/OAK/datasets/oak-fcc-3/core/raw_processor_preview.py index c9f251703..bb845d3d2 100644 --- a/Python/OAK/datasets/oak-fcc-3/core/raw_processor_preview.py +++ b/Python/OAK/datasets/oak-fcc-3/core/raw_processor_preview.py @@ -44,14 +44,15 @@ class RawProcessorPreview: def _debayer_code(self): mapping = { - # Mapeamento ajustado para OpenCV gerar BGR correto a partir do padrão Bayer informado. - "GBRG": cv2.COLOR_BayerGR2BGR, - "GRBG": cv2.COLOR_BayerGB2BGR, - "RGGB": cv2.COLOR_BayerBG2BGR, - "BGGR": cv2.COLOR_BayerRG2BGR, + "GBRG": cv2.COLOR_BayerGB2BGR, + "GRBG": cv2.COLOR_BayerGR2BGR, + "RGGB": cv2.COLOR_BayerRG2BGR, + "BGGR": cv2.COLOR_BayerBG2BGR, } + if self.bayer_pattern not in mapping: raise ValueError(f"Padrão Bayer não suportado: {self.bayer_pattern}") + return mapping[self.bayer_pattern] def apply_preview_white_balance(self, bgr: np.ndarray, strength: float = 1.0) -> np.ndarray: diff --git a/Python/OAK/datasets/oak-fcc-3/tests/test_oak_fcc3_client.py b/Python/OAK/datasets/oak-fcc-3/tests/test_oak_fcc3_client.py index c133ed9db..b31aaacab 100644 --- a/Python/OAK/datasets/oak-fcc-3/tests/test_oak_fcc3_client.py +++ b/Python/OAK/datasets/oak-fcc-3/tests/test_oak_fcc3_client.py @@ -10,16 +10,16 @@ with OakFcc3Client( frame_type="RAW_BRUTO", output_dtype="uint8", capture_mode="AUTO", - raw_policy="require_triple", + raw_policy="allow_single", sync_mode="best", sync_tolerance_ms=25.0, - #module_calibration_json="calibration/module_params.json", - #radiometric_enabled=False + module_calibration_json="calibration/module_params.json", + radiometric_enabled=False ) as cam: print("STATUS:", cam.get_status()) while True: - previews, meta = cam.get_next_preview(timeout=2.0) + previews, meta = cam.get_next_tensor_preview(timeout=2.0) print( "frame_id:", meta["frame_id"], @@ -28,14 +28,14 @@ with OakFcc3Client( "sync_ok:", meta.get("sync_ok"), ) - camera_info = meta.get("camera_info", {}) or {} + title_map = { + "rgb_tensor": "RGB (tensor final)", + "re_tensor": "RE (tensor final)", + "nir_tensor": "NIR (tensor final)", + } - for cam_id, preview in previews.items(): - info = camera_info.get(cam_id, {}) or {} - role = info.get("role", "unknown") - sensor = info.get("sensor", "") - - title = f"{cam_id} | {role.upper()} | {sensor}" + for name, preview in previews.items(): + title = title_map.get(name, name) cv2.imshow(title, preview) if cv2.waitKey(1) in (27, ord("q")): diff --git a/Python/OAK/datasets/oak-fcc-3/utils/build_module_params.py b/Python/OAK/datasets/oak-fcc-3/utils/build_module_params.py index 57be6ddab..3b452f44c 100644 --- a/Python/OAK/datasets/oak-fcc-3/utils/build_module_params.py +++ b/Python/OAK/datasets/oak-fcc-3/utils/build_module_params.py @@ -27,10 +27,31 @@ def main(): if args.radiometric_json: radiometric_data = load_json(args.radiometric_json) + # ========================= + # 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 (NOVO) + # ========================= + 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 = radiometric_data.get("radiometric_config") if not isinstance(radiometric_config, dict): @@ -55,10 +76,16 @@ def main(): "gain_apply_threshold": 0.02, } + # ========================= + # 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", "cam2"), + + # 🔥 AJUSTE IMPORTANTE + "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), @@ -66,8 +93,11 @@ def main(): "target_size": fusion_data.get("target_size", None), } + # ========================= + # MODULE PARAMS FINAL + # ========================= module_params = { - "schema": "multispec_module_params_v1", + "schema": "multispec_module_params_v2", # 👈 versão nova "saved_at": now_str(), "frame_type": cam_data.get("frame_type", fusion_data.get("frame_type", "RAW_BRUTO")), @@ -79,9 +109,11 @@ def main(): "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")), + # 🔥 BLOCOS PRINCIPAIS "camera_settings": camera_settings, "fusion_config": fusion_config, "radiometric_config": radiometric_config, + "rgb_calibration": rgb_calibration, # 👈 NOVO } with open(args.out, "w", encoding="utf-8") as f: diff --git a/Python/OAK/datasets/oak-fcc-3/utils/check_saved_files.py b/Python/OAK/datasets/oak-fcc-3/utils/check_saved_files.py index 1ddafc041..37bdc66ba 100644 --- a/Python/OAK/datasets/oak-fcc-3/utils/check_saved_files.py +++ b/Python/OAK/datasets/oak-fcc-3/utils/check_saved_files.py @@ -61,7 +61,7 @@ def build_visual_from_saved_payload(payload_path: Path, meta: dict, cam_id: str # Busca metadados da câmera no stream_meta stream_meta = meta.get("stream_meta", {}) or {} - cam_frames = stream_meta.get("camera_frames", {}) or {} + cam_frames = stream_meta.get("camera_info", {}) or {} cam_meta = cam_frames.get(cam_id, {}) or {} role = cam_meta.get("role", cam_id) @@ -133,7 +133,7 @@ def build_visual_from_saved_payload(payload_path: Path, meta: dict, cam_id: str raise RuntimeError(f"Payload MULTISPEC inválido, shape={arr.shape}") rgb_hwc = chw_to_hwc(arr[:3].astype(np.float32)) - preview_bgr = normalize_float01_to_bgr(rgb_hwc) + rgb_bgr = normalize_float01_to_bgr(rgb_hwc) re01 = arr[3].astype(np.float32) nir01 = arr[4].astype(np.float32) @@ -148,13 +148,14 @@ def build_visual_from_saved_payload(payload_path: Path, meta: dict, cam_id: str cv2.COLOR_GRAY2BGR ) - combined = np.vstack([ - np.hstack([preview_bgr, re_bgr]), - np.hstack([nir_bgr, np.zeros_like(preview_bgr)]) - ]) + panels = [ + ("RGB reconstruido", rgb_bgr, "canais 0,1,2"), + ("RE reconstruido", re_bgr, "canal 3"), + ("NIR reconstruido", nir_bgr, "canal 4"), + ] desc = f"Reconstruido de MULTISPEC | dtype={arr.dtype} | shape={arr.shape} | canais=[R,G,B,RE,NIR]" - return combined, desc + return panels, desc if saved_type == "raw_native_single": if arr.ndim == 3 and arr.shape[2] == 3 and arr.dtype == np.uint8: @@ -242,8 +243,13 @@ def build_panels_from_group(group): panels.append(("Preview salvo", preview_saved, f"{preview_saved.shape[1]}x{preview_saved.shape[0]}")) if group["final_raw"] is not None: - img, desc = build_visual_from_saved_payload(group["final_raw"], meta) - panels.append(("Reconstruido (final)", img, desc)) + result, desc = build_visual_from_saved_payload(group["final_raw"], meta) + + if isinstance(result, list): + for title, img, subtitle in result: + panels.append((title, img, subtitle)) + else: + panels.append(("Reconstruido (final)", result, desc)) for cam_id, path in group["cameras"].items(): img, desc = build_visual_from_saved_payload(path, meta, cam_id=cam_id) @@ -308,7 +314,18 @@ def compose_panels(panels, max_width=1600): def sort_panels(panels): - order = ["Preview salvo", "cam2", "cam0", "cam1"] + order = [ + "preview salvo", + "rgb reconstruido", + "re reconstruido", + "nir reconstruido", + "rgb", + "re", + "nir", + "cam_a", + "cam_b", + "cam_c", + ] def key(p): title = p[0].lower() diff --git a/Python/OAK/datasets/oak-fcc-3/utils/manual_fusion_calibrator.py b/Python/OAK/datasets/oak-fcc-3/utils/manual_fusion_calibrator.py index fe9efd9d0..cc0c2902b 100644 --- a/Python/OAK/datasets/oak-fcc-3/utils/manual_fusion_calibrator.py +++ b/Python/OAK/datasets/oak-fcc-3/utils/manual_fusion_calibrator.py @@ -10,10 +10,6 @@ import numpy as np from core.oak_fcc3_client import OakFcc3Client as MultiSpectralClient -# ============================================================ -# Helpers gerais -# ============================================================ - def now_str() -> str: return datetime.now().strftime("%Y-%m-%d %H:%M:%S") @@ -22,14 +18,7 @@ def ensure_dir(path: str): os.makedirs(path, exist_ok=True) -def overlay_hud( - img_bgr: np.ndarray, - lines: list[str], - x: int = 12, - y: int = 22, - font_scale: float = 0.6, - line_step: int = 24, -): +def overlay_hud(img_bgr, lines, x=12, y=22, font_scale=0.6, line_step=24): yy = y for s in lines: cv2.putText(img_bgr, s, (x, yy), cv2.FONT_HERSHEY_SIMPLEX, font_scale, (0, 0, 0), 3, cv2.LINE_AA) @@ -37,30 +26,22 @@ def overlay_hud( yy += line_step -def normalize_gray01(img: np.ndarray) -> np.ndarray: - arr = img.astype(np.float32) - mn = float(arr.min()) - mx = float(arr.max()) - if mx <= mn + 1e-9: - return np.zeros_like(arr, dtype=np.float32) - return (arr - mn) / (mx - mn) - - -def to_bgr_u8_from_rgb01(rgb01: np.ndarray) -> np.ndarray: +def to_bgr_u8_from_rgb01(rgb01): rgb_u8 = np.clip(rgb01 * 255.0, 0, 255).astype(np.uint8) return cv2.cvtColor(rgb_u8, cv2.COLOR_RGB2BGR) -def gray_to_color_bgr(gray01: np.ndarray, color_name: str) -> np.ndarray: +def gray_to_color_bgr(gray01, color_name): + if gray01 is None: + raise ValueError("gray01 não pode ser None") + g = np.clip(gray01 * 255.0, 0, 255).astype(np.uint8) z = np.zeros_like(g, dtype=np.uint8) color_name = color_name.upper() if color_name == "RE": - # vermelho artificial rgb = np.stack([g, z, z], axis=2) elif color_name == "NIR": - # ciano artificial rgb = np.stack([z, g, g], axis=2) else: rgb = np.stack([g, g, g], axis=2) @@ -68,33 +49,25 @@ def gray_to_color_bgr(gray01: np.ndarray, color_name: str) -> np.ndarray: return cv2.cvtColor(rgb, cv2.COLOR_RGB2BGR) -def apply_affine(img: np.ndarray, dx: int, dy: int, theta_deg: float = 0.0) -> np.ndarray: +def apply_affine(img, dx, dy, theta_deg=0.0): h, w = img.shape[:2] center = (w * 0.5, h * 0.5) + M = cv2.getRotationMatrix2D(center, theta_deg, 1.0) M[0, 2] += dx M[1, 2] += dy - if img.ndim == 2: - return cv2.warpAffine( - img, - M, - (w, h), - flags=cv2.INTER_LINEAR, - borderMode=cv2.BORDER_CONSTANT, - borderValue=0, - ) return cv2.warpAffine( img, M, (w, h), flags=cv2.INTER_LINEAR, borderMode=cv2.BORDER_CONSTANT, - borderValue=(0, 0, 0), + borderValue=0 if img.ndim == 2 else (0, 0, 0), ) -def apply_homography(img: np.ndarray, H) -> np.ndarray: +def apply_homography(img, H): if H is None: return img @@ -112,17 +85,18 @@ def apply_homography(img: np.ndarray, H) -> np.ndarray: def build_overlay_fuse( - rgb01: np.ndarray, - spec01: np.ndarray | None, - spec_name: str, - dx: int, - dy: int, - theta_deg: float = 0.0, - alpha: float = 0.45, - calibration_mode: str = "manual_affine", + rgb01, + spec01, + spec_name, + dx, + dy, + theta_deg=0.0, + alpha=0.45, + calibration_mode="manual_affine", H=None, ): base_bgr = to_bgr_u8_from_rgb01(rgb01) + if spec01 is None: return base_bgr @@ -132,47 +106,48 @@ def build_overlay_fuse( warped = apply_affine(spec01, dx, dy, theta_deg) spec_bgr = gray_to_color_bgr(warped, spec_name) - fused = cv2.addWeighted(base_bgr, 1.0 - alpha, spec_bgr, alpha, 0.0) - return fused + return cv2.addWeighted(base_bgr, 1.0 - alpha, spec_bgr, alpha, 0.0) -def resize_if_needed(img: np.ndarray, target_hw: tuple[int, int]) -> np.ndarray: +def resize_if_needed(img, target_hw): + if img is None: + return None + target_h, target_w = target_hw + if img.shape[:2] == (target_h, target_w): return img - interp = cv2.INTER_LINEAR - return cv2.resize(img, (target_w, target_h), interpolation=interp) + + return cv2.resize(img, (target_w, target_h), interpolation=cv2.INTER_LINEAR) -def stack_2x2(a: np.ndarray, b: np.ndarray, c: np.ndarray, d: np.ndarray) -> np.ndarray: - h = max(a.shape[0], b.shape[0], c.shape[0], d.shape[0]) - w = max(a.shape[1], b.shape[1], c.shape[1], d.shape[1]) - - def fit(img): - if img.shape[:2] != (h, w): - return cv2.resize(img, (w, h), interpolation=cv2.INTER_NEAREST) - return img - - a = fit(a) - b = fit(b) - c = fit(c) - d = fit(d) - top = np.hstack([a, b]) - bottom = np.hstack([c, d]) - return np.vstack([top, bottom]) - - -def build_empty_panel_like(ref_bgr: np.ndarray, title: str) -> np.ndarray: +def build_empty_panel_like(ref_bgr, title): img = np.zeros_like(ref_bgr) overlay_hud(img, [title, "sem frame disponivel"], x=18, y=40, font_scale=0.8, line_step=34) return img -def validate_module_ready(status: dict, frame_type: str, raw_policy: str, capture_mode: str): +def get_decoded_by_role(decoded, role): + role = str(role).lower() + + for cam_id, item in decoded.items(): + if str(item.get("role", "")).lower() == role: + return cam_id, item + + return None, None + + +def get_image_by_role(decoded, role): + cam_id, item = get_decoded_by_role(decoded, role) + if item is None: + return cam_id, None + return cam_id, item.get("image") + + +def validate_module_ready(status, frame_type, raw_policy): if not status.get("ok", True): raise RuntimeError(f"Status inválido retornado pelo módulo: {status}") - active_ids = list(status.get("active_camera_ids", [])) active_roles = status.get("active_roles", {}) or {} active_count = int(status.get("camera_count_active", 0)) @@ -181,24 +156,19 @@ def validate_module_ready(status: dict, frame_type: str, raw_policy: str, captur missing = [role for role in ("rgb", "nir", "re") if role not in active_roles] if missing: raise RuntimeError( - f"RAW_BRUTO com política require_triple exige três câmeras ativas. " - f"Faltando: {missing}. Ativas atuais: {active_ids}" + f"RAW_BRUTO com require_triple exige rgb/nir/re ativas. " + f"Faltando: {missing}. Ativas: {active_roles}" ) - else: - if active_count < 1: - raise RuntimeError("RAW_BRUTO requer ao menos uma câmera ativa, mas nenhuma foi detectada.") + elif active_count < 1: + raise RuntimeError("RAW_BRUTO requer ao menos uma câmera ativa.") return raise RuntimeError(f"frame_type desconhecido para validação: {frame_type}") -# ============================================================ -# Persistência dos offsets -# ============================================================ - -def default_offsets_payload(args, effective_capture_mode: str): +def default_offsets_payload(args, effective_capture_mode): return { - "schema": "manual_multispec_offsets_v1", + "schema": "manual_multispec_offsets_v2", "saved_at": now_str(), "frame_type": "RAW_BRUTO", "capture_mode_requested": args.capture_mode, @@ -207,9 +177,9 @@ def default_offsets_payload(args, effective_capture_mode: str): "sensor_width": args.width, "sensor_height": args.height, "bayer_pattern": args.bayer, + "reference_camera": "rgb", "baseline_mm": args.baseline_mm, "alignment_mode": "manual_affine", - "reference_camera": "rgb", "manual_offsets": { "re": {"dx": 0, "dy": 0, "theta_deg": 0.0}, "nir": {"dx": 0, "dy": 0, "theta_deg": 0.0}, @@ -222,106 +192,61 @@ def default_offsets_payload(args, effective_capture_mode: str): } -def load_offsets_json(path: str, args, effective_capture_mode: str): +def load_offsets_json(path, args, effective_capture_mode): if not path or not os.path.isfile(path): return default_offsets_payload(args, effective_capture_mode) with open(path, "r", encoding="utf-8") as f: data = json.load(f) - data.setdefault("schema", "manual_multispec_offsets_v1") + data.setdefault("schema", "manual_multispec_offsets_v2") + data.setdefault("reference_camera", "rgb") data.setdefault("baseline_mm", args.baseline_mm) data.setdefault("alignment_mode", "manual_affine") - data.setdefault("reference_camera", "rgb") + data.setdefault("manual_offsets", {}) + data.setdefault("homographies", {}) + data["manual_offsets"].setdefault("re", {"dx": 0, "dy": 0, "theta_deg": 0.0}) data["manual_offsets"].setdefault("nir", {"dx": 0, "dy": 0, "theta_deg": 0.0}) + data["homographies"].setdefault("re_to_rgb", None) data["homographies"].setdefault("nir_to_rgb", None) + return data -def save_offsets_json(path: str, data: dict): +def save_offsets_json(path, data): ensure_dir(os.path.dirname(path) or ".") data = dict(data) data["saved_at"] = now_str() + with open(path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) -def get_decoded_by_role(decoded, role): - role = str(role).lower() - for cam_id, item in decoded.items(): - if str(item.get("role", "")).lower() == role: - return cam_id, item - return None, None - - -# ============================================================ -# Main UI -# ============================================================ - def main(): parser = argparse.ArgumentParser( description="Calibrador manual de offsets para fusão RGB/RE/NIR a partir do stream RAW_BRUTO.", formatter_class=argparse.ArgumentDefaultsHelpFormatter, ) + parser.add_argument("--fps", type=int, default=20) parser.add_argument("--width", type=int, default=640) parser.add_argument("--height", type=int, default=480) - parser.add_argument("--bayer", default="GBRG", choices=["GBRG", "GRBG", "RGGB", "BGGR"]) + parser.add_argument("--bayer", default="RGGB", choices=["GBRG", "GRBG", "RGGB", "BGGR"]) parser.add_argument("--capture_mode", default="AUTO", choices=["AUTO", "SINGLE", "DOUBLE", "TRIPLE"]) parser.add_argument("--raw_policy", default="allow_single", choices=["allow_single", "require_triple"]) parser.add_argument("--baseline_mm", type=float, default=75.0) parser.add_argument("--preview_scale", type=float, default=1.0) - parser.add_argument("--step", type=int, default=1, help="Passo inicial em pixels ao usar as setas.") - parser.add_argument("--alpha", type=float, default=0.45, help="Alpha do overlay sobre RGB.") - parser.add_argument("--angle_step", type=float, default=0.10, help="Passo angular em graus para rotação manual.") + parser.add_argument("--step", type=int, default=1) + parser.add_argument("--alpha", type=float, default=0.45) + parser.add_argument("--angle_step", type=float, default=0.10) parser.add_argument("--out_json", default="calibration/manual_offsets.json") - parser.add_argument("--load_json", default="", help="Se informado, carrega offsets iniciais deste arquivo.") + parser.add_argument("--load_json", default="") parser.add_argument("--notes", default="") + args = parser.parse_args() - def on_mouse(event, x, y, flags, param): - nonlocal last_msg, last_msg_t - - if event != cv2.EVENT_LBUTTONDOWN: - return - - if calibration_mode != "homography": - return - - if selected_cam not in ("cam0", "cam1"): - return - - rgb_rect = panel_rects.get("rgb") - spec_rect = panel_rects.get(selected_cam) - - def inside(rect, px, py): - if rect is None: - return False - x0, y0, x1, y1 = rect - return x0 <= px < x1 and y0 <= py < y1 - - def to_local(rect, px, py): - x0, y0, x1, y1 = rect - return float(px - x0), float(py - y0) - - if inside(spec_rect, x, y): - pt = to_local(spec_rect, x, y) - #if len(selected_points_spec[selected_cam]) < 4: - selected_points_spec[selected_cam].append(pt) - last_msg = f"{selected_cam}: ponto SPEC #{len(selected_points_spec[selected_cam])}" - last_msg_t = time.time() - return - - if inside(rgb_rect, x, y): - pt = to_local(rgb_rect, x, y) - #if len(selected_points_rgb[selected_cam]) < 4: - selected_points_rgb[selected_cam].append(pt) - last_msg = f"{selected_cam}: ponto RGB #{len(selected_points_rgb[selected_cam])}" - last_msg_t = time.time() - return - effective_capture_mode = args.capture_mode offsets_data = load_offsets_json(args.load_json, args, effective_capture_mode) @@ -330,23 +255,21 @@ def main(): selected_role = "re" calibration_mode = offsets_data.get("alignment_mode", "manual_affine") - selected_points_spec = { - "re": [], - "nir": [], - } - selected_points_rgb = { - "re": [], - "nir": [], - } + selected_points_spec = {"re": [], "nir": []} + selected_points_rgb = {"re": [], "nir": []} + panel_rects = { "fuse": None, "rgb": None, "re": None, "nir": None, } + + decoded_last = {} last_msg = "" last_msg_t = 0.0 last_frame_id = -1 + fps_view = 0.0 fps_stream = 0.0 t_view_fps = time.time() @@ -355,8 +278,57 @@ def main(): stream_frames_accum = 0 last_stream_frame_id = None - decoded_last = {} window_name = "Manual Fusion Calibrator" + + def inside(rect, px, py): + if rect is None: + return False + x0, y0, x1, y1 = rect + return x0 <= px < x1 and y0 <= py < y1 + + def to_local(rect, px, py): + x0, y0, _, _ = rect + return float(px - x0), float(py - y0) + + def on_mouse(event, x, y, flags, param): + nonlocal last_msg, last_msg_t, selected_role, calibration_mode + + if event != cv2.EVENT_LBUTTONDOWN: + return + + if calibration_mode != "homography": + return + + if selected_role not in ("re", "nir"): + return + + rgb_rect = panel_rects.get("rgb") + spec_rect = panel_rects.get(selected_role) + + if inside(spec_rect, x, y): + pt = to_local(spec_rect, x, y) + selected_points_spec[selected_role].append(pt) + last_msg = f"{selected_role.upper()}: ponto SPEC #{len(selected_points_spec[selected_role])}" + last_msg_t = time.time() + return + + if inside(rgb_rect, x, y): + pt = to_local(rgb_rect, x, y) + selected_points_rgb[selected_role].append(pt) + last_msg = f"{selected_role.upper()}: ponto RGB #{len(selected_points_rgb[selected_role])}" + last_msg_t = time.time() + return + + def get_preview_panel_by_role(previews, meta, role): + camera_info = meta.get("camera_info", {}) or {} + + for cam_id, preview in previews.items(): + info = camera_info.get(cam_id, {}) or {} + if str(info.get("role", "")).lower() == role: + return preview + + return None + cv2.namedWindow(window_name, cv2.WINDOW_NORMAL) cv2.setMouseCallback(window_name, on_mouse) @@ -366,17 +338,22 @@ def main(): height=args.height, bayer=args.bayer, fps=args.fps, - frame_type="RAW_BRUTO", + frame_type="PREVIEW", output_dtype="uint8", capture_mode=effective_capture_mode, raw_policy=args.raw_policy, - module_calibration_json=None, - radiometric_enabled=True + module_calibration_json="calibration/module_params.json", + radiometric_enabled=False, ) as cam: + + validate_module_ready(cam.get_status(), "RAW_BRUTO", args.raw_policy) + previews_last = {} + while True: t0 = time.time() frame, meta, decoded = cam.get_next_decoded(timeout=2.0) + previews_last = cam.build_visual_preview_from_raw(frame, meta) if meta is not None and frame is not None and meta.get("frame_id") != last_frame_id: last_frame_id = meta["frame_id"] @@ -405,40 +382,46 @@ def main(): t_view_fps = time.time() if decoded_last: - rgb_id, rgb_item = get_decoded_by_role(decoded_last, "rgb") - re_id, re_item = get_decoded_by_role(decoded_last, "re") - nir_id, nir_item = get_decoded_by_role(decoded_last, "nir") - - rgb01 = rgb_item.get("image") if rgb_item else None - re01 = re_item.get("image") if re_item else None - nir01 = nir_item.get("image") if nir_item else None + rgb_id, rgb01 = get_image_by_role(decoded_last, "rgb") + re_id, re01 = get_image_by_role(decoded_last, "re") + nir_id, nir01 = get_image_by_role(decoded_last, "nir") if rgb01 is None: - # fallback para exibição quando não houver RGB if re01 is not None: rgb01 = np.stack([re01, re01, re01], axis=2) + rgb_id = "fallback_re" elif nir01 is not None: rgb01 = np.stack([nir01, nir01, nir01], axis=2) + rgb_id = "fallback_nir" else: rgb01 = np.zeros((args.height, args.width, 3), dtype=np.float32) + rgb_id = "empty" base_h, base_w = rgb01.shape[:2] - if re01 is not None: - re01 = resize_if_needed(re01, (base_h, base_w)) - if nir01 is not None: - nir01 = resize_if_needed(nir01, (base_h, base_w)) + re01 = resize_if_needed(re01, (base_h, base_w)) + nir01 = resize_if_needed(nir01, (base_h, base_w)) - rgb_panel = to_bgr_u8_from_rgb01(rgb01) - re_panel = gray_to_color_bgr(re01, "RE") if re01 is not None else build_empty_panel_like(rgb_panel, "RE") - nir_panel = gray_to_color_bgr(nir01, "NIR") if nir01 is not None else build_empty_panel_like(rgb_panel, "NIR") + rgb_panel = get_preview_panel_by_role(previews_last, meta, "rgb") + re_panel = get_preview_panel_by_role(previews_last, meta, "re") + nir_panel = get_preview_panel_by_role(previews_last, meta, "nir") + + if rgb_panel is None: + rgb_panel = to_bgr_u8_from_rgb01(rgb01) + + if re_panel is None: + re_panel = gray_to_color_bgr(re01, "RE") if re01 is not None else build_empty_panel_like(rgb_panel, "RE") + + if nir_panel is None: + nir_panel = gray_to_color_bgr(nir01, "NIR") if nir01 is not None else build_empty_panel_like(rgb_panel, "NIR") + + active_spec_name = selected_role.upper() + active_spec = re01 if selected_role == "re" else nir01 - active_spec_name = "RE" if selected_role == "cam0" else "NIR" - active_spec = re01 if selected_role == "cam0" else nir01 dx = int(offsets.get(selected_role, {}).get("dx", 0)) dy = int(offsets.get(selected_role, {}).get("dy", 0)) theta_deg = float(offsets.get(selected_role, {}).get("theta_deg", 0.0)) - H_key = f"{selected_role}_to_cam2" + H_key = f"{selected_role}_to_rgb" H = offsets_data.get("homographies", {}).get(H_key) fuse_panel = build_overlay_fuse( @@ -453,28 +436,37 @@ def main(): H=H, ) - spec_pts = len(selected_points_spec[selected_cam]) - rgb_pts = len(selected_points_rgb[selected_cam]) + spec_pts = len(selected_points_spec[selected_role]) + rgb_pts = len(selected_points_rgb[selected_role]) lines_fuse = [ f"FUSE: RGB + {active_spec_name}", - f"mode={calibration_mode} | selecionada={selected_cam}", + f"mode={calibration_mode} | selecionada={selected_role.upper()}", f"dx={dx} | dy={dy} | theta={theta_deg:.2f}g | step={args.step} | ang_step={args.angle_step:.2f}g", - f"pts_spec={spec_pts} | pts_rgb={rgb_pts} | min=4 | fps_stream={fps_stream:.1f} | fps_view={fps_view:.1f}" + f"pts_spec={spec_pts} | pts_rgb={rgb_pts} | min=4 | fps_stream={fps_stream:.1f} | fps_view={fps_view:.1f}", ] overlay_hud(fuse_panel, lines_fuse) - lines_rgb = ["RGB (cam2)"] - overlay_hud(rgb_panel, lines_rgb) + overlay_hud(rgb_panel, [f"RGB ({rgb_id})"], y=24) - re_dx = int(offsets.get("cam0", {}).get("dx", 0)) - re_dy = int(offsets.get("cam0", {}).get("dy", 0)) - re_theta = float(offsets.get("cam0", {}).get("theta_deg", 0.0)) - nir_dx = int(offsets.get("cam1", {}).get("dx", 0)) - nir_dy = int(offsets.get("cam1", {}).get("dy", 0)) - nir_theta = float(offsets.get("cam1", {}).get("theta_deg", 0.0)) - overlay_hud(re_panel, [f"RE (cam0) | dx={re_dx} dy={re_dy} th={re_theta:.2f}g", "2 seleciona RE"], y=24) - overlay_hud(nir_panel, [f"NIR (cam1) | dx={nir_dx} dy={nir_dy} th={nir_theta:.2f}g", "3 seleciona NIR"], y=24) + re_dx = int(offsets.get("re", {}).get("dx", 0)) + re_dy = int(offsets.get("re", {}).get("dy", 0)) + re_theta = float(offsets.get("re", {}).get("theta_deg", 0.0)) + + nir_dx = int(offsets.get("nir", {}).get("dx", 0)) + nir_dy = int(offsets.get("nir", {}).get("dy", 0)) + nir_theta = float(offsets.get("nir", {}).get("theta_deg", 0.0)) + + overlay_hud( + re_panel, + [f"RE ({re_id}) | dx={re_dx} dy={re_dy} th={re_theta:.2f}g", "2 seleciona RE"], + y=24, + ) + overlay_hud( + nir_panel, + [f"NIR ({nir_id}) | dx={nir_dx} dy={nir_dy} th={nir_theta:.2f}g", "3 seleciona NIR"], + y=24, + ) ph = max(fuse_panel.shape[0], rgb_panel.shape[0], re_panel.shape[0], nir_panel.shape[0]) pw = max(fuse_panel.shape[1], rgb_panel.shape[1], re_panel.shape[1], nir_panel.shape[1]) @@ -491,21 +483,52 @@ def main(): panel_rects["fuse"] = (0, 0, pw, ph) panel_rects["rgb"] = (pw, 0, pw * 2, ph) - panel_rects["cam0"] = (0, ph, pw, ph * 2) - panel_rects["cam1"] = (pw, ph, pw * 2, ph * 2) + panel_rects["re"] = (0, ph, pw, ph * 2) + panel_rects["nir"] = (pw, ph, pw * 2, ph * 2) top = np.hstack([fuse_panel, rgb_panel]) bottom = np.hstack([re_panel, nir_panel]) board = np.vstack([top, bottom]) help_lines = [ - "M=manual_affine | H=homography | clique pares correspondentes | >=4 pares | SPACE=salva | C=limpa pts | Z=zera sel | X=zera tudo", - "A/W/S/D movem | J/L rotacionam | O/P muda passo angular | I/U remove ultimo ponto | ENTER calcula H | TAB alterna camera | Q/Esc sai", + "M=manual_affine | H=homography | clique pares | >=4 pares | SPACE=salva | C=limpa pts | Z=zera sel | X=zera tudo", + "A/W/S/D movem | J/L rotacionam | O/P ang_step | I/U remove ponto | ENTER calcula H | TAB alterna RE/NIR | Q/Esc sai", ] overlay_hud(board, help_lines, x=16, y=board.shape[0] - 44, font_scale=0.55, line_step=20) if last_msg and (time.time() - last_msg_t) < 2.5: - cv2.putText(board, last_msg, (16, board.shape[0] - 72), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2, cv2.LINE_AA) + cv2.putText( + board, + last_msg, + (16, board.shape[0] - 72), + cv2.FONT_HERSHEY_SIMPLEX, + 0.7, + (0, 255, 0), + 2, + cv2.LINE_AA, + ) + + if calibration_mode == "homography": + color_spec = (0, 255, 255) + color_rgb = (0, 255, 0) + + for idx, pt in enumerate(selected_points_spec[selected_role]): + rect = panel_rects[selected_role] + if rect is not None: + x0, y0, _, _ = rect + px = int(x0 + pt[0]) + py = int(y0 + pt[1]) + cv2.circle(board, (px, py), 5, color_spec, -1) + cv2.putText(board, str(idx + 1), (px + 6, py - 6), cv2.FONT_HERSHEY_SIMPLEX, 0.5, color_spec, 1, cv2.LINE_AA) + + for idx, pt in enumerate(selected_points_rgb[selected_role]): + rect = panel_rects["rgb"] + if rect is not None: + x0, y0, _, _ = rect + px = int(x0 + pt[0]) + py = int(y0 + pt[1]) + cv2.circle(board, (px, py), 5, color_rgb, -1) + cv2.putText(board, str(idx + 1), (px + 6, py - 6), cv2.FONT_HERSHEY_SIMPLEX, 0.5, color_rgb, 1, cv2.LINE_AA) if args.preview_scale != 1.0: board = cv2.resize( @@ -514,57 +537,39 @@ def main(): interpolation=cv2.INTER_NEAREST, ) - if calibration_mode == "homography": - color_spec = (0, 255, 255) - color_rgb = (0, 255, 0) - - for idx, pt in enumerate(selected_points_spec[selected_cam]): - rect = panel_rects[selected_cam] - if rect is not None: - x0, y0, _, _ = rect - px = int(x0 + pt[0]) - py = int(y0 + pt[1]) - cv2.circle(board, (px, py), 5, color_spec, -1) - cv2.putText(board, str(idx + 1), (px + 6, py - 6), - cv2.FONT_HERSHEY_SIMPLEX, 0.5, color_spec, 1, cv2.LINE_AA) - - for idx, pt in enumerate(selected_points_rgb[selected_cam]): - rect = panel_rects["rgb"] - if rect is not None: - x0, y0, _, _ = rect - px = int(x0 + pt[0]) - py = int(y0 + pt[1]) - cv2.circle(board, (px, py), 5, color_rgb, -1) - cv2.putText(board, str(idx + 1), (px + 6, py - 6), - cv2.FONT_HERSHEY_SIMPLEX, 0.5, color_rgb, 1, cv2.LINE_AA) - cv2.imshow(window_name, board) + else: blank = np.zeros((720, 1280, 3), dtype=np.uint8) overlay_hud(blank, ["Aguardando frames do módulo..."], x=40, y=80, font_scale=1.0, line_step=34) cv2.imshow(window_name, blank) k = cv2.waitKey(1) & 0xFF + if k in (ord("q"), ord("Q"), 27): break + elif k in (ord("m"), ord("M")): calibration_mode = "manual_affine" offsets_data["alignment_mode"] = calibration_mode last_msg = "Modo: manual_affine" last_msg_t = time.time() + elif k in (ord("h"), ord("H")): calibration_mode = "homography" offsets_data["alignment_mode"] = calibration_mode last_msg = "Modo: homography" last_msg_t = time.time() + elif k in (ord("c"), ord("C")): - selected_points_spec[selected_cam] = [] - selected_points_rgb[selected_cam] = [] - last_msg = f"Pontos limpos: {selected_cam}" + selected_points_spec[selected_role] = [] + selected_points_rgb[selected_role] = [] + last_msg = f"Pontos limpos: {selected_role.upper()}" last_msg_t = time.time() - elif k == 13: # ENTER - spec_pts = selected_points_spec[selected_cam] - rgb_pts = selected_points_rgb[selected_cam] + + elif k == 13: + spec_pts = selected_points_spec[selected_role] + rgb_pts = selected_points_rgb[selected_role] if len(spec_pts) >= 4 and len(rgb_pts) >= 4 and len(spec_pts) == len(rgb_pts): src = np.array(spec_pts, dtype=np.float32) @@ -573,109 +578,131 @@ def main(): H, status = cv2.findHomography(src, dst, method=cv2.RANSAC) if H is not None: offsets_data.setdefault("homographies", {}) - offsets_data["homographies"][f"{selected_cam}_to_cam2"] = H.tolist() + offsets_data["homographies"][f"{selected_role}_to_rgb"] = H.tolist() inliers = int(status.sum()) if status is not None else len(spec_pts) - last_msg = f"H calculada para {selected_cam} | pts={len(spec_pts)} | inliers={inliers}" + last_msg = f"H calculada para {selected_role.upper()} | pts={len(spec_pts)} | inliers={inliers}" else: - last_msg = f"Falha ao calcular H para {selected_cam}" + last_msg = f"Falha ao calcular H para {selected_role.upper()}" else: - last_msg = f"{selected_cam}: precisa de >=4 pares e mesmo numero de pontos" + last_msg = f"{selected_role.upper()}: precisa de >=4 pares e mesmo numero de pontos" last_msg_t = time.time() + elif k == ord("2"): - if "cam0" in decoded_last: - selected_cam = "cam0" - last_msg = "Selecionada: cam0 / RE" + _, item = get_decoded_by_role(decoded_last, "re") + if item is not None: + selected_role = "re" + last_msg = "Selecionada: RE" else: - last_msg = "cam0 / RE nao disponivel neste frame" + last_msg = "RE nao disponivel neste frame" last_msg_t = time.time() + elif k == ord("3"): - if "cam1" in decoded_last: - selected_cam = "cam1" - last_msg = "Selecionada: cam1 / NIR" + _, item = get_decoded_by_role(decoded_last, "nir") + if item is not None: + selected_role = "nir" + last_msg = "Selecionada: NIR" else: - last_msg = "cam1 / NIR nao disponivel neste frame" + last_msg = "NIR nao disponivel neste frame" last_msg_t = time.time() - elif k == 9: # TAB - choices = [cid for cid in ("cam0", "cam1") if cid in decoded_last] + + elif k == 9: + choices = [role for role in ("re", "nir") if get_decoded_by_role(decoded_last, role)[1] is not None] if len(choices) >= 2: - selected_cam = choices[1] if selected_cam == choices[0] else choices[0] - last_msg = f"Selecionada: {selected_cam}" + selected_role = choices[1] if selected_role == choices[0] else choices[0] + last_msg = f"Selecionada: {selected_role.upper()}" last_msg_t = time.time() + elif k in (ord("z"), ord("Z")): - offsets.setdefault(selected_cam, {"dx": 0, "dy": 0, "theta_deg": 0.0}) - offsets[selected_cam]["dx"] = 0 - offsets[selected_cam]["dy"] = 0 - offsets[selected_cam]["theta_deg"] = 0.0 - last_msg = f"Offset zerado: {selected_cam}" + offsets.setdefault(selected_role, {"dx": 0, "dy": 0, "theta_deg": 0.0}) + offsets[selected_role] = {"dx": 0, "dy": 0, "theta_deg": 0.0} + last_msg = f"Offset zerado: {selected_role.upper()}" last_msg_t = time.time() + elif k in (ord("x"), ord("X")): - offsets["cam0"] = {"dx": 0, "dy": 0, "theta_deg": 0.0} - offsets["cam1"] = {"dx": 0, "dy": 0, "theta_deg": 0.0} + offsets["re"] = {"dx": 0, "dy": 0, "theta_deg": 0.0} + offsets["nir"] = {"dx": 0, "dy": 0, "theta_deg": 0.0} last_msg = "Todos offsets zerados" last_msg_t = time.time() + elif k == 32: - # Se uma câmera não apareceu, salva zerada como pedido - if "cam0" not in decoded_last: - offsets["cam0"] = {"dx": 0, "dy": 0, "theta_deg": 0.0} - if "cam1" not in decoded_last: - offsets["cam1"] = {"dx": 0, "dy": 0, "theta_deg": 0.0} + if get_decoded_by_role(decoded_last, "re")[1] is None: + offsets["re"] = {"dx": 0, "dy": 0, "theta_deg": 0.0} + + if get_decoded_by_role(decoded_last, "nir")[1] is None: + offsets["nir"] = {"dx": 0, "dy": 0, "theta_deg": 0.0} offsets_data["manual_offsets"] = offsets + offsets_data.setdefault("homographies", {}) + offsets_data["schema"] = "manual_multispec_offsets_v2" + offsets_data["reference_camera"] = "rgb" + save_offsets_json(args.out_json, offsets_data) last_msg = f"Offsets salvos em: {args.out_json}" last_msg_t = time.time() + elif k in (ord("+"), ord("=")): args.step = min(args.step + 1, 50) last_msg = f"Step -> {args.step}px" last_msg_t = time.time() + elif k in (ord("-"), ord("_")): args.step = max(args.step - 1, 1) last_msg = f"Step -> {args.step}px" last_msg_t = time.time() + elif k in (ord("a"), ord("A")): - if selected_cam in decoded_last: - offsets.setdefault(selected_cam, {"dx": 0, "dy": 0, "theta_deg": 0.0}) - offsets[selected_cam]["dx"] -= args.step + if get_decoded_by_role(decoded_last, selected_role)[1] is not None: + offsets.setdefault(selected_role, {"dx": 0, "dy": 0, "theta_deg": 0.0}) + offsets[selected_role]["dx"] -= args.step + elif k in (ord("d"), ord("D")): - if selected_cam in decoded_last: - offsets.setdefault(selected_cam, {"dx": 0, "dy": 0, "theta_deg": 0.0}) - offsets[selected_cam]["dx"] += args.step + if get_decoded_by_role(decoded_last, selected_role)[1] is not None: + offsets.setdefault(selected_role, {"dx": 0, "dy": 0, "theta_deg": 0.0}) + offsets[selected_role]["dx"] += args.step + elif k in (ord("w"), ord("W")): - if selected_cam in decoded_last: - offsets.setdefault(selected_cam, {"dx": 0, "dy": 0, "theta_deg": 0.0}) - offsets[selected_cam]["dy"] -= args.step + if get_decoded_by_role(decoded_last, selected_role)[1] is not None: + offsets.setdefault(selected_role, {"dx": 0, "dy": 0, "theta_deg": 0.0}) + offsets[selected_role]["dy"] -= args.step + elif k in (ord("s"), ord("S")): - if selected_cam in decoded_last: - offsets.setdefault(selected_cam, {"dx": 0, "dy": 0, "theta_deg": 0.0}) - offsets[selected_cam]["dy"] += args.step + if get_decoded_by_role(decoded_last, selected_role)[1] is not None: + offsets.setdefault(selected_role, {"dx": 0, "dy": 0, "theta_deg": 0.0}) + offsets[selected_role]["dy"] += args.step + elif k in (ord("j"), ord("J")): - if selected_cam in decoded_last: - offsets.setdefault(selected_cam, {"dx": 0, "dy": 0, "theta_deg": 0.0}) - offsets[selected_cam]["theta_deg"] -= args.angle_step + if get_decoded_by_role(decoded_last, selected_role)[1] is not None: + offsets.setdefault(selected_role, {"dx": 0, "dy": 0, "theta_deg": 0.0}) + offsets[selected_role]["theta_deg"] -= args.angle_step + elif k in (ord("l"), ord("L")): - if selected_cam in decoded_last: - offsets.setdefault(selected_cam, {"dx": 0, "dy": 0, "theta_deg": 0.0}) - offsets[selected_cam]["theta_deg"] += args.angle_step + if get_decoded_by_role(decoded_last, selected_role)[1] is not None: + offsets.setdefault(selected_role, {"dx": 0, "dy": 0, "theta_deg": 0.0}) + offsets[selected_role]["theta_deg"] += args.angle_step + elif k in (ord("o"), ord("O")): args.angle_step = max(args.angle_step - 0.05, 0.01) last_msg = f"Angle step -> {args.angle_step:.2f}°" last_msg_t = time.time() + elif k in (ord("p"), ord("P")): args.angle_step = min(args.angle_step + 0.05, 5.0) last_msg = f"Angle step -> {args.angle_step:.2f}°" last_msg_t = time.time() + elif k in (ord("u"), ord("U")): - if selected_points_spec[selected_cam]: - selected_points_spec[selected_cam].pop() - last_msg = f"Removido ultimo ponto SPEC de {selected_cam}" + if selected_points_spec[selected_role]: + selected_points_spec[selected_role].pop() + last_msg = f"Removido ultimo ponto SPEC de {selected_role.upper()}" last_msg_t = time.time() + elif k in (ord("i"), ord("I")): - if selected_points_rgb[selected_cam]: - selected_points_rgb[selected_cam].pop() - last_msg = f"Removido ultimo ponto RGB de {selected_cam}" + if selected_points_rgb[selected_role]: + selected_points_rgb[selected_role].pop() + last_msg = f"Removido ultimo ponto RGB de {selected_role.upper()}" last_msg_t = time.time() - + dt_loop = time.time() - t0 if dt_loop < 0.001: time.sleep(0.001) @@ -686,4 +713,4 @@ def main(): if __name__ == "__main__": - main() + main() \ No newline at end of file diff --git a/Python/OAK/datasets/oak-fcc-3/utils/sensor_calibration_tool.py b/Python/OAK/datasets/oak-fcc-3/utils/sensor_calibration_tool.py index 47ec11075..cb28b619c 100644 --- a/Python/OAK/datasets/oak-fcc-3/utils/sensor_calibration_tool.py +++ b/Python/OAK/datasets/oak-fcc-3/utils/sensor_calibration_tool.py @@ -9,18 +9,36 @@ import numpy as np from core.oak_fcc3_client import OakFcc3Client as MultiSpectralClient + # ============================================================ -# Helpers +# Helpers gerais # ============================================================ def now_str() -> str: return datetime.now().strftime("%Y-%m-%d %H:%M:%S") +def ts_name() -> str: + return datetime.now().strftime("%Y%m%d_%H%M%S_%f")[:-3] + + def ensure_dir(path: str): os.makedirs(path, exist_ok=True) +def role_label(role: str) -> str: + return str(role).upper() + + +def channel_title(role: str) -> str: + mapping = { + "rgb": "RGB", + "re": "RE", + "nir": "NIR", + } + return mapping.get(str(role).lower(), str(role).upper()) + + def overlay_hud( img_bgr, lines, @@ -84,26 +102,29 @@ def gray_to_bgr_u8(gray01: np.ndarray) -> np.ndarray: def resize_if_needed(img: np.ndarray, target_hw: tuple[int, int]) -> np.ndarray: + if img is None: + return None + target_h, target_w = target_hw if img.shape[:2] == (target_h, target_w): return img return cv2.resize(img, (target_w, target_h), interpolation=cv2.INTER_LINEAR) -def validate_module_ready(status: dict, frame_type: str, raw_policy: str, capture_mode: str): +def validate_module_ready(status: dict, frame_type: str, raw_policy: str): if not status.get("ok", True): raise RuntimeError(f"Status inválido retornado pelo módulo: {status}") - active_ids = list(status.get("active_camera_ids", [])) + active_roles = status.get("active_roles", {}) or {} active_count = int(status.get("camera_count_active", 0)) if frame_type == "RAW_BRUTO": if raw_policy == "require_triple": - missing = [cid for cid in ("cam0", "cam1", "cam2") if cid not in active_ids] + missing = [role for role in ("rgb", "nir", "re") if role not in active_roles] if missing: raise RuntimeError( - f"RAW_BRUTO com política require_triple exige três câmeras ativas. " - f"Faltando: {missing}. Ativas atuais: {active_ids}" + "RAW_BRUTO com política require_triple exige rgb/nir/re ativos. " + f"Faltando: {missing}. Ativos atuais: {active_roles}" ) else: if active_count < 1: @@ -134,6 +155,25 @@ def color_for_index(idx: int) -> tuple[int, int, int]: return palette[idx % len(palette)] +def get_decoded_by_role(decoded: dict, role: str): + role = str(role).lower() + for cam_id, item in decoded.items(): + if str(item.get("role", "")).lower() == role: + return cam_id, item + return None, None + + +def get_image_by_role(decoded: dict, role: str): + cam_id, item = get_decoded_by_role(decoded, role) + if item is None: + return cam_id, None + return cam_id, item.get("image") + + +# ============================================================ +# Estatísticas e ROIs +# ============================================================ + def compute_stats_from_roi(img01: np.ndarray, rect: tuple[int, int, int, int]) -> dict: x0, y0, x1, y1 = rect x0, x1 = sorted((int(x0), int(x1))) @@ -170,6 +210,17 @@ def compute_stats_from_roi(img01: np.ndarray, rect: tuple[int, int, int, int]) - def compute_scene_health(img01: np.ndarray) -> dict: + if img01 is None: + return { + "mean": 0.0, + "std": 0.0, + "p05": 0.0, + "p95": 0.0, + "pct_saturated": 0.0, + "pct_dark": 0.0, + "comment": "sem imagem", + } + arr = img01.astype(np.float32).reshape(-1) mean = float(arr.mean()) std = float(arr.std()) @@ -254,9 +305,13 @@ def compute_stats_from_polygon_roi(img01: np.ndarray, points: list) -> dict: if arr.size == 0: return { - "valid": False, "mean": 0.0, "std": 0.0, - "min": 0.0, "max": 0.0, - "p05": 0.0, "p95": 0.0, + "valid": False, + "mean": 0.0, + "std": 0.0, + "min": 0.0, + "max": 0.0, + "p05": 0.0, + "p95": 0.0, "pct_saturated": 0.0, "pct_dark": 0.0, "pixels": 0, @@ -297,21 +352,19 @@ def draw_current_polygon(panel_bgr: np.ndarray, points: list): # ============================================================ -# MOCK +# Mock / offline # ============================================================ def load_mock_image_rgb(path: str, fallback_shape=(480, 640)): if not path: h, w = fallback_shape - img = np.zeros((h, w, 3), dtype=np.float32) - return img + return np.zeros((h, w, 3), dtype=np.float32) bgr = cv2.imread(path, cv2.IMREAD_COLOR) if bgr is None: raise RuntimeError(f"Falha ao carregar mock RGB: {path}") - rgb = bgr[:, :, ::-1].astype(np.float32) / 255.0 - return rgb + return bgr[:, :, ::-1].astype(np.float32) / 255.0 def load_mock_image_gray(path: str, fallback_shape=(480, 640)): @@ -329,25 +382,17 @@ def load_mock_image_gray(path: str, fallback_shape=(480, 640)): def build_mock_decoded(args): shape = (args.height, args.width) - cam2 = load_mock_image_rgb(args.mock_cam2, fallback_shape=shape) - cam0 = load_mock_image_gray(args.mock_cam0, fallback_shape=shape) - cam1 = load_mock_image_gray(args.mock_cam1, fallback_shape=shape) + rgb = load_mock_image_rgb(args.mock_rgb, fallback_shape=shape) + re = load_mock_image_gray(args.mock_re, fallback_shape=shape) + nir = load_mock_image_gray(args.mock_nir, fallback_shape=shape) return { - "cam2": {"name": "RGB", "image": cam2, "meta": {"mock": True}}, - "cam0": {"name": "RE", "image": cam0, "meta": {"mock": True}}, - "cam1": {"name": "NIR", "image": cam1, "meta": {"mock": True}}, + "mock_rgb": {"name": "RGB", "role": "rgb", "image": rgb, "meta": {"mock": True, "role": "rgb"}}, + "mock_re": {"name": "RE", "role": "re", "image": re, "meta": {"mock": True, "role": "re"}}, + "mock_nir": {"name": "NIR", "role": "nir", "image": nir, "meta": {"mock": True, "role": "nir"}}, } -# ============================================================ -# Análise de dados offline -# ============================================================ - -def ts_name() -> str: - return datetime.now().strftime("%Y%m%d_%H%M%S_%f")[:-3] - - def save_offline_sample( base_dir: str, preview_bgr: np.ndarray, @@ -419,8 +464,6 @@ def load_offline_sample_decoded(json_path: str, cam: MultiSpectralClient): frame[cam_id] = arr stream_meta = meta.get("stream_meta") or meta - - # Garante campos mínimos usados pelo decoder. stream_meta.setdefault("camera_frames", meta.get("camera_frames", {})) stream_meta.setdefault("frame_type", "RAW_BRUTO") @@ -437,12 +480,49 @@ def load_offline_sample_decoded(json_path: str, cam: MultiSpectralClient): # ============================================================ -# Persistência dos parâmetros/snapshots +# Persistência # ============================================================ +def default_rgb_calibration(): + return { + "enabled": True, + "gains": { + "R": 1.0, + "G": 1.0, + "B": 1.0, + }, + } + + +def default_camera_controls(): + return { + "rgb": { + "ae_enable": True, + "awb_enable": True, + "exposure_time_us": 15000, + "analogue_gain": 1.0, + "colour_gains": [1.0, 1.0], + }, + "re": { + "ae_enable": False, + "awb_enable": False, + "exposure_time_us": 15000, + "analogue_gain": 1.0, + "colour_gains": None, + }, + "nir": { + "ae_enable": False, + "awb_enable": False, + "exposure_time_us": 15000, + "analogue_gain": 1.0, + "colour_gains": None, + }, + } + + def default_payload(args, effective_capture_mode: str): return { - "schema": "manual_sensor_calibration_v1", + "schema": "manual_sensor_calibration_v2", "saved_at": now_str(), "frame_type": "RAW_BRUTO", "capture_mode_requested": args.capture_mode, @@ -452,16 +532,41 @@ def default_payload(args, effective_capture_mode: str): "sensor_height": args.height, "bayer_pattern": args.bayer, "notes": args.notes or "", - "camera_settings": { - "cam0": {}, - "cam1": {}, - "cam2": {}, - }, + "camera_settings": default_camera_controls(), + "rgb_calibration": default_rgb_calibration(), + "rois": {"rgb": [], "re": [], "nir": []}, "snapshots": [], "calibration_guidance_log": [], } +def _migrate_old_role_key(key: str) -> str: + key = str(key).lower() + old_map = { + "cam2": "rgb", + "cam0": "re", + "cam1": "nir", + } + return old_map.get(key, key) + + +def _migrate_role_dict(d: dict, default: dict) -> dict: + out = json.loads(json.dumps(default)) + + if isinstance(d, dict): + for key, value in d.items(): + role = _migrate_old_role_key(key) + if role in out: + if isinstance(out[role], dict) and isinstance(value, dict): + out[role].update(value) + elif isinstance(out[role], list) and isinstance(value, list): + out[role] = value + else: + out[role] = value + + return out + + def load_payload(path: str, args, effective_capture_mode: str): if not path or not os.path.isfile(path): return default_payload(args, effective_capture_mode) @@ -469,11 +574,23 @@ def load_payload(path: str, args, effective_capture_mode: str): with open(path, "r", encoding="utf-8") as f: data = json.load(f) - data.setdefault("schema", "manual_sensor_calibration_v1") - data.setdefault("camera_settings", {"cam0": {}, "cam1": {}, "cam2": {}}) - data.setdefault("snapshots", []) - data.setdefault("calibration_guidance_log", []) - return data + base = default_payload(args, effective_capture_mode) + base.update(data) + base["schema"] = "manual_sensor_calibration_v2" + base["camera_settings"] = _migrate_role_dict(data.get("camera_settings", {}), default_camera_controls()) + base["rois"] = _migrate_role_dict(data.get("rois", {}), {"rgb": [], "re": [], "nir": []}) + + rgb_cal = default_rgb_calibration() + if isinstance(data.get("rgb_calibration"), dict): + rgb_cal.update(data["rgb_calibration"]) + gains = default_rgb_calibration()["gains"] + gains.update(data["rgb_calibration"].get("gains", {}) or {}) + rgb_cal["gains"] = gains + base["rgb_calibration"] = rgb_cal + + base.setdefault("snapshots", []) + base.setdefault("calibration_guidance_log", []) + return base def save_payload(path: str, data: dict): @@ -484,13 +601,10 @@ def save_payload(path: str, data: dict): json.dump(data, f, ensure_ascii=False, indent=2) -def build_camera_params_payload(args, effective_capture_mode, camera_controls, rois=None, snapshots=None, guidance_log=None): +def build_camera_params_payload(args, effective_capture_mode, camera_controls, rgb_calibration, rois=None, snapshots=None, guidance_log=None): return { - "schema": "multispec_camera_params_v1", + "schema": "multispec_camera_params_v2", "saved_at": now_str(), - "pi_host": args.pi_host, - "pc_host": args.pc_host, - "stream_port": args.stream_port, "frame_type": "RAW_BRUTO", "capture_mode_requested": args.capture_mode, "capture_mode_effective": effective_capture_mode, @@ -499,7 +613,8 @@ def build_camera_params_payload(args, effective_capture_mode, camera_controls, r "sensor_height": args.height, "bayer_pattern": args.bayer, "camera_settings": json.loads(json.dumps(camera_controls)), - "rois": rois or {}, + "rgb_calibration": json.loads(json.dumps(rgb_calibration)), + "rois": rois or {"rgb": [], "re": [], "nir": []}, "snapshots": snapshots or [], "notes": args.notes or "", "calibration_guidance_log": guidance_log or [], @@ -560,7 +675,7 @@ def mean_of_classes(summary: dict, classes: list[str], key: str = "mean"): return float(np.mean(vals)) -def analyze_spectral_guidance(selected_cam: str, img01: np.ndarray, rois_for_cam: list[dict], ctrl: dict, exp_step: int, gain_step: float) -> dict: +def analyze_spectral_guidance(role: str, img01: np.ndarray, rois_for_cam: list[dict], ctrl: dict, exp_step: int, gain_step: float) -> dict: summary = collect_roi_metrics_by_class(img01, rois_for_cam) veg_mean = mean_of_classes(summary, ["cana", "erva"], "mean") @@ -579,6 +694,7 @@ def analyze_spectral_guidance(selected_cam: str, img01: np.ndarray, rois_for_cam "status": "need_rois", "action": "none", "reason": "Crie pelo menos uma ROI de cana ou erva para analisar canal espectral.", + "channel": role, "class_metrics": summary, "before_settings": before, "after_settings": new_ctrl, @@ -588,60 +704,45 @@ def analyze_spectral_guidance(selected_cam: str, img01: np.ndarray, rois_for_cam if solo_mean is not None: separation = float(veg_mean - solo_mean) - exp = new_ctrl.get("exposure_time_us") - gain = new_ctrl.get("analogue_gain") - - if exp is None: - exp = 15000 - if gain is None: - gain = 1.0 + exp = new_ctrl.get("exposure_time_us") or 15000 + gain = new_ctrl.get("analogue_gain") or 1.0 new_ctrl["ae_enable"] = False new_ctrl["awb_enable"] = False - # 1) Proteção contra estouro - MIN_EXP_US = 100 - MIN_GAIN = 1.0 + min_exp_us = 100 + min_gain = 1.0 if veg_sat is not None and veg_sat > 1.0: - if exp > MIN_EXP_US: - new_ctrl["exposure_time_us"] = int(max(exp - exp_step, MIN_EXP_US)) + if exp > min_exp_us: + new_ctrl["exposure_time_us"] = int(max(exp - exp_step, min_exp_us)) action = "decrease_exposure" status = "adjust" reason = f"Vegetacao saturando ({veg_sat:.2f}%). Reduzir exposicao." - elif gain > MIN_GAIN: - new_ctrl["analogue_gain"] = float(max(gain / (1.0 + gain_step), MIN_GAIN)) + elif gain > min_gain: + new_ctrl["analogue_gain"] = float(max(gain / (1.0 + gain_step), min_gain)) action = "decrease_gain" status = "adjust" - reason = ( - f"Vegetacao saturando ({veg_sat:.2f}%), mas exposicao ja esta no minimo. " - "Reduzir ganho." - ) + reason = "Vegetacao saturando, mas exposicao ja esta no minimo. Reduzir ganho." else: action = "keep" status = "limit" - reason = ( - f"Vegetacao saturando ({veg_sat:.2f}%), mas exposicao e ganho ja estao no minimo. " - "Nao ha ajuste possivel por software." - ) + reason = "Vegetacao saturando, mas exposicao e ganho ja estao no minimo." - # 2) Vegetação pouco iluminada elif veg_p95 is not None and veg_p95 < 0.75: new_ctrl["exposure_time_us"] = int(min(exp + exp_step, 200000)) action = "increase_exposure" status = "adjust" reason = f"p95 da vegetacao baixo ({veg_p95:.3f}). Aumentar exposicao." - # 3) Vegetação muito perto do teto elif veg_p95 is not None and veg_p95 > 0.96: new_ctrl["exposure_time_us"] = int(max(exp - exp_step, 100)) action = "decrease_exposure" status = "adjust" reason = f"p95 da vegetacao alto ({veg_p95:.3f}). Reduzir exposicao." - # 4) Separação ruim elif separation is not None and separation < 0.25: if veg_p95 is not None and veg_p95 < 0.90: new_ctrl["exposure_time_us"] = int(min(exp + exp_step, 200000)) @@ -658,7 +759,7 @@ def analyze_spectral_guidance(selected_cam: str, img01: np.ndarray, rois_for_cam "status": status, "action": action, "reason": reason, - "channel": selected_cam, + "channel": role, "class_metrics": summary, "veg_mean": veg_mean, "solo_mean": solo_mean, @@ -670,7 +771,7 @@ def analyze_spectral_guidance(selected_cam: str, img01: np.ndarray, rois_for_cam } -def analyze_rgb_guidance(selected_cam: str, img01: np.ndarray, rois_for_cam: list[dict], ctrl: dict, exp_step: int, gain_step: float) -> dict: +def analyze_rgb_guidance(role: str, img01: np.ndarray, rois_for_cam: list[dict], ctrl: dict, exp_step: int, gain_step: float, rgb_calibration: dict) -> dict: scene = compute_scene_health(img01) summary = collect_roi_metrics_by_class(img01, rois_for_cam) @@ -680,15 +781,9 @@ def analyze_rgb_guidance(selected_cam: str, img01: np.ndarray, rois_for_cam: lis status = "ok" reason = "RGB parece aceitável." - exp = new_ctrl.get("exposure_time_us") - gain = new_ctrl.get("analogue_gain") + exp = new_ctrl.get("exposure_time_us") or 15000 + gain = new_ctrl.get("analogue_gain") or 1.0 - if exp is None: - exp = 15000 - if gain is None: - gain = 1.0 - - # Para RGB calibrado fixo: desligar AE/AWB quando for aplicar preset final. new_ctrl["ae_enable"] = False new_ctrl["awb_enable"] = False @@ -714,15 +809,16 @@ def analyze_rgb_guidance(selected_cam: str, img01: np.ndarray, rois_for_cam: lis "status": status, "action": action, "reason": reason, - "channel": selected_cam, + "channel": role, "scene_health": scene, "class_metrics": summary, + "rgb_calibration": json.loads(json.dumps(rgb_calibration)), "before_settings": before, "after_settings": new_ctrl, } -def run_guidance_analysis(selected_cam: str, img01: np.ndarray, rois_for_cam: list[dict], ctrl: dict, exp_step: int, gain_step: float) -> dict: +def run_guidance_analysis(role: str, img01: np.ndarray, rois_for_cam: list[dict], ctrl: dict, exp_step: int, gain_step: float, rgb_calibration: dict) -> dict: if img01 is None: return { "status": "error", @@ -732,10 +828,37 @@ def run_guidance_analysis(selected_cam: str, img01: np.ndarray, rois_for_cam: li "after_settings": json.loads(json.dumps(ctrl)), } - if selected_cam == "cam2": - return analyze_rgb_guidance(selected_cam, img01, rois_for_cam, ctrl, exp_step, gain_step) + if role == "rgb": + return analyze_rgb_guidance(role, img01, rois_for_cam, ctrl, exp_step, gain_step, rgb_calibration) - return analyze_spectral_guidance(selected_cam, img01, rois_for_cam, ctrl, exp_step, gain_step) + return analyze_spectral_guidance(role, img01, rois_for_cam, ctrl, exp_step, gain_step) + + +# ============================================================ +# RGB calibration gains helpers +# ============================================================ + +def clamp_rgb_gain(v: float) -> float: + return float(max(0.10, min(5.0, v))) + + +def apply_rgb_calibration_preview(rgb01: np.ndarray, rgb_calibration: dict) -> np.ndarray: + if rgb01 is None: + return None + + if not rgb_calibration.get("enabled", False): + return rgb01 + + gains = rgb_calibration.get("gains", {}) or {} + r_gain = float(gains.get("R", 1.0)) + g_gain = float(gains.get("G", 1.0)) + b_gain = float(gains.get("B", 1.0)) + + out = rgb01.astype(np.float32).copy() + out[:, :, 0] *= r_gain + out[:, :, 1] *= g_gain + out[:, :, 2] *= b_gain + return np.clip(out, 0.0, 1.0) # ============================================================ @@ -744,25 +867,26 @@ def run_guidance_analysis(selected_cam: str, img01: np.ndarray, rois_for_cam: li def main(): parser = argparse.ArgumentParser( - description="Ferramenta de calibração dos sensores RGB/RE/NIR com controle manual e ROIs em tempo real.", + description="Ferramenta de calibração dos sensores RGB/RE/NIR com controles por role e rgb_calibration.", formatter_class=argparse.ArgumentDefaultsHelpFormatter, ) parser.add_argument("--fps", type=int, default=20) parser.add_argument("--width", type=int, default=640) parser.add_argument("--height", type=int, default=480) - parser.add_argument("--bayer", default="GBRG", choices=["GBRG", "GRBG", "RGGB", "BGGR"]) + parser.add_argument("--bayer", default="RGGB", choices=["GBRG", "GRBG", "RGGB", "BGGR"]) parser.add_argument("--capture_mode", default="AUTO", choices=["AUTO", "SINGLE", "DOUBLE", "TRIPLE"]) parser.add_argument("--raw_policy", default="allow_single", choices=["allow_single", "require_triple"]) parser.add_argument("--preview_scale", type=float, default=1.0) parser.add_argument("--exp_step", type=int, default=1000, help="Passo de exposição em us") - parser.add_argument("--gain_step", type=float, default=0.10, help="Passo multiplicativo do ganho") + parser.add_argument("--gain_step", type=float, default=0.10, help="Passo multiplicativo do ganho de câmera") + parser.add_argument("--rgb_gain_step", type=float, default=0.05, help="Passo de ganho fixo por canal RGB do tensor") parser.add_argument("--out_json", default="calibration/sensor_calibration.json") parser.add_argument("--load_json", default="") parser.add_argument("--notes", default="") parser.add_argument("--mock", action="store_true") - parser.add_argument("--mock_cam0", default="", help="Imagem mock para cam0 / RE") - parser.add_argument("--mock_cam1", default="", help="Imagem mock para cam1 / NIR") - parser.add_argument("--mock_cam2", default="", help="Imagem mock para cam2 / RGB") + parser.add_argument("--mock_rgb", default="", help="Imagem mock para role rgb") + parser.add_argument("--mock_re", default="", help="Imagem mock para role re") + parser.add_argument("--mock_nir", default="", help="Imagem mock para role nir") parser.add_argument("--offline_sample_json", default="", help="JSON de sample salvo para análise offline") parser.add_argument("--offline_save_dir", default="calibration/offline_samples", help="Pasta para salvar frames brutos offline") args = parser.parse_args() @@ -777,17 +901,22 @@ def main(): capture_mode=args.capture_mode, raw_policy=args.raw_policy, module_calibration_json=None, - radiometric_enabled=True + radiometric_enabled=True, ) offline_mode = bool(args.offline_sample_json) live_mode = not args.mock and not offline_mode - effective_capture_mode = args.capture_mode data_payload = load_payload(args.load_json, args, effective_capture_mode) - selected_cam = "cam2" + selected_role = "rgb" + selected_rgb_channel = "R" + + camera_controls = data_payload.get("camera_settings", default_camera_controls()) + rgb_calibration = data_payload.get("rgb_calibration", default_rgb_calibration()) + rois = data_payload.get("rois", {"rgb": [], "re": [], "nir": []}) + last_msg = "" last_msg_t = 0.0 last_frame_id = -1 @@ -811,51 +940,20 @@ def main(): guidance_log = data_payload.get("calibration_guidance_log", []) last_guidance = guidance_log[-1]["result"] if guidance_log else None - window_name = "Sensor Calibration Tool" + window_name = "Sensor Calibration Tool - role-first" cv2.namedWindow(window_name, cv2.WINDOW_NORMAL) panel_rects = { - "cam2": None, - "cam0": None, - "cam1": None, + "rgb": None, + "re": None, + "nir": None, "data": None, } - # Controle de câmera - camera_controls = { - "cam0": { - "ae_enable": False, - "awb_enable": False, - "exposure_time_us": 15000, - "analogue_gain": 1.0, - "colour_gains": None, - }, - "cam1": { - "ae_enable": False, - "awb_enable": False, - "exposure_time_us": 15000, - "analogue_gain": 1.0, - "colour_gains": None, - }, - "cam2": { - "ae_enable": True, - "awb_enable": True, - "exposure_time_us": 15000, - "analogue_gain": 1.0, - "colour_gains": [1.0, 1.0], - }, - } - - rois = { - "cam2": [], - "cam0": [], - "cam1": [], - } - current_polygon_points = [] def get_active_rect_for_mouse(): - return panel_rects.get(selected_cam) + return panel_rects.get(selected_role) def on_mouse(event, x, y, flags, param): nonlocal current_polygon_points, last_msg, last_msg_t @@ -874,7 +972,7 @@ def main(): if event == cv2.EVENT_LBUTTONDOWN: current_polygon_points.append((lx, ly)) - last_msg = f"{selected_cam}: ponto #{len(current_polygon_points)} adicionado" + last_msg = f"{selected_role.upper()}: ponto #{len(current_polygon_points)} adicionado" last_msg_t = time.time() cv2.setMouseCallback(window_name, on_mouse) @@ -885,30 +983,30 @@ def main(): if offline_mode: decoded_last, last_meta_stream, last_raw_frame, last_preview_bgr = load_offline_sample_decoded(args.offline_sample_json, cam) - def apply_controls_to_selected_cam(): + def apply_controls_to_selected_role(): nonlocal cam, last_msg, last_msg_t - ctrl = camera_controls[selected_cam] + ctrl = camera_controls[selected_role] try: - resp = cam.svc.set_ae_enable(selected_cam, bool(ctrl["ae_enable"])) + resp = cam.svc.set_ae_enable(role=selected_role, enable=bool(ctrl["ae_enable"])) ctrl["ae_enable"] = bool(resp.get("ae_enable", ctrl["ae_enable"])) - if selected_cam == "cam2": - resp = cam.svc.set_awb_enable(selected_cam, bool(ctrl["awb_enable"])) + if selected_role == "rgb": + resp = cam.svc.set_awb_enable(role=selected_role, enable=bool(ctrl["awb_enable"])) ctrl["awb_enable"] = bool(resp.get("awb_enable", ctrl["awb_enable"])) if not ctrl["ae_enable"]: if ctrl["exposure_time_us"] is not None: - resp = cam.svc.set_exposure_time(selected_cam, int(ctrl["exposure_time_us"])) + resp = cam.svc.set_exposure_time(role=selected_role, exposure_time_us=int(ctrl["exposure_time_us"])) exp_val = resp.get("exposure_time_us", ctrl["exposure_time_us"]) ctrl["exposure_time_us"] = int(exp_val) if exp_val is not None else None if ctrl["analogue_gain"] is not None: - resp = cam.svc.set_analogue_gain(selected_cam, float(ctrl["analogue_gain"])) + resp = cam.svc.set_analogue_gain(role=selected_role, analogue_gain=float(ctrl["analogue_gain"])) gain_val = resp.get("analogue_gain", ctrl["analogue_gain"]) ctrl["analogue_gain"] = float(gain_val) if gain_val is not None else None - last_msg = f"Controles aplicados em {selected_cam}" + last_msg = f"Controles aplicados em {selected_role.upper()}" last_msg_t = time.time() except Exception as e: @@ -916,15 +1014,14 @@ def main(): last_msg_t = time.time() def snapshot_current_state(): - active_img = None - if selected_cam in decoded_last: - active_img = decoded_last[selected_cam]["image"] + _, item = get_decoded_by_role(decoded_last, selected_role) + active_img = item.get("image") if item else None if active_img is None: return None roi_entries = [] - for roi in rois[selected_cam]: + for roi in rois[selected_role]: stats = compute_stats_for_roi(active_img, roi) entry = { @@ -942,8 +1039,9 @@ def main(): snap = { "timestamp": now_str(), - "camera": selected_cam, - "camera_settings": json.loads(json.dumps(camera_controls[selected_cam])), + "role": selected_role, + "camera_settings": json.loads(json.dumps(camera_controls[selected_role])), + "rgb_calibration": json.loads(json.dumps(rgb_calibration)), "scene_health": compute_scene_health(active_img), "rois": roi_entries, } @@ -955,35 +1053,36 @@ def main(): if not live_mode: return - for cam_id in camera_controls.keys(): + for role in camera_controls.keys(): try: - initial_ctrl = cam.svc.get_camera_controls(cam_id) + initial_ctrl = cam.svc.get_camera_controls(role=role) - camera_controls[cam_id]["ae_enable"] = bool( - initial_ctrl.get("ae_enable", camera_controls[cam_id]["ae_enable"]) + camera_controls[role]["ae_enable"] = bool( + initial_ctrl.get("ae_enable", camera_controls[role]["ae_enable"]) ) - camera_controls[cam_id]["awb_enable"] = bool( - initial_ctrl.get("awb_enable", camera_controls[cam_id]["awb_enable"]) + camera_controls[role]["awb_enable"] = bool( + initial_ctrl.get("awb_enable", camera_controls[role]["awb_enable"]) ) - exp_val = initial_ctrl.get("exposure_time_us", camera_controls[cam_id]["exposure_time_us"]) - camera_controls[cam_id]["exposure_time_us"] = int(exp_val) if exp_val is not None else None + exp_val = initial_ctrl.get("exposure_time_us", camera_controls[role]["exposure_time_us"]) + camera_controls[role]["exposure_time_us"] = int(exp_val) if exp_val is not None else None - gain_val = initial_ctrl.get("analogue_gain", camera_controls[cam_id]["analogue_gain"]) - camera_controls[cam_id]["analogue_gain"] = float(gain_val) if gain_val is not None else None + gain_val = initial_ctrl.get("analogue_gain", camera_controls[role]["analogue_gain"]) + camera_controls[role]["analogue_gain"] = float(gain_val) if gain_val is not None else None - camera_controls[cam_id]["colour_gains"] = initial_ctrl.get( + camera_controls[role]["colour_gains"] = initial_ctrl.get( "colour_gains", - camera_controls[cam_id]["colour_gains"] + camera_controls[role]["colour_gains"] ) except Exception as e: - print(f"[WARN] Falha ao ler controles iniciais de {cam_id}: {e}") + print(f"[WARN] Falha ao ler controles iniciais de {role}: {e}") try: if live_mode: cam.start(print_debug=True) + validate_module_ready(cam.get_status(), "RAW_BRUTO", args.raw_policy) sync_camera_controls_from_pi() else: last_msg = "MODO OFFLINE ativo" if offline_mode else "MODO MOCK ativo" @@ -1027,31 +1126,33 @@ def main(): fps_view = 0.0 if decoded_last: - rgb01 = decoded_last.get("cam2", {}).get("image") - re01 = decoded_last.get("cam0", {}).get("image") - nir01 = decoded_last.get("cam1", {}).get("image") + rgb_id, rgb01_raw = get_image_by_role(decoded_last, "rgb") + re_id, re01 = get_image_by_role(decoded_last, "re") + nir_id, nir01 = get_image_by_role(decoded_last, "nir") - if rgb01 is None: + rgb01_display = apply_rgb_calibration_preview(rgb01_raw, rgb_calibration) if rgb01_raw is not None else None + + if rgb01_display is None: rgb_panel = build_empty_panel((args.height, args.width), "RGB") base_h, base_w = args.height, args.width else: - rgb_panel = to_bgr_u8_from_rgb01(rgb01) - base_h, base_w = rgb01.shape[:2] + rgb_panel = to_bgr_u8_from_rgb01(rgb01_display) + base_h, base_w = rgb01_display.shape[:2] re_panel = gray_to_bgr_u8(resize_if_needed(re01, (base_h, base_w))) if re01 is not None else build_empty_panel((base_h, base_w), "RE") nir_panel = gray_to_bgr_u8(resize_if_needed(nir01, (base_h, base_w))) if nir01 is not None else build_empty_panel((base_h, base_w), "NIR") - draw_rois(rgb_panel, rois["cam2"]) - draw_rois(re_panel, rois["cam0"]) - draw_rois(nir_panel, rois["cam1"]) + draw_rois(rgb_panel, rois["rgb"]) + draw_rois(re_panel, rois["re"]) + draw_rois(nir_panel, rois["nir"]) - active_panel = {"cam2": rgb_panel, "cam0": re_panel, "cam1": nir_panel}.get(selected_cam) + active_panel = {"rgb": rgb_panel, "re": re_panel, "nir": nir_panel}.get(selected_role) if active_panel is not None: draw_current_polygon(active_panel, current_polygon_points) - overlay_hud(rgb_panel, ["RGB (cam2)", f"ativo={selected_cam == 'cam2'}"]) - overlay_hud(re_panel, ["RE (cam0)", f"ativo={selected_cam == 'cam0'}"]) - overlay_hud(nir_panel, ["NIR (cam1)", f"ativo={selected_cam == 'cam1'}"]) + overlay_hud(rgb_panel, [f"RGB ({rgb_id})", f"ativo={selected_role == 'rgb'}"]) + overlay_hud(re_panel, [f"RE ({re_id})", f"ativo={selected_role == 're'}"]) + overlay_hud(nir_panel, [f"NIR ({nir_id})", f"ativo={selected_role == 'nir'}"]) ph = max(rgb_panel.shape[0], re_panel.shape[0], nir_panel.shape[0], base_h) pw = max(rgb_panel.shape[1], re_panel.shape[1], nir_panel.shape[1], base_w) @@ -1064,32 +1165,44 @@ def main(): rgb_panel = fit_panel(rgb_panel) re_panel = fit_panel(re_panel) nir_panel = fit_panel(nir_panel) - if rgb01 is not None: - last_preview_bgr = to_bgr_u8_from_rgb01(rgb01) + + if rgb01_display is not None: + last_preview_bgr = to_bgr_u8_from_rgb01(rgb01_display) else: last_preview_bgr = rgb_panel.copy() data_panel = np.zeros((ph, pw, 3), dtype=np.uint8) - panel_rects["cam2"] = (0, 0, pw, ph) - panel_rects["cam0"] = (pw, 0, pw * 2, ph) - panel_rects["cam1"] = (0, ph, pw, ph * 2) + panel_rects["rgb"] = (0, 0, pw, ph) + panel_rects["re"] = (pw, 0, pw * 2, ph) + panel_rects["nir"] = (0, ph, pw, ph * 2) panel_rects["data"] = (pw, ph, pw * 2, ph * 2) top = np.hstack([rgb_panel, re_panel]) bottom = np.hstack([nir_panel, data_panel]) board = np.vstack([top, bottom]) - active_img = decoded_last.get(selected_cam, {}).get("image") + _, active_item = get_decoded_by_role(decoded_last, selected_role) + active_img = active_item.get("image") if active_item else None global_stats = compute_scene_health(active_img) if active_img is not None else None - ctrl = camera_controls[selected_cam] + ctrl = camera_controls[selected_role] + + rgb_gains = rgb_calibration.get("gains", {}) or {} + rgb_gain_lines = [ + f"RGB_CAL enabled={'ON' if rgb_calibration.get('enabled', False) else 'OFF'} | canal={selected_rgb_channel}", + f"R={float(rgb_gains.get('R', 1.0)):.3f} | G={float(rgb_gains.get('G', 1.0)):.3f} | B={float(rgb_gains.get('B', 1.0)):.3f}", + ] + lines = [ - f"CAM ATIVA: {selected_cam}", + f"ROLE ATIVA: {selected_role.upper()}", f"AE={'ON' if ctrl['ae_enable'] else 'OFF'} | AWB={'ON' if ctrl['awb_enable'] else 'OFF'}", f"EXP={ctrl['exposure_time_us']} us", f"GAIN={ctrl['analogue_gain']:.2f}", f"fps_stream={fps_stream:.1f} | fps_view={fps_view:.1f}", ] + if selected_role == "rgb": + lines.extend(rgb_gain_lines) + if global_stats is not None: lines.extend([ f"mean={global_stats['mean']:.3f} | std={global_stats['std']:.3f}", @@ -1120,8 +1233,8 @@ def main(): ]) lines.append("-") - lines.append(f"ROIs: {len(rois[selected_cam])}") - for idx, roi in enumerate(rois[selected_cam][:6]): + lines.append(f"ROIs: {len(rois[selected_role])}") + for idx, roi in enumerate(rois[selected_role][:6]): if active_img is None: break stats = compute_stats_for_roi(active_img, roi) @@ -1132,6 +1245,7 @@ def main(): "-", "1=RGB | 2=RE | 3=NIR | E=AE | B=AWB", "I/K exp +/- | O/L gain +/- | G guia | A aplica", + "RGB: R/G/T seleciona ganho | Y/H ajusta ganho | V liga/desliga rgb_cal", "mouse: clique pontos | ENTER fecha ROI | U desfaz ponto/ROI | X limpa poligono", "F salva frame bruto | SPACE salva PARAMS | S snapshot | Q sai", ]) @@ -1164,36 +1278,36 @@ def main(): k = cv2.waitKey(1) & 0xFF if roi_name_input_active: - if k in (13, 10): # ENTER + if k in (13, 10): name = roi_name_buffer.strip() if not name: - name = f"roi_{len(rois[selected_cam]) + 1}" + name = f"roi_{len(rois[selected_role]) + 1}" roi = { "name": name, "type": "polygon", "points": list(roi_name_points_pending), - "color": color_for_index(len(rois[selected_cam])), + "color": color_for_index(len(rois[selected_role])), } - rois[selected_cam].append(roi) + rois[selected_role].append(roi) current_polygon_points = [] roi_name_points_pending = [] roi_name_buffer = "" roi_name_input_active = False - last_msg = f"ROI criada em {selected_cam}: {name}" + last_msg = f"ROI criada em {selected_role.upper()}: {name}" last_msg_t = time.time() - elif k in (27,): # ESC + elif k in (27,): roi_name_input_active = False roi_name_buffer = "" roi_name_points_pending = [] last_msg = "Criacao de ROI cancelada" last_msg_t = time.time() - elif k in (8, 127): # BACKSPACE + elif k in (8, 127): roi_name_buffer = roi_name_buffer[:-1] elif 32 <= k <= 126: @@ -1203,126 +1317,185 @@ def main(): if k in (ord("q"), ord("Q"), 27): break + elif k == ord("1"): - selected_cam = "cam2" - last_msg = "Selecionada: cam2 / RGB" + selected_role = "rgb" + last_msg = "Selecionada: RGB" last_msg_t = time.time() + elif k == ord("2"): - selected_cam = "cam0" - last_msg = "Selecionada: cam0 / RE" + selected_role = "re" + last_msg = "Selecionada: RE" last_msg_t = time.time() + elif k == ord("3"): - selected_cam = "cam1" - last_msg = "Selecionada: cam1 / NIR" + selected_role = "nir" + last_msg = "Selecionada: NIR" last_msg_t = time.time() + elif k in (ord("e"), ord("E")): - camera_controls[selected_cam]["ae_enable"] = not camera_controls[selected_cam]["ae_enable"] - last_msg = f"AE {selected_cam} -> {'ON' if camera_controls[selected_cam]['ae_enable'] else 'OFF'}" + camera_controls[selected_role]["ae_enable"] = not camera_controls[selected_role]["ae_enable"] + last_msg = f"AE {selected_role.upper()} -> {'ON' if camera_controls[selected_role]['ae_enable'] else 'OFF'}" last_msg_t = time.time() + elif k in (ord("b"), ord("B")): - if selected_cam == "cam2": - camera_controls[selected_cam]["awb_enable"] = not camera_controls[selected_cam]["awb_enable"] - last_msg = f"AWB {selected_cam} -> {'ON' if camera_controls[selected_cam]['awb_enable'] else 'OFF'}" + if selected_role == "rgb": + camera_controls[selected_role]["awb_enable"] = not camera_controls[selected_role]["awb_enable"] + last_msg = f"AWB RGB -> {'ON' if camera_controls[selected_role]['awb_enable'] else 'OFF'}" else: last_msg = "AWB só se aplica ao RGB" last_msg_t = time.time() + elif k in (ord("i"), ord("I")): - if camera_controls[selected_cam]["exposure_time_us"] is None: - camera_controls[selected_cam]["exposure_time_us"] = 15000 + if camera_controls[selected_role]["exposure_time_us"] is None: + camera_controls[selected_role]["exposure_time_us"] = 15000 else: - camera_controls[selected_cam]["exposure_time_us"] = int( - min(camera_controls[selected_cam]["exposure_time_us"] + args.exp_step, 200000) + camera_controls[selected_role]["exposure_time_us"] = int( + min(camera_controls[selected_role]["exposure_time_us"] + args.exp_step, 200000) ) - last_msg = f"EXP {selected_cam} -> {camera_controls[selected_cam]['exposure_time_us']} us" + last_msg = f"EXP {selected_role.upper()} -> {camera_controls[selected_role]['exposure_time_us']} us" last_msg_t = time.time() + elif k in (ord("k"), ord("K")): - if camera_controls[selected_cam]["exposure_time_us"] is None: - camera_controls[selected_cam]["exposure_time_us"] = 15000 + if camera_controls[selected_role]["exposure_time_us"] is None: + camera_controls[selected_role]["exposure_time_us"] = 15000 else: - camera_controls[selected_cam]["exposure_time_us"] = int( - max(camera_controls[selected_cam]["exposure_time_us"] - args.exp_step, 100) + camera_controls[selected_role]["exposure_time_us"] = int( + max(camera_controls[selected_role]["exposure_time_us"] - args.exp_step, 100) ) - last_msg = f"EXP {selected_cam} -> {camera_controls[selected_cam]['exposure_time_us']} us" + last_msg = f"EXP {selected_role.upper()} -> {camera_controls[selected_role]['exposure_time_us']} us" last_msg_t = time.time() + elif k in (ord("o"), ord("O")): - if camera_controls[selected_cam]["analogue_gain"] is None: - camera_controls[selected_cam]["analogue_gain"] = 1.0 + if camera_controls[selected_role]["analogue_gain"] is None: + camera_controls[selected_role]["analogue_gain"] = 1.0 else: - camera_controls[selected_cam]["analogue_gain"] = float( - min(camera_controls[selected_cam]["analogue_gain"] * (1.0 + args.gain_step), 32.0) + camera_controls[selected_role]["analogue_gain"] = float( + min(camera_controls[selected_role]["analogue_gain"] * (1.0 + args.gain_step), 32.0) ) - last_msg = f"GAIN {selected_cam} -> {camera_controls[selected_cam]['analogue_gain']:.2f}" + last_msg = f"GAIN {selected_role.upper()} -> {camera_controls[selected_role]['analogue_gain']:.2f}" last_msg_t = time.time() + elif k in (ord("l"), ord("L")): - if camera_controls[selected_cam]["analogue_gain"] is None: - camera_controls[selected_cam]["analogue_gain"] = 1.0 + if camera_controls[selected_role]["analogue_gain"] is None: + camera_controls[selected_role]["analogue_gain"] = 1.0 else: - camera_controls[selected_cam]["analogue_gain"] = float( - max(camera_controls[selected_cam]["analogue_gain"] / (1.0 + args.gain_step), 1.0) + camera_controls[selected_role]["analogue_gain"] = float( + max(camera_controls[selected_role]["analogue_gain"] / (1.0 + args.gain_step), 1.0) ) - last_msg = f"GAIN {selected_cam} -> {camera_controls[selected_cam]['analogue_gain']:.2f}" + last_msg = f"GAIN {selected_role.upper()} -> {camera_controls[selected_role]['analogue_gain']:.2f}" last_msg_t = time.time() + + elif k in (ord("v"), ord("V")): + if selected_role == "rgb": + rgb_calibration["enabled"] = not rgb_calibration.get("enabled", False) + last_msg = f"RGB_CAL -> {'ON' if rgb_calibration['enabled'] else 'OFF'}" + else: + last_msg = "rgb_calibration só se aplica ao RGB" + last_msg_t = time.time() + + elif k in (ord("r"), ord("R")): + if selected_role == "rgb": + selected_rgb_channel = "R" + last_msg = "RGB gain channel -> R" + last_msg_t = time.time() + elif k in (ord("g"), ord("G")): - active_img = decoded_last.get(selected_cam, {}).get("image") - ctrl = camera_controls[selected_cam] + if selected_role == "rgb": + selected_rgb_channel = "G" + last_msg = "RGB gain channel -> G" + last_msg_t = time.time() + else: + active_img = get_image_by_role(decoded_last, selected_role)[1] + ctrl = camera_controls[selected_role] - result = run_guidance_analysis( - selected_cam=selected_cam, - img01=active_img, - rois_for_cam=rois[selected_cam], - ctrl=ctrl, - exp_step=args.exp_step, - gain_step=args.gain_step, - ) + result = run_guidance_analysis( + role=selected_role, + img01=active_img, + rois_for_cam=rois[selected_role], + ctrl=ctrl, + exp_step=args.exp_step, + gain_step=args.gain_step, + rgb_calibration=rgb_calibration, + ) - last_guidance = result + last_guidance = result + guidance_entry = { + "timestamp": now_str(), + "role": selected_role, + "result": result, + } - guidance_entry = { - "timestamp": now_str(), - "camera": selected_cam, - "result": result, - } + guidance_log.append(guidance_entry) + data_payload.setdefault("calibration_guidance_log", []).append(guidance_entry) - guidance_log.append(guidance_entry) - data_payload.setdefault("calibration_guidance_log", []).append(guidance_entry) + after = result.get("after_settings") + if isinstance(after, dict): + camera_controls[selected_role].update(after) - after = result.get("after_settings") - if isinstance(after, dict): - camera_controls[selected_cam].update(after) + last_msg = f"GUIDE {selected_role.upper()}: {result.get('action')} | {result.get('status')}" + last_msg_t = time.time() - last_msg = f"GUIDE {selected_cam}: {result.get('action')} | {result.get('status')}" + elif k in (ord("t"), ord("T")): + if selected_role == "rgb": + selected_rgb_channel = "B" + last_msg = "RGB gain channel -> B" + last_msg_t = time.time() + + elif k in (ord("y"), ord("Y")): + if selected_role == "rgb": + gains = rgb_calibration.setdefault("gains", {"R": 1.0, "G": 1.0, "B": 1.0}) + gains[selected_rgb_channel] = clamp_rgb_gain(float(gains.get(selected_rgb_channel, 1.0)) + args.rgb_gain_step) + last_msg = f"RGB_CAL {selected_rgb_channel} -> {gains[selected_rgb_channel]:.3f}" + else: + last_msg = "Ganho RGB só se aplica ao RGB" last_msg_t = time.time() + + elif k in (ord("h"), ord("H")): + if selected_role == "rgb": + gains = rgb_calibration.setdefault("gains", {"R": 1.0, "G": 1.0, "B": 1.0}) + gains[selected_rgb_channel] = clamp_rgb_gain(float(gains.get(selected_rgb_channel, 1.0)) - args.rgb_gain_step) + last_msg = f"RGB_CAL {selected_rgb_channel} -> {gains[selected_rgb_channel]:.3f}" + else: + last_msg = "Ganho RGB só se aplica ao RGB" + last_msg_t = time.time() + elif k in (ord("a"), ord("A")): if live_mode: - apply_controls_to_selected_cam() + apply_controls_to_selected_role() else: last_msg = "Controles só aplicam no modo ao vivo" last_msg_t = time.time() + elif k in (ord("u"), ord("U")): if current_polygon_points: current_polygon_points.pop() last_msg = f"Ponto removido | restantes={len(current_polygon_points)}" last_msg_t = time.time() - elif rois[selected_cam]: - removed = rois[selected_cam].pop() + elif rois[selected_role]: + removed = rois[selected_role].pop() last_msg = f"ROI removida: {removed['name']}" last_msg_t = time.time() + elif k in (ord("x"), ord("X")): current_polygon_points = [] last_msg = "Polígono atual limpo" last_msg_t = time.time() + elif k in (ord("c"), ord("C")): - rois[selected_cam] = [] - last_msg = f"ROIs limpas em {selected_cam}" + rois[selected_role] = [] + last_msg = f"ROIs limpas em {selected_role.upper()}" last_msg_t = time.time() + elif k in (ord("s"), ord("S")): snap = snapshot_current_state() if snap is not None: data_payload.setdefault("snapshots", []).append(snap) - last_msg = f"Snapshot salvo: {selected_cam} | rois={len(snap['rois'])}" + last_msg = f"Snapshot salvo: {selected_role.upper()} | rois={len(snap['rois'])}" else: last_msg = "Sem frame ativo para snapshot" last_msg_t = time.time() + elif k in (ord("f"), ord("F")): if not live_mode: last_msg = "Salvar frame bruto só faz sentido no modo ao vivo" @@ -1347,6 +1520,7 @@ def main(): "raw_policy": args.raw_policy, "stream_meta": last_meta_stream, "camera_settings": json.loads(json.dumps(camera_controls)), + "rgb_calibration": json.loads(json.dumps(rgb_calibration)), "note": "offline_sample_from_sensor_calibration_tool", } @@ -1359,11 +1533,13 @@ def main(): last_msg = f"FRAME salvo offline: {os.path.basename(json_path)}" last_msg_t = time.time() - elif k == 32: # SPACE + + elif k == 32: payload_to_save = build_camera_params_payload( args=args, effective_capture_mode=effective_capture_mode, camera_controls=camera_controls, + rgb_calibration=rgb_calibration, rois=rois, snapshots=data_payload.get("snapshots", []), guidance_log=guidance_log, @@ -1373,7 +1549,8 @@ def main(): data_payload = payload_to_save last_msg = f"PARAMS salvos em: {args.out_json}" last_msg_t = time.time() - elif k == 13: # ENTER + + elif k == 13: if len(current_polygon_points) < 3: last_msg = "ROI poligonal precisa de pelo menos 3 pontos" last_msg_t = time.time()