import os import json import argparse from pathlib import Path from datetime import datetime import cv2 import numpy as np from core.raw_processor_core import RawProcessorCore from core.oak_fcc3_client import OakFcc3Client def load_json(path: Path) -> dict: with open(path, "r", encoding="utf-8") as f: return json.load(f) def ts_name() -> str: return datetime.now().strftime("%Y%m%d_%H%M%S_%f")[:-3] def ensure_dir(path: Path | str): Path(path).mkdir(parents=True, exist_ok=True) def save_multispec_tensor_from_raw_group( group: dict, meta: dict, out_dir: str = "calibration/offline_samples", ): """ Gera e salva um tensor MULTISPEC [5,H,W] float32 a partir de uma captura RAW_BRUTO. Saídas: .raw -> tensor float32 CHW .json -> metadados do tensor gerado offline .png -> preview RGB do tensor """ if meta.get("saved_payload_type") != "raw_native_multi": raise RuntimeError("Só é possível gerar tensor offline a partir de saved_payload_type='raw_native_multi'.") ensure_dir(out_dir) out_dir = Path(out_dir) tensor, desc, processing_info = build_multispec_from_raw_native_multi(group, meta) if tensor is None: raise RuntimeError(f"Falha ao gerar tensor MULTISPEC: {desc}") base_name = Path(group["json"]).stem name = f"{base_name}_offline_multispec" raw_path = out_dir / f"{name}.raw" json_path = out_dir / f"{name}.json" png_path = out_dir / f"{name}.png" tensor = np.ascontiguousarray(tensor.astype(np.float32, copy=False)) tensor.tofile(str(raw_path)) # Preview RGB do tensor rgb_hwc = np.transpose(tensor[:3], (1, 2, 0)) preview_bgr = normalize_float01_to_bgr(rgb_hwc) cv2.imwrite(str(png_path), preview_bgr) # JSON compatível com o validador e com análise posterior out_meta = { "ts": datetime.now().isoformat(timespec="milliseconds"), "schema": "offline_multispec_from_raw_native_multi_v1", "source_json": str(group["json"]), "source_saved_payload_type": meta.get("saved_payload_type"), "source_saved_payload_paths": meta.get("saved_payload_paths"), "source_saved_payload_shapes": meta.get("saved_payload_shapes"), "source_saved_payload_dtypes": meta.get("saved_payload_dtypes"), "camera_params_json": meta.get("camera_params_json"), "frame_type": "MULTISPEC", "saved_payload_type": "multispec", "saved_payload_path": raw_path.name, "saved_payload_dtype": "float32", "saved_payload_shape": list(tensor.shape), "channels": ["R", "G", "B", "RE", "NIR"], "saved_preview_path": png_path.name, "generation": { "method": "build_multispec_from_raw_native_multi", "description": desc, "same_frame_as_raw_bruto": True, }, "processing": processing_info or {}, "frame_quality": (processing_info or {}).get("frame_quality"), "patch_normalization_result": (processing_info or {}).get("patch_normalization_result"), "source_capture_meta": { "ts": meta.get("ts"), "sensor_width": meta.get("sensor_width"), "sensor_height": meta.get("sensor_height"), "bayer_pattern": meta.get("bayer_pattern"), "fps_target": meta.get("fps_target"), "startup_camera_controls": meta.get("startup_camera_controls"), "actual_camera_controls": meta.get("actual_camera_controls"), "radiometric_last_result": meta.get("radiometric_last_result"), "stream_meta": meta.get("stream_meta"), }, } with open(json_path, "w", encoding="utf-8") as f: json.dump(out_meta, f, ensure_ascii=False, indent=2) return { "tensor": tensor, "raw_path": raw_path, "json_path": json_path, "png_path": png_path, "desc": desc, } def normalize_float01_to_bgr(img_float: np.ndarray) -> np.ndarray: """ Recebe RGB float32 [0..1] em HWC e devolve BGR uint8. """ rgb_u8 = np.clip(img_float * 255.0, 0, 255).astype(np.uint8) return cv2.cvtColor(rgb_u8, cv2.COLOR_RGB2BGR) def chw_to_hwc(arr: np.ndarray) -> np.ndarray: if arr.ndim != 3: raise ValueError(f"Esperado CHW 3D, recebido shape={arr.shape}") return np.transpose(arr, (1, 2, 0)) def format_frame_quality_for_overlay(frame_quality: dict | None, patch_result: dict | None = None) -> str: """ Gera uma linha curta para mostrar no preview do tensor final. Exemplo: Q=good | sat=0.00% | dark=12.3% | clip=0.00% | scale=0.91-1.08 """ if not isinstance(frame_quality, dict): return "Q=n/a" status = frame_quality.get("status", "unknown") metrics = frame_quality.get("metrics", {}) or {} sat = float(metrics.get("max_tensor_sat_pct", 0.0) or 0.0) dark = float(metrics.get("max_tensor_dark_pct", 0.0) or 0.0) clip = float(metrics.get("max_patch_would_clip_pct", 0.0) or 0.0) scale_txt = "scale=n/a" if isinstance(patch_result, dict): summary = patch_result.get("summary", {}) or {} smin = summary.get("scale_min_applied") smax = summary.get("scale_max_applied") if smin is not None and smax is not None: try: scale_txt = f"scale={float(smin):.2f}-{float(smax):.2f}" except Exception: pass reasons = frame_quality.get("reasons", []) or [] reason_txt = "" if status != "good" and reasons: reason_txt = f" | {str(reasons[0])[:38]}" return f"Q={status} | sat={sat:.2f}% | dark={dark:.1f}% | clip={clip:.2f}% | {scale_txt}{reason_txt}" def tensor_to_preview_panels(tensor: np.ndarray, frame_quality: dict | None = None, patch_result: dict | None = None): """ Recebe tensor CHW [R,G,B,RE,NIR] float32 e devolve painéis visuais. Quando disponível, adiciona um resumo de qualidade no subtítulo do painel RGB final. """ if tensor.ndim != 3 or tensor.shape[0] < 5: raise RuntimeError(f"Tensor MULTISPEC inválido: shape={tensor.shape}") rgb_hwc = np.transpose(tensor[:3].astype(np.float32), (1, 2, 0)) rgb_bgr = normalize_float01_to_bgr(rgb_hwc) re01 = tensor[3].astype(np.float32) nir01 = tensor[4].astype(np.float32) re_bgr = cv2.cvtColor( np.clip(re01 * 255.0, 0, 255).astype(np.uint8), cv2.COLOR_GRAY2BGR ) nir_bgr = cv2.cvtColor( np.clip(nir01 * 255.0, 0, 255).astype(np.uint8), cv2.COLOR_GRAY2BGR ) quality_subtitle = format_frame_quality_for_overlay(frame_quality, patch_result) rgb_subtitle = f"tensor {list(tensor.shape)} | canais 0,1,2" if quality_subtitle: rgb_subtitle = f"{rgb_subtitle} | {quality_subtitle}" return [ ("MULTISPEC RGB final", rgb_bgr, rgb_subtitle), ("MULTISPEC RE final", re_bgr, "tensor canal 3"), ("MULTISPEC NIR final", nir_bgr, "tensor canal 4"), ] def build_multispec_from_raw_native_multi(group: dict, meta: dict): """ Reconstrói o tensor MULTISPEC final a partir dos .bin RAW_BRUTO salvos. Usa: - saved_payload_paths - saved_payload_shapes - saved_payload_dtypes - stream_meta.camera_info - camera_params_json/module_params.json """ if meta.get("saved_payload_type") != "raw_native_multi": return None, "captura não é raw_native_multi", {} stream_meta = meta.get("stream_meta", {}) or {} camera_info = stream_meta.get("camera_info", {}) or {} saved_dtypes = meta.get("saved_payload_dtypes", {}) or {} saved_shapes = meta.get("saved_payload_shapes", {}) or {} frame = {} for cam_id, path in group["cameras"].items(): saved_dtype = saved_dtypes.get(cam_id) saved_shape = saved_shapes.get(cam_id) if saved_dtype is None or saved_shape is None: raise RuntimeError(f"Faltam dtype/shape para {cam_id}") arr = np.fromfile(str(path), dtype=np.dtype(saved_dtype)).reshape(tuple(saved_shape)) frame[cam_id] = arr if not frame: raise RuntimeError("Nenhum payload de câmera encontrado para reconstruir MULTISPEC.") sensor_width = int(meta.get("sensor_width", 1280)) sensor_height = int(meta.get("sensor_height", 800)) bayer = meta.get("bayer_pattern", "BGGR") # Tenta usar o mesmo module_params que foi usado na captura. calib_path = meta.get("camera_params_json") or "calibration/module_params.json" # Se vier relativo, tenta resolver relativo ao diretório atual. # Normalmente seu script roda da raiz do projeto, então calibration/module_params.json funciona. if calib_path and not os.path.isfile(calib_path): # fallback: tenta relativo à pasta do JSON json_dir = Path(group["json"]).parent alt = json_dir / calib_path if alt.exists(): calib_path = str(alt) else: print(f"[WARN] module_params não encontrado: {calib_path}. Tentando sem calibração.") calib_path = None core = RawProcessorCore( sensor_width=sensor_width, sensor_height=sensor_height, bayer_pattern=bayer, calibration_json_path=calib_path, ) # O decode precisa do stream_meta com camera_info. processing_meta = dict(stream_meta) # A normalização radiométrica precisa dos controles reais salvos no JSON da captura. if meta.get("actual_camera_controls") is not None: processing_meta["actual_camera_controls"] = meta.get("actual_camera_controls") if meta.get("startup_camera_controls") is not None: processing_meta["startup_camera_controls"] = meta.get("startup_camera_controls") tensor = core.build_infer_tensor_from_stream(frame, processing_meta, 5) processing_info = { "patch_normalization_result": getattr(core, "last_patch_normalization_result", None), "frame_quality": getattr(core, "last_frame_quality_result", None), } return tensor, f"MULTISPEC gerado offline do RAW_BRUTO | shape={list(tensor.shape)}", processing_info def build_visual_from_saved_payload( payload_path: Path, meta: dict, cam_id: str | None = None, client: OakFcc3Client | None = None, ) -> tuple[np.ndarray, str]: """ Retorna: preview_bgr_reconstructed texto_descritivo """ saved_type = meta.get("saved_payload_type") # ========================================================= # Caso MULTI payload por câmera # ========================================================= if saved_type == "raw_native_multi": if cam_id is None: raise RuntimeError("cam_id é obrigatório para saved_payload_type='raw_native_multi'") if client is not None: return build_camera_preview_with_client( client=client, payload_path=payload_path, meta=meta, cam_id=cam_id, ) # ========================================================= # Caso payload único # ========================================================= saved_dtype = meta.get("saved_payload_dtype") saved_shape = meta.get("saved_payload_shape") if saved_type is None or saved_dtype is None or saved_shape is None: raise RuntimeError( "JSON não contém saved_payload_type / saved_payload_dtype / saved_payload_shape" ) np_dtype = np.dtype(saved_dtype) raw = np.fromfile(str(payload_path), dtype=np_dtype) arr = raw.reshape(tuple(saved_shape)) if saved_type == "rgb": if arr.ndim != 3 or arr.shape[0] != 3: raise RuntimeError(f"Payload RGB inválido, shape={arr.shape}") rgb_hwc = chw_to_hwc(arr.astype(np.float32)) preview_bgr = normalize_float01_to_bgr(rgb_hwc) desc = f"Reconstruido de RGB salvo | dtype={arr.dtype} | shape={arr.shape}" return preview_bgr, desc if saved_type == "multispec": if arr.ndim != 3 or arr.shape[0] < 5: raise RuntimeError(f"Payload MULTISPEC inválido, shape={arr.shape}") processing = meta.get("processing", {}) or {} frame_quality = meta.get("frame_quality") or processing.get("frame_quality") patch_result = meta.get("patch_normalization_result") or processing.get("patch_normalization_result") panels = tensor_to_preview_panels( arr.astype(np.float32), frame_quality=frame_quality, patch_result=patch_result, ) # Renomeia os painéis para indicar que vieram de um MULTISPEC já salvo. renamed = [] for title, img, subtitle in panels: title = title.replace("MULTISPEC RGB final", "RGB reconstruido") title = title.replace("MULTISPEC RE final", "RE reconstruido") title = title.replace("MULTISPEC NIR final", "NIR reconstruido") renamed.append((title, img, subtitle)) desc = f"Reconstruido de MULTISPEC | dtype={arr.dtype} | shape={arr.shape} | canais=[R,G,B,RE,NIR]" return renamed, desc if saved_type == "raw_native_single": if arr.ndim == 3 and arr.shape[2] == 3 and arr.dtype == np.uint8: preview_bgr = arr.copy() desc = f"Reconstruido de RAW nativo USB | dtype={arr.dtype} | shape={arr.shape}" return preview_bgr, desc stream_meta = meta.get("stream_meta", {}) source_camera = stream_meta.get("source_camera", {}) or {} bayer = source_camera.get("bayer_pattern", meta.get("bayer_pattern", "BGGR")) bit_depth = int(source_camera.get("bit_depth", 10)) sensor_height = int(meta.get("sensor_height")) sensor_width = int(meta.get("sensor_width")) core = RawProcessorCore( sensor_width=sensor_width, sensor_height=sensor_height, bayer_pattern=bayer, ) preview = RawProcessorPreview( sensor_width=sensor_width, sensor_height=sensor_height, bayer_pattern=bayer, ) packed = arr if packed.ndim == 3 and packed.shape[2] == 1: packed = packed[:, :, 0] raw16 = core.unpack_raw10_packed(packed) preview_bgr = preview.raw16_to_preview_bgr(raw16, bit_depth=bit_depth) desc = f"Reconstruido de RAW packed mono | dtype={arr.dtype} | shape={arr.shape} | bayer={bayer} | bit_depth={bit_depth}" return preview_bgr, desc if saved_type == "raw10_packed": stream_meta = meta.get("stream_meta", {}) or {} source_camera = stream_meta.get("source_camera", {}) or {} bayer = source_camera.get("bayer_pattern", meta.get("bayer_pattern", "BGGR")) bit_depth = int(source_camera.get("bit_depth", 10)) sensor_height = int(meta.get("sensor_height")) sensor_width = int(meta.get("sensor_width")) core = RawProcessorCore( sensor_width=sensor_width, sensor_height=sensor_height, bayer_pattern=bayer, ) preview = RawProcessorPreview( sensor_width=sensor_width, sensor_height=sensor_height, bayer_pattern=bayer, ) packed = arr if packed.ndim == 3 and packed.shape[2] == 1: packed = packed[:, :, 0] raw16 = core.unpack_raw10_packed(packed) preview_bgr = preview.raw16_to_preview_bgr(raw16, bit_depth=bit_depth) desc = ( f"Reconstruido de RAW10 packed | " f"dtype={arr.dtype} | shape={arr.shape} | " f"sensor={sensor_width}x{sensor_height} | " f"bayer={bayer} | bit_depth={bit_depth}" ) return preview_bgr, desc raise RuntimeError(f"saved_payload_type não suportado neste script: {saved_type}") def build_panels_from_group(group, client: OakFcc3Client | None = None): panels = [] meta = load_json(group["json"]) saved_type = meta.get("saved_payload_type") # ========================================================= # Para RAW_BRUTO multi, o primeiro painel vira o tensor final # gerado offline a partir dos .bin salvos. # ========================================================= if saved_type == "raw_native_multi": try: tensor, desc, processing_info = build_multispec_from_raw_native_multi(group, meta) tensor_panels = tensor_to_preview_panels( tensor, frame_quality=(processing_info or {}).get("frame_quality"), patch_result=(processing_info or {}).get("patch_normalization_result"), ) # Aqui colocamos só o RGB final como painel principal, # para substituir o antigo PNG salvo. panels.append(tensor_panels[0]) panels.append(tensor_panels[1]) panels.append(tensor_panels[2]) except Exception as e: # Fallback para o PNG salvo caso a reconstrução falhe. preview_saved = cv2.imread(str(group["png"]), cv2.IMREAD_COLOR) if preview_saved is None: raise RuntimeError(f"Falha ao ler preview PNG: {group['png']}") panels.append(( "Preview salvo fallback", preview_saved, f"Falha ao gerar MULTISPEC offline: {e}" )) else: # Para RGB/MULTISPEC salvos direto, mantém comportamento antigo. preview_saved = cv2.imread(str(group["png"]), cv2.IMREAD_COLOR) if preview_saved is None: raise RuntimeError(f"Falha ao ler preview PNG: {group['png']}") panels.append(("Preview salvo", preview_saved, f"{preview_saved.shape[1]}x{preview_saved.shape[0]}")) # ========================================================= # Se houver payload final único, reconstrói normalmente. # Ex: saved_payload_type == multispec # ========================================================= if group["final_raw"] is not None: 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)) # ========================================================= # Continua mostrando CAM_A/CAM_B/CAM_C reconstruídas individualmente. # ========================================================= for cam_id, path in group["cameras"].items(): img, desc = build_visual_from_saved_payload( path, meta, cam_id=cam_id, client=client, ) panels.append((f"{cam_id} reconstruido", img, desc)) return panels def compose_panels(panels, max_width=1600): imgs = [] # aplica label for title, img, subtitle in panels: img_labeled = put_label(img, title, subtitle) imgs.append(img_labeled) # normaliza tamanho base max_h = max(img.shape[0] for img in imgs) resized = [] for img in imgs: scale = max_h / img.shape[0] w = int(img.shape[1] * scale) resized.append(cv2.resize(img, (w, max_h), interpolation=cv2.INTER_NEAREST)) # ========================= # Montagem em grid 2x2 # ========================= rows = [] gap = np.full((max_h, 20, 3), 30, dtype=np.uint8) for i in range(0, len(resized), 2): row_imgs = resized[i:i+2] # se só tiver 1 imagem na linha, duplica espaço vazio if len(row_imgs) == 1: blank = np.zeros_like(row_imgs[0]) row_imgs.append(blank) row = np.hstack([row_imgs[0], gap, row_imgs[1]]) rows.append(row) # junta linhas gap_h = np.full((20, rows[0].shape[1], 3), 30, dtype=np.uint8) canvas = rows[0] for r in rows[1:]: canvas = np.vstack([canvas, gap_h, r]) # ========================= # Resize final # ========================= if canvas.shape[1] > max_width: scale = max_width / canvas.shape[1] canvas = cv2.resize( canvas, (int(canvas.shape[1] * scale), int(canvas.shape[0] * scale)), interpolation=cv2.INTER_AREA ) return canvas def sort_panels(panels): order = [ "multispec rgb final", "cam_a reconstruido", "multispec re final", "cam_b reconstruido", "multispec nir final", "cam_c reconstruido", "rgb reconstruido", "re reconstruido", "nir reconstruido" ] def key(p): title = p[0].lower() for i, k in enumerate(order): if k in title: return i return 99 return sorted(panels, key=key) def fit_same_height(img_a: np.ndarray, img_b: np.ndarray, target_h: int = None): if target_h is None: target_h = max(img_a.shape[0], img_b.shape[0]) def resize_to_h(img, h): scale = h / img.shape[0] w = int(img.shape[1] * scale) return cv2.resize(img, (w, h), interpolation=cv2.INTER_NEAREST) return resize_to_h(img_a, target_h), resize_to_h(img_b, target_h) def put_label(img: np.ndarray, title: str, subtitle: str = "") -> np.ndarray: out = img.copy() cv2.putText(out, title, (12, 28), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 0, 0), 3, cv2.LINE_AA) cv2.putText(out, title, (12, 28), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 255), 2, cv2.LINE_AA) if subtitle: # Quebra visual simples para linhas longas de debug/qualidade. subtitle_lines = [] current = "" for part in str(subtitle).split(" | "): candidate = part if not current else current + " | " + part if len(candidate) > 95 and current: subtitle_lines.append(current) current = part else: current = candidate if current: subtitle_lines.append(current) y = 56 for line in subtitle_lines[:3]: cv2.putText(out, line, (12, y), cv2.FONT_HERSHEY_SIMPLEX, 0.55, (0, 0, 0), 3, cv2.LINE_AA) cv2.putText(out, line, (12, y), cv2.FONT_HERSHEY_SIMPLEX, 0.55, (255, 255, 255), 1, cv2.LINE_AA) y += 22 return out def find_dataset_layout_root(path: Path) -> Path | None: """ Detecta layout de dataset: root/metas root/previews root/bins root/masks opcional Pode receber: - root do dataset - root/metas - root/previews - root/bins - arquivo dentro dessas pastas """ p = path.resolve() candidates = [] if p.is_file(): candidates.append(p.parent) candidates.append(p.parent.parent) else: candidates.append(p) candidates.append(p.parent) for c in candidates: if not c: continue # Se o usuário apontou diretamente para metas/previews/bins if c.name.lower() in ("metas", "metadata", "jsons", "previews", "bins", "masks"): root = c.parent else: root = c metas = root / "metas" previews = root / "previews" bins = root / "bins" if metas.is_dir() and previews.is_dir() and bins.is_dir(): return root return None def is_dataset_layout(path: Path) -> bool: return find_dataset_layout_root(path) is not None def resolve_dataset_sibling_file(root: Path, subdir: str, stem: str, exts: tuple[str, ...]) -> Path | None: folder = root / subdir if not folder.is_dir(): return None for ext in exts: p = folder / f"{stem}{ext}" if p.exists(): return p return None def resolve_dataset_json_from_any_path(input_path: Path, dataset_root: Path) -> Path: """ Dado um caminho qualquer dentro do dataset, descobre o JSON correspondente em metas/. """ input_path = input_path.resolve() if input_path.is_file(): stem = input_path.stem # Se vier algo tipo xxx_CAM_A.bin ou xxx_cam0.bin, remove sufixo de câmera. for token in ("_CAM_A", "_CAM_B", "_CAM_C", "_cam0", "_cam1", "_cam2", "_rgb", "_re", "_nir"): if token in stem: stem = stem.split(token)[0] break json_path = dataset_root / "metas" / f"{stem}.json" if json_path.exists(): return json_path if input_path.suffix.lower() == ".json" and input_path.parent.name.lower() == "metas": return input_path raise FileNotFoundError(f"Não consegui resolver JSON do dataset para: {input_path}") def resolve_capture_group(input_path: Path): """ Resolve todos os arquivos relacionados a uma captura. Suporta dois layouts: 1) Layout antigo, tudo na mesma pasta: captura.json captura.png captura.raw/.bin captura_CAM_A.bin ... 2) Layout dataset: root/metas/captura.json root/previews/captura.png root/bins/captura_CAM_A.bin ... root/masks/captura.png opcional Retorna: { "json": Path, "png": Path, "mask": Path | None, "final_raw": Path | None, "cameras": { "CAM_A": Path, ... }, "dataset_root": Path | None, } """ input_path = input_path.resolve() dataset_root = find_dataset_layout_root(input_path) # ========================================================= # Caso novo: dataset/metas, previews, bins, masks # ========================================================= if dataset_root is not None: if input_path.is_dir(): # Se vier a raiz do dataset ou subpasta, pega o primeiro json depois pela navegação. # Aqui só usamos fallback defensivo. metas = sorted((dataset_root / "metas").glob("*.json")) if not metas: raise RuntimeError(f"Nenhum JSON encontrado em: {dataset_root / 'metas'}") json_path = metas[0] elif input_path.suffix.lower() == ".json": json_path = input_path else: json_path = resolve_dataset_json_from_any_path(input_path, dataset_root) meta = load_json(json_path) base_name = json_path.stem png_path = resolve_dataset_sibling_file( dataset_root, "previews", base_name, (".png", ".jpg", ".jpeg"), ) if png_path is None: raise FileNotFoundError( f"Preview não encontrado para {base_name} em {dataset_root / 'previews'}" ) mask_path = resolve_dataset_sibling_file( dataset_root, "masks", base_name, (".png", ".tif", ".tiff", ".npy"), ) group = { "json": json_path, "png": png_path, "mask": mask_path, "final_raw": None, "cameras": {}, "dataset_root": dataset_root, } bins_dir = dataset_root / "bins" # Payload multi por câmera if "saved_payload_paths" in meta: for cam_id, fname in meta["saved_payload_paths"].items(): fname_path = Path(fname) candidates = [] # 1) caminho exatamente como veio no JSON, relativo ao bins/ candidates.append(bins_dir / fname_path.name) # 2) relativo à pasta do JSON candidates.append(json_path.parent / fname) # 3) relativo à raiz do dataset candidates.append(dataset_root / fname) # 4) fallback por padrões comuns candidates.extend([ bins_dir / f"{base_name}_{cam_id}.bin", bins_dir / f"{base_name}_{cam_id}.raw", bins_dir / f"{base_name}_{cam_id.lower()}.bin", bins_dir / f"{base_name}_{cam_id.lower()}.raw", ]) found = None for c in candidates: if c.exists(): found = c break if found is not None: group["cameras"][cam_id] = found else: print(f"[WARN] bin não encontrado para {cam_id}: {fname}") # Payload final único elif "saved_payload_path" in meta: fname = Path(meta["saved_payload_path"]) candidates = [ bins_dir / fname.name, json_path.parent / fname, dataset_root / fname, bins_dir / f"{base_name}.raw", bins_dir / f"{base_name}.bin", ] for c in candidates: if c.exists(): group["final_raw"] = c break if group["final_raw"] is None: print(f"[WARN] payload final não encontrado para {base_name}: {fname}") return group # ========================================================= # Caso antigo: tudo na mesma pasta # ========================================================= folder = input_path.parent name = input_path.stem # remove sufixo _camX se existir if "_cam" in name: base_name = name.split("_cam")[0] else: base_name = name json_path = folder / f"{base_name}.json" png_path = folder / f"{base_name}.png" if not json_path.exists(): raise FileNotFoundError(f"JSON não encontrado: {json_path}") if not png_path.exists(): raise FileNotFoundError(f"PNG não encontrado: {png_path}") meta = load_json(json_path) group = { "json": json_path, "png": png_path, "mask": None, "final_raw": None, "cameras": {}, "dataset_root": None, } # Caso MULTI payload if "saved_payload_paths" in meta: for cam_id, fname in meta["saved_payload_paths"].items(): path = folder / fname if path.exists(): group["cameras"][cam_id] = path # Caso payload único (.raw) elif "saved_payload_path" in meta: path = folder / meta["saved_payload_path"] if path.exists(): group["final_raw"] = path return group def resolve_sensor_dims_for_raw10_packed(arr: np.ndarray, cam_meta: dict, meta: dict) -> tuple[int, int]: """ Para CSI RAW10 packed: packed_width = ceil(sensor_width * 5 / 4) Na prática aqui usamos: sensor_width = packed_width * 4 // 5 Altura permanece a mesma. """ packed_h = int(arr.shape[0]) packed_w = int(arr.shape[1]) interface = str(cam_meta.get("interface", "")).upper() bit_depth = int(cam_meta.get("bit_depth", 10)) # USB ou RGB HWC não entra nessa lógica if interface == "USB": return packed_w, packed_h # Caso esperado: CSI RAW10 packed mono if bit_depth == 10: sensor_w = (packed_w * 4) // 5 sensor_h = packed_h return sensor_w, sensor_h # fallback conservador return packed_w, packed_h def list_capture_groups_from_dir(folder: Path) -> list[Path]: """ Lista todos os JSONs de captura. Suporta: - pasta antiga com *.json direto - dataset root com metas/*.json - dataset/metas - dataset/previews - dataset/bins """ folder = folder.resolve() if not folder.exists() or not folder.is_dir(): raise FileNotFoundError(f"Diretório não encontrado: {folder}") dataset_root = find_dataset_layout_root(folder) if dataset_root is not None: metas_dir = dataset_root / "metas" items = sorted(metas_dir.glob("*.json")) if not items: raise RuntimeError(f"Nenhum arquivo .json encontrado em: {metas_dir}") return items items = sorted(folder.glob("*.json")) if not items: raise RuntimeError(f"Nenhum arquivo .json encontrado em: {folder}") return items def resolve_navigation_inputs(input_path: Path) -> tuple[list[Path], int]: """ Retorna: entries: lista de JSONs de captura start_index: índice inicial baseado no input fornecido Aceita: - arquivo .json/.png/.bin/.raw do layout antigo - pasta antiga - root dataset - dataset/metas - dataset/previews - dataset/bins - arquivo dentro de metas/previews/bins """ input_path = input_path.resolve() # ========================================================= # Dataset layout # ========================================================= dataset_root = find_dataset_layout_root(input_path) if dataset_root is not None: entries = list_capture_groups_from_dir(dataset_root) if input_path.is_dir(): return entries, 0 if input_path.suffix.lower() == ".json": target_json = input_path.resolve() else: target_json = resolve_dataset_json_from_any_path(input_path, dataset_root).resolve() try: idx = entries.index(target_json) except ValueError: idx = 0 return entries, idx # ========================================================= # Layout antigo # ========================================================= if input_path.is_dir(): entries = list_capture_groups_from_dir(input_path) return entries, 0 if not input_path.exists(): raise FileNotFoundError(f"Arquivo não encontrado: {input_path}") folder = input_path.parent entries = list_capture_groups_from_dir(folder) if input_path.suffix.lower() == ".json": target_json = input_path.resolve() else: group = resolve_capture_group(input_path) target_json = group["json"].resolve() try: idx = entries.index(target_json) except ValueError: idx = 0 return entries, idx def render_group_to_canvas(json_path: Path, max_width: int, client: OakFcc3Client | None = None): group = resolve_capture_group(json_path) meta = load_json(group["json"]) panels = build_panels_from_group(group, client=client) panels = sort_panels(panels) canvas = compose_panels(panels, max_width=max_width) info = { "group": group, "json": group["json"], "png": group["png"], "mask": group.get("mask"), "dataset_root": group.get("dataset_root"), "final_raw": group["final_raw"], "cameras": group["cameras"], "meta": meta, } return canvas, info def load_camera_payload_from_saved(meta: dict, payload_path: Path, cam_id: str) -> np.ndarray: saved_dtypes = meta.get("saved_payload_dtypes", {}) or {} saved_shapes = meta.get("saved_payload_shapes", {}) or {} saved_dtype = saved_dtypes.get(cam_id) saved_shape = saved_shapes.get(cam_id) if saved_dtype is None or saved_shape is None: raise RuntimeError( f"JSON não contém saved_payload_dtypes/saved_payload_shapes para {cam_id}" ) raw = np.fromfile(str(payload_path), dtype=np.dtype(saved_dtype)) return raw.reshape(tuple(saved_shape)) def build_camera_preview_with_client( client: OakFcc3Client, payload_path: Path, meta: dict, cam_id: str, ) -> tuple[np.ndarray, str]: stream_meta = meta.get("stream_meta", {}) or {} camera_info = stream_meta.get("camera_info", {}) or {} cam_meta = camera_info.get(cam_id, {}) or {} role = cam_meta.get("role", cam_id) interface = str(cam_meta.get("interface", "")).upper() arr = load_camera_payload_from_saved(meta, payload_path, cam_id) # Caso RGB/USB nativo, mantém comportamento direto. if interface == "USB" or (arr.ndim == 3 and arr.shape[2] == 3 and arr.dtype == np.uint8): desc = f"{cam_id} | role={role} | USB/RGB nativo | dtype={arr.dtype} | shape={arr.shape}" return arr.copy(), desc if cam_id == "CAM_A": preview_bgr = client.build_save_preview_from_cam_a( packed_raw_by_camera={"CAM_A": arr}, meta_stream=stream_meta, sensor_width=int(meta.get("sensor_width", 1280)), sensor_height=int(meta.get("sensor_height", 800)), bayer_pattern=meta.get("bayer_pattern", "RGGB"), ) if preview_bgr is None: raise RuntimeError("client.build_save_preview_from_cam_a retornou None para CAM_A.") bayer = ( cam_meta.get("bayer_pattern") or cam_meta.get("bayer") or meta.get("bayer_pattern") or "RGGB" ) desc = ( f"{cam_id} | role={role} | preview centralizado no OakFcc3Client | " f"dtype={arr.dtype} | shape={arr.shape} | bayer={bayer}" ) return preview_bgr, desc # Para CAM_B/C por enquanto mantém mono visual simples via caminho antigo? # Melhor: usar build_visual_preview_from_raw do client, que também está centralizado. previews = client.build_visual_preview_from_raw( frame={cam_id: arr}, meta={"camera_info": {cam_id: cam_meta}}, ) if cam_id not in previews: raise RuntimeError(f"client.build_visual_preview_from_raw não retornou {cam_id}. Chaves={list(previews.keys())}") desc = ( f"{cam_id} | role={role} | preview via OakFcc3Client.build_visual_preview_from_raw | " f"dtype={arr.dtype} | shape={arr.shape}" ) return previews[cam_id], desc def main(): parser = argparse.ArgumentParser( description="Valida visualmente payload salvo (.bin/.raw/.json/.png) comparando com o preview .png" ) parser.add_argument("--input_path", help="Caminho para .json, .png, .bin, .raw ou diretório") parser.add_argument("--max-width", type=int, default=1600, help="Largura máxima da janela final") args = parser.parse_args() input_path = Path(args.input_path) entries, current_idx = resolve_navigation_inputs(input_path) client = OakFcc3Client( width=1280, height=800, bayer="RGGB", frame_type="RAW_BRUTO", capture_mode="SINGLE", raw_policy="allow_single", ) window_name = "Validacao payload | A=anterior | D=proximo | T=salva tensor offline | Q/Esc=sair" cv2.namedWindow(window_name, cv2.WINDOW_NORMAL) while True: current_json = entries[current_idx] canvas, info = render_group_to_canvas( current_json, max_width=args.max_width, client=client, ) # Cabeçalho adicional na imagem overlay = canvas.copy() text = f"{current_idx + 1}/{len(entries)} | {current_json.name}" cv2.putText(overlay, text, (12, overlay.shape[0] - 16), cv2.FONT_HERSHEY_SIMPLEX, 0.65, (0, 0, 0), 3, cv2.LINE_AA) cv2.putText(overlay, text, (12, overlay.shape[0] - 16), cv2.FONT_HERSHEY_SIMPLEX, 0.65, (0, 255, 0), 1, cv2.LINE_AA) cv2.imshow(window_name, overlay) meta = info["meta"] print("=" * 60) print(f"[{current_idx + 1}/{len(entries)}]") print("Entrada JSON :", info["json"]) print("PNG :", info["png"]) print("Dataset root :", info.get("dataset_root")) print("Mask :", info.get("mask")) print("Final RAW :", info["final_raw"]) print("Câmeras :", {k: str(v) for k, v in info["cameras"].items()}) print("saved_payload_type :", meta.get("saved_payload_type")) print("saved_payload_dtype:", meta.get("saved_payload_dtype")) print("saved_payload_shape:", meta.get("saved_payload_shape")) print("saved_payload_dtypes:", meta.get("saved_payload_dtypes")) print("saved_payload_shapes:", meta.get("saved_payload_shapes")) print("stream frame_type :", (meta.get("stream_meta") or {}).get("frame_type")) print("=" * 60) k = cv2.waitKey(0) & 0xFF if k in (ord("q"), ord("Q"), 27): break elif k in (ord("d"), ord("D")): current_idx = min(current_idx + 1, len(entries) - 1) elif k in (ord("a"), ord("A")): current_idx = max(current_idx - 1, 0) elif k in (ord("t"), ord("T")): meta = info["meta"] group = info["group"] try: result = save_multispec_tensor_from_raw_group( group=group, meta=meta, out_dir="calibration/offline_samples", ) print("[OK] Tensor MULTISPEC offline salvo:") print(" RAW :", result["raw_path"]) print(" JSON:", result["json_path"]) print(" PNG :", result["png_path"]) print(" DESC:", result["desc"]) except Exception as e: print("[ERRO] Falha ao salvar tensor MULTISPEC offline:", e) cv2.destroyAllWindows() if __name__ == "__main__": main()