import argparse import csv import gc import json import math import unicodedata from dataclasses import dataclass, field from pathlib import Path from typing import Dict, List, Optional, Tuple, Any import cv2 import numpy as np import shutil try: from core.raw_processor_core import RawProcessorCore except Exception: RawProcessorCore = None # ============================================================ # Auditoria final de dataset multiespectral OAK-FCC-3P # ------------------------------------------------------------ # Foco: tensor final pronto para o modelo, CHW float32: # [R, G, B, RE, NIR] # # Gera: # - audit_summary.json # - audit_samples.csv # - audit_by_class_channel.csv # - audit_by_sample_class_channel.csv # - audit_warnings.json # - visuals/*.png com painéis de sanidade # # Layout esperado: # dataset_root/ # metas/*.json # bins/*.raw ou *.bin # masks/*.png/.tif/.npy # previews/*.png opcional # # Também aceita apontar diretamente para dataset_root/metas etc. # # Auditoria read-only recomendada para o domínio OV9782: # python -m audit.audit_dataset --input_path .\dataset\original\group --out_dir .\audit\audit_multispec_ov9782_20260904 --module-params .\calibration\mp_ov9782\module_params.json --save-visuals --visual-every 25 --print-every 25 # # Esta versão é memory-safe: estatísticas globais usam reservoir limitado e # CSVs volumosos são gravados incrementalmente durante o processamento. # O modo --build-fixed-dataset continua disponível, mas não é necessário para uma auditoria diagnóstica. # # ============================================================ CHANNELS = ["R", "G", "B", "RE", "NIR"] DERIVED = [ "NDVI", "NDRE", "NIR_minus_RE", "NIR_over_RE", "NIR_over_R", "RE_over_R", "NIR_over_G", "RE_over_G", "G_minus_R", ] ALL_FEATURES = CHANNELS + DERIVED DEFAULT_CLASS_MAP = { 0: "chao", 1: "cana", 2: "erva", } # Labelmap das mascaras coloridas, informado pelo projeto. # Valores em RGB, como normalmente aparecem em ferramentas de anotacao. # Como o OpenCV le PNG em BGR, a funcao decode_color_mask converte internamente. DEFAULT_MASK_COLOR_MAP_RGB = { (128, 0, 0): 0, # chao (0, 0, 128): 1, # cana (0, 128, 0): 2, # erva } # Paleta de visualizacao em BGR para o painel GT mask. # Mantem visual equivalente ao labelmap RGB acima. DEFAULT_VIS_PALETTE_BGR = { 0: (0, 0, 128), # chao: RGB 128,0,0 1: (128, 0, 0), # cana: RGB 0,0,128 2: (0, 128, 0), # erva: RGB 0,128,0 255: (0, 0, 0), } EPS = 1e-6 # ============================================================ # Utilidades básicas # ============================================================ def ensure_dir(path: Path | str) -> Path: p = Path(path) p.mkdir(parents=True, exist_ok=True) return p def load_json(path: Path) -> dict: with open(path, "r", encoding="utf-8") as f: return json.load(f) def write_json(path: Path, data: Any): with open(path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) class IncrementalCsvWriter: """CSV incremental para não acumular milhares de linhas em RAM. O cabeçalho é definido pela primeira linha. As linhas são gravadas conforme cada amostra termina e podem ser inspecionadas mesmo durante a auditoria. """ def __init__(self, path: Path): self.path = Path(path) ensure_dir(self.path.parent) self.fp = None self.writer = None self.fieldnames = None self.rows_written = 0 def write(self, row: Dict[str, Any]): if self.writer is None: self.fieldnames = list(row.keys()) self.fp = self.path.open("w", encoding="utf-8", newline="") self.writer = csv.DictWriter(self.fp, fieldnames=self.fieldnames, extrasaction="ignore") self.writer.writeheader() self.writer.writerow(row) self.rows_written += 1 def flush(self): if self.fp is not None: self.fp.flush() def close(self): if self.fp is not None: self.fp.flush() self.fp.close() self.fp = None def ensure_file(self): if self.rows_written == 0 and not self.path.exists(): self.path.write_text("", encoding="utf-8") def safe_float(x: Any, default: float = 0.0) -> float: try: if x is None: return default v = float(x) if math.isnan(v) or math.isinf(v): return default return v except Exception: return default def parse_class_map(text: Optional[str]) -> Dict[int, str]: if not text: return dict(DEFAULT_CLASS_MAP) out = {} # Formatos aceitos: # "0:chao,1:cana,2:erva" # "0=chao,1=cana,2=erva" for item in text.split(","): item = item.strip() if not item: continue if ":" in item: k, v = item.split(":", 1) elif "=" in item: k, v = item.split("=", 1) else: raise ValueError(f"Classe inválida em --class-map: {item}") out[int(k.strip())] = v.strip() return out def normalize_to_u8(x: np.ndarray, p_low: float = 1.0, p_high: float = 99.0) -> np.ndarray: arr = x.astype(np.float32, copy=False) finite = np.isfinite(arr) if not np.any(finite): return np.zeros(arr.shape, dtype=np.uint8) vals = arr[finite] lo = np.percentile(vals, p_low) hi = np.percentile(vals, p_high) if hi <= lo + EPS: hi = lo + 1.0 y = (arr - lo) / (hi - lo) y = np.clip(y, 0, 1) return (y * 255.0).astype(np.uint8) def float01_to_u8(x: np.ndarray) -> np.ndarray: return np.clip(x.astype(np.float32) * 255.0, 0, 255).astype(np.uint8) def rgb_from_tensor(tensor: np.ndarray, stretch: bool = False) -> np.ndarray: rgb = np.transpose(tensor[:3], (1, 2, 0)).astype(np.float32) if stretch: chans = [normalize_to_u8(rgb[:, :, i]) for i in range(3)] rgb_u8 = np.dstack(chans) else: rgb_u8 = float01_to_u8(rgb) return cv2.cvtColor(rgb_u8, cv2.COLOR_RGB2BGR) def apply_colormap_gray(x: np.ndarray, stretch: bool = True) -> np.ndarray: if stretch: u8 = normalize_to_u8(x) else: u8 = float01_to_u8(x) return cv2.applyColorMap(u8, cv2.COLORMAP_VIRIDIS) def cv_text(text: Any) -> str: """ OpenCV putText nao lida bem com acentos/cedilha em muitos ambientes. Converte qualquer texto para ASCII seguro. """ s = str(text) s = unicodedata.normalize("NFKD", s) s = s.encode("ascii", "ignore").decode("ascii") return s def put_label(img: np.ndarray, title: str, subtitle: str = "") -> np.ndarray: out = img.copy() title = cv_text(title) subtitle = cv_text(subtitle) cv2.rectangle(out, (0, 0), (out.shape[1], 58 if subtitle else 36), (0, 0, 0), -1) cv2.putText(out, title, (10, 24), cv2.FONT_HERSHEY_SIMPLEX, 0.65, (0, 255, 255), 2, cv2.LINE_AA) if subtitle: cv2.putText(out, subtitle[:120], (10, 48), cv2.FONT_HERSHEY_SIMPLEX, 0.45, (255, 255, 255), 1, cv2.LINE_AA) return out def resize_keep(img: np.ndarray, size: Tuple[int, int]) -> np.ndarray: return cv2.resize(img, size, interpolation=cv2.INTER_AREA) def make_grid(panels: List[Tuple[str, np.ndarray, str]], panel_w: int = 360) -> np.ndarray: rendered = [] for title, img, subtitle in panels: scale = panel_w / img.shape[1] panel_h = max(1, int(img.shape[0] * scale)) small = cv2.resize(img, (panel_w, panel_h), interpolation=cv2.INTER_AREA) rendered.append(put_label(small, title, subtitle)) if not rendered: return np.zeros((200, 400, 3), dtype=np.uint8) max_h = max(x.shape[0] for x in rendered) padded = [] for img in rendered: if img.shape[0] < max_h: pad = np.zeros((max_h - img.shape[0], img.shape[1], 3), dtype=np.uint8) img = np.vstack([img, pad]) padded.append(img) cols = 3 rows = [] gap_w = np.full((max_h, 12, 3), 25, dtype=np.uint8) gap_h = np.full((12, cols * panel_w + (cols - 1) * 12, 3), 25, dtype=np.uint8) for i in range(0, len(padded), cols): row_imgs = padded[i:i + cols] while len(row_imgs) < cols: row_imgs.append(np.zeros_like(padded[0])) row = np.hstack([row_imgs[0], gap_w, row_imgs[1], gap_w, row_imgs[2]]) rows.append(row) canvas = rows[0] for r in rows[1:]: canvas = np.vstack([canvas, gap_h, r]) return canvas # ============================================================ # Detecção de layout e leitura dos tensores/máscaras # ============================================================ def is_dataset_root(path: Path) -> bool: """ Um dataset final válido tem, no mínimo: root/metas root/bins masks/ e previews/ são opcionais para a auditoria, embora masks/ seja necessário para estatísticas por classe. """ return (path / "metas").is_dir() and (path / "bins").is_dir() def find_dataset_roots(path: Path) -> List[Path]: """ Detecta um ou vários datasets. Suporta: 1) dataset direto: root/metas root/bins root/masks 2) subpasta do dataset: root/metas, root/bins etc, mas input_path=root/metas ou root/bins 3) super-root com grupos dentro: group/chao/metas group/chao/bins group/chao_cana/metas group/chao_cana/bins ... Isso resolve o caso: --input_path dataset/1024x640/group quando group contém vários datasets filhos. """ p = path.resolve() candidates: List[Path] = [] if p.is_file(): candidates.extend([p.parent, p.parent.parent]) else: candidates.extend([p, p.parent]) roots: List[Path] = [] # Primeiro tenta o próprio caminho ou pai direto. for c in candidates: if c.name.lower() in ("metas", "bins", "masks", "previews"): root = c.parent else: root = c if is_dataset_root(root): roots.append(root) # Se não achou, procura recursivamente datasets filhos. search_base = p if p.is_dir() else p.parent if not roots and search_base.exists(): for metas_dir in search_base.rglob("metas"): root = metas_dir.parent if is_dataset_root(root): roots.append(root) # Remove duplicados preservando ordem. unique: List[Path] = [] seen = set() for r in roots: rr = r.resolve() if rr not in seen: unique.append(rr) seen.add(rr) if not unique: raise FileNotFoundError( f"Não consegui detectar dataset_root a partir de {path}. " "Esperado root/metas e root/bins, ou um super-root contendo grupos com metas/bins." ) return unique def find_dataset_root(path: Path) -> Path: """ Compatibilidade com chamadas antigas: retorna o primeiro root encontrado. Para a auditoria principal, use find_dataset_roots(). """ return find_dataset_roots(path)[0] def list_meta_files(dataset_root: Path) -> List[Path]: metas = sorted((dataset_root / "metas").glob("*.json")) if not metas: raise RuntimeError(f"Nenhum .json encontrado em {dataset_root / 'metas'}") return metas def find_sibling(dataset_root: Path, subdir: str, stem: str, exts: Tuple[str, ...]) -> Optional[Path]: folder = dataset_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_mask_path(dataset_root: Path, meta_path: Path) -> Optional[Path]: """ Resolve mascara da amostra. Importante: o sample_name usado no relatorio pode receber prefixo do grupo tipo 'chao__2026...', mas o arquivo real da mascara continua usando meta_path.stem, sem prefixo. Esse foi o bug da rodada anterior. """ real_stem = meta_path.stem masks_dir = dataset_root / "masks" candidates: List[Path] = [] for ext in (".png", ".tif", ".tiff", ".npy"): candidates.append(masks_dir / f"{real_stem}{ext}") for suffix in ("_mask", "_gt", "_label", "_labels", "_seg"): for ext in (".png", ".tif", ".tiff", ".npy"): candidates.append(masks_dir / f"{real_stem}{suffix}{ext}") if masks_dir.is_dir(): candidates.extend(sorted(masks_dir.glob(f"{real_stem}*.*"))) for p in candidates: if p.exists() and p.suffix.lower() in (".png", ".tif", ".tiff", ".npy"): return p return None def resolve_tensor_payload(meta_path: Path, dataset_root: Path, meta: dict) -> Optional[Path]: """ Resolve payload final já pronto, quando o dataset já contém MULTISPEC salvo. Para RAW_BRUTO multi por câmera, use build_multispec_from_raw_native_multi(). """ stem = meta_path.stem bins = dataset_root / "bins" candidates = [] saved_payload_path = meta.get("saved_payload_path") if saved_payload_path: sp = Path(saved_payload_path) candidates.extend([ bins / sp.name, meta_path.parent / sp, dataset_root / sp, ]) candidates.extend([ bins / f"{stem}.raw", bins / f"{stem}.bin", bins / f"{stem}_multispec.raw", bins / f"{stem}_multispec.bin", bins / f"{stem}_offline_multispec.raw", bins / f"{stem}_offline_multispec.bin", ]) for c in candidates: if c.exists(): return c return None def resolve_camera_payloads(meta_path: Path, dataset_root: Path, meta: dict) -> Dict[str, Path]: """ Resolve os .bin/.raw por câmera quando saved_payload_type='raw_native_multi'. """ stem = meta_path.stem bins = dataset_root / "bins" paths = {} saved_payload_paths = meta.get("saved_payload_paths", {}) or {} for cam_id, fname in saved_payload_paths.items(): fp = Path(fname) candidates = [ bins / fp.name, meta_path.parent / fname, dataset_root / fname, bins / f"{stem}_{cam_id}.bin", bins / f"{stem}_{cam_id}.raw", bins / f"{stem}_{cam_id.lower()}.bin", bins / f"{stem}_{cam_id.lower()}.raw", ] found = None for c in candidates: if Path(c).exists(): found = Path(c) break if found is None: raise FileNotFoundError(f"Payload bruto não encontrado para {cam_id} em {meta_path.name}: {fname}") paths[cam_id] = found return paths def resolve_module_params_path( meta_path: Path, dataset_root: Path, meta: dict, cli_module_params: Optional[str] = None, ) -> str: """ Resolve o module_params usado para reconstruir o tensor. Regra importante: 1) se --module-params foi informado, ele é AUTORITATIVO e fail-closed; 2) sem override, tenta o caminho salvo no meta; 3) por último, mantém os fallbacks legados. Isso evita auditar RAW antigo com um module_params histórico diferente daquele usado no normalize atual. """ if cli_module_params: override = Path(str(cli_module_params)) candidates = [override] if override.is_absolute() else [Path.cwd() / override] for c in candidates: if c.is_file(): return str(c.resolve()) raise FileNotFoundError( f"--module-params não encontrado: {cli_module_params}. " "Como o override foi informado explicitamente, não farei fallback para o meta." ) calib_path = ( meta.get("camera_params_json") or meta.get("module_params_json") or "calibration/module_params.json" ) p = Path(calib_path) candidates = [] if p.is_absolute(): candidates.append(p) else: candidates.extend([ Path.cwd() / p, meta_path.parent / p, dataset_root / p, dataset_root.parent / p, dataset_root.parent.parent / p, Path.cwd() / "calibration" / "module_params.json", ]) for c in candidates: if c.is_file(): return str(c.resolve()) raise FileNotFoundError( f"module_params.json não encontrado. Valor no meta={calib_path}. " "Sem ele eu não consigo aplicar homografia/flat/radiometria igual ao pipeline final." ) def get_raw_processor_core( core_cache: Dict[Tuple[int, int, str, str], Any], sensor_width: int, sensor_height: int, bayer: str, calib_path: str, ): """ Reaproveita RawProcessorCore entre amostras. Isso evita recarregar flat-field e recriar caches de gain a cada imagem. A chave considera tamanho, bayer e module_params.json. """ key = (int(sensor_width), int(sensor_height), str(bayer), str(Path(calib_path).resolve())) if key not in core_cache: if RawProcessorCore is None: raise RuntimeError( "Não consegui importar RawProcessorCore. Rode com python -m utils.audit_dataset " "a partir da raiz do projeto, igual você fez, e confirme se core/raw_processor_core.py existe." ) core_cache[key] = RawProcessorCore( sensor_width=int(sensor_width), sensor_height=int(sensor_height), bayer_pattern=str(bayer), calibration_json_path=str(calib_path), ) return core_cache[key] def build_multispec_from_raw_native_multi( meta_path: Path, dataset_root: Path, meta: dict, core_cache: Dict[Tuple[int, int, str, str], Any], cli_module_params: Optional[str] = None, ) -> Tuple[np.ndarray, Path, Optional[dict]]: """ Reconstrói o tensor final [R,G,B,RE,NIR] a partir do RAW_BRUTO multi por câmera. Usa o mesmo princípio do script visual antigo: - saved_payload_paths - saved_payload_shapes - saved_payload_dtypes - stream_meta.camera_info - actual_camera_controls/startup_camera_controls - module_params.json com homografia/flat/radiometria """ saved_dtypes = meta.get("saved_payload_dtypes", {}) or {} saved_shapes = meta.get("saved_payload_shapes", {}) or {} cam_paths = resolve_camera_payloads(meta_path, dataset_root, meta) frame = {} for cam_id, payload_path in cam_paths.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} em {meta_path.name}") arr = np.fromfile(str(payload_path), dtype=np.dtype(saved_dtype)).reshape(tuple(saved_shape)) frame[cam_id] = arr sensor_width = int(meta.get("sensor_width", 1280)) sensor_height = int(meta.get("sensor_height", 800)) bayer = meta.get("bayer_pattern", "RGGB") calib_path = resolve_module_params_path( meta_path, dataset_root, meta, cli_module_params=cli_module_params, ) core = get_raw_processor_core( core_cache=core_cache, sensor_width=sensor_width, sensor_height=sensor_height, bayer=bayer, calib_path=calib_path, ) stream_meta = meta.get("stream_meta", {}) or {} processing_meta = dict(stream_meta) # Espelha o contrato do _6_normalize.py o mais de perto possível. processing_meta["frame_type"] = "RAW_BRUTO" if "camera_info" not in processing_meta and isinstance(meta.get("camera_info"), dict): processing_meta["camera_info"] = meta.get("camera_info") 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") if meta.get("radiometric_last_result") is not None: processing_meta["radiometric_last_result"] = meta.get("radiometric_last_result") tensor = core.build_infer_tensor_from_stream(frame, processing_meta, 5) fusion_result = getattr(core, "last_fusion_result", None) if isinstance(fusion_result, dict): # Evita depender de um objeto que possa ser sobrescrito na próxima amostra. fusion_result = json.loads(json.dumps(fusion_result, default=str)) else: fusion_result = None if tensor is None: raise RuntimeError(f"RawProcessorCore retornou tensor None para {meta_path.name}") tensor = np.asarray(tensor, dtype=np.float32) if tensor.ndim != 3: raise RuntimeError(f"Tensor reconstruído inválido em {meta_path.name}: shape={tensor.shape}") if tensor.shape[0] != 5 and tensor.shape[-1] == 5: tensor = np.transpose(tensor, (2, 0, 1)) if tensor.shape[0] < 5: raise RuntimeError(f"Tensor reconstruído precisa ter 5 canais, recebido shape={tensor.shape}") # Retorna o primeiro payload bruto só como referência de origem no CSV. first_payload = next(iter(cam_paths.values())) return np.ascontiguousarray(tensor[:5]), first_payload, fusion_result def load_multispec_tensor( meta_path: Path, dataset_root: Path, core_cache: Dict[Tuple[int, int, str, str], Any], cli_module_params: Optional[str] = None, ) -> Tuple[np.ndarray, dict, Path, Optional[dict]]: meta = load_json(meta_path) saved_type = meta.get("saved_payload_type") # Caso 1: dataset original bruto por câmera. Aqui o script monta o tensor final. if saved_type == "raw_native_multi" or "saved_payload_paths" in meta: tensor, source_payload, fusion_result = build_multispec_from_raw_native_multi( meta_path, dataset_root, meta, core_cache, cli_module_params=cli_module_params, ) return tensor, meta, source_payload, fusion_result # Caso 2: dataset já normalizado/MULTISPEC salvo como payload único. payload_path = resolve_tensor_payload(meta_path, dataset_root, meta) if payload_path is None: raise FileNotFoundError(f"Payload tensor final não encontrado para {meta_path.name}") dtype = meta.get("saved_payload_dtype", "float32") shape = meta.get("saved_payload_shape") if shape is None: # Alguns JSONs podem guardar isso em outro campo. shape = meta.get("tensor_shape") or meta.get("shape") if shape is None: raise RuntimeError( f"Shape do tensor não encontrado em {meta_path.name}. " "Esperado saved_payload_shape=[5,H,W]." ) arr = np.fromfile(str(payload_path), dtype=np.dtype(dtype)).reshape(tuple(shape)) arr = arr.astype(np.float32, copy=False) if arr.ndim != 3: raise RuntimeError(f"Tensor inválido em {payload_path}: shape={arr.shape}") if arr.shape[0] != 5 and arr.shape[-1] == 5: arr = np.transpose(arr, (2, 0, 1)) if arr.shape[0] < 5: raise RuntimeError(f"Tensor precisa ter 5 canais [R,G,B,RE,NIR], recebido shape={arr.shape}") # Payload já pronto não precisa reconstruir fusão. Se o meta normalizado # tiver guardado o fusion_result, reaproveita-o para alinhar a máscara. processing = meta.get("processing", {}) or {} fusion_result = processing.get("fusion_result") if not isinstance(fusion_result, dict): fusion_result = None return arr[:5], meta, payload_path, fusion_result def decode_color_mask(mask_img: np.ndarray, ignore_index: int) -> np.ndarray: """ Converte mascara colorida do labelmap para indices de classe. Seu labelmap atual em RGB: chao = 128,0,0 -> classe 0 cana = 0,0,128 -> classe 1 erva = 0,128,0 -> classe 2 O OpenCV le PNG como BGR, entao fazemos a conversao RGB->BGR antes de comparar. """ if mask_img.ndim == 2: return mask_img.astype(np.int32, copy=False) if mask_img.shape[2] == 4: bgr = mask_img[:, :, :3] else: bgr = mask_img[:, :, :3] out = np.full(mask_img.shape[:2], ignore_index, dtype=np.int32) for rgb_color, cls_id in DEFAULT_MASK_COLOR_MAP_RGB.items(): r, g, b = rgb_color bgr_color = np.array([b, g, r], dtype=np.uint8) hit = np.all(bgr == bgr_color, axis=2) out[hit] = int(cls_id) # Fallback: se por algum motivo a imagem ja veio em ordem RGB, tenta tambem RGB direto. # Isso evita quebrar caso alguma leitura futura nao use cv2. for rgb_color, cls_id in DEFAULT_MASK_COLOR_MAP_RGB.items(): rgb_arr = np.array(rgb_color, dtype=np.uint8) hit = np.all(bgr == rgb_arr, axis=2) # So preenche pixels ainda nao identificados para evitar troca em cores ambiguas. out[(out == ignore_index) & hit] = int(cls_id) return out def load_mask( mask_path: Optional[Path], target_hw: Tuple[int, int], ignore_index: int, fusion_result: Optional[dict] = None, ) -> Optional[np.ndarray]: """ Carrega a máscara e reproduz o MESMO contrato geométrico do normalize: mask no espaço RGB/CAM_A -> ref_shape da fusão -> crop_box comum da fusão -> resize final do tensor Se fusion_result não existir, mantém o fallback legado de resize direto. """ if mask_path is None: return None if mask_path.suffix.lower() == ".npy": mask = np.load(str(mask_path)) if mask.ndim == 3: mask = decode_color_mask(mask.astype(np.uint8), ignore_index) else: mask_img = cv2.imread(str(mask_path), cv2.IMREAD_UNCHANGED) if mask_img is None: raise RuntimeError(f"Falha ao ler mascara: {mask_path}") mask = decode_color_mask(mask_img, ignore_index) mask = mask.astype(np.int32, copy=False) if isinstance(fusion_result, dict): ref_shape = fusion_result.get("ref_shape") if isinstance(ref_shape, (list, tuple)) and len(ref_shape) == 2: ref_h, ref_w = int(ref_shape[0]), int(ref_shape[1]) if ref_h > 0 and ref_w > 0 and mask.shape[:2] != (ref_h, ref_w): mask = cv2.resize( mask, (ref_w, ref_h), interpolation=cv2.INTER_NEAREST, ) crop_box = fusion_result.get("crop_box") if isinstance(crop_box, (list, tuple)) and len(crop_box) == 4: x0, y0, x1, y1 = [int(v) for v in crop_box] mh, mw = mask.shape[:2] x0 = max(0, min(mw - 1, x0)) x1 = max(x0 + 1, min(mw, x1)) y0 = max(0, min(mh - 1, y0)) y1 = max(y0 + 1, min(mh, y1)) mask = mask[y0:y1, x0:x1] h, w = target_hw if mask.shape[:2] != (h, w): mask = cv2.resize(mask, (w, h), interpolation=cv2.INTER_NEAREST) return mask def mask_unique_summary(mask: Optional[np.ndarray], max_items: int = 20) -> str: if mask is None: return "none" vals, counts = np.unique(mask, return_counts=True) parts = [] for v, c in zip(vals[:max_items], counts[:max_items]): cls_name = DEFAULT_CLASS_MAP.get(int(v), "ignore" if int(v) == 255 else "unk") parts.append(f"{int(v)}:{cls_name}:{int(c)}") if len(vals) > max_items: parts.append("...") return ",".join(parts) # ============================================================ # Features espectrais e estatísticas # ============================================================ def compute_feature_maps(tensor: np.ndarray) -> Dict[str, np.ndarray]: r, g, b, re, nir = [tensor[i].astype(np.float32, copy=False) for i in range(5)] features = { "R": r, "G": g, "B": b, "RE": re, "NIR": nir, "NDVI": (nir - r) / (nir + r + EPS), "NDRE": (nir - re) / (nir + re + EPS), "NIR_minus_RE": nir - re, "NIR_over_RE": nir / (re + EPS), "NIR_over_R": nir / (r + EPS), "RE_over_R": re / (r + EPS), "NIR_over_G": nir / (g + EPS), "RE_over_G": re / (g + EPS), "G_minus_R": g - r, } # Evita explosões absurdas em razão quando denominador está quase zero. for k in list(features.keys()): features[k] = np.nan_to_num(features[k], nan=0.0, posinf=0.0, neginf=0.0).astype(np.float32) return features def calc_stats(values: np.ndarray, raw01: bool = False) -> Dict[str, float]: v = values.astype(np.float32, copy=False) v = v[np.isfinite(v)] if v.size == 0: return { "count": 0, "mean": 0.0, "std": 0.0, "min": 0.0, "p01": 0.0, "p05": 0.0, "p25": 0.0, "p50": 0.0, "p75": 0.0, "p95": 0.0, "p99": 0.0, "max": 0.0, "iqr": 0.0, "p95_p05": 0.0, "dark_pct": 0.0, "sat_pct": 0.0, } p = np.percentile(v, [1, 5, 25, 50, 75, 95, 99]) out = { "count": int(v.size), "mean": float(np.mean(v)), "std": float(np.std(v)), "min": float(np.min(v)), "p01": float(p[0]), "p05": float(p[1]), "p25": float(p[2]), "p50": float(p[3]), "p75": float(p[4]), "p95": float(p[5]), "p99": float(p[6]), "max": float(np.max(v)), "iqr": float(p[4] - p[2]), "p95_p05": float(p[5] - p[1]), "dark_pct": 0.0, "sat_pct": 0.0, } if raw01: out["dark_pct"] = float(np.mean(v <= 0.01) * 100.0) out["sat_pct"] = float(np.mean(v >= 0.99) * 100.0) return out class RunningFeatureStats: """Acumulador de estatísticas com memória limitada. A versão antiga guardava até ``max_samples`` pixels POR FEATURE/POR AMOSTRA em listas Python. Em milhares de imagens isso cresce para dezenas de GB. Esta versão mantém: - contagem, soma, soma de quadrados, min/max e dark/sat de forma EXATA; - um reservoir float32 de tamanho fixo por feature apenas para percentis. Assim a RAM fica praticamente constante ao longo da auditoria. """ def __init__(self, reservoir_size: int = 200000): self.reservoir_size = max(1000, int(reservoir_size)) self.reservoir: Dict[str, np.ndarray] = { f: np.empty((0,), dtype=np.float32) for f in ALL_FEATURES } self.sampled_seen: Dict[str, int] = {f: 0 for f in ALL_FEATURES} self.counts: Dict[str, int] = {f: 0 for f in ALL_FEATURES} self.sums: Dict[str, float] = {f: 0.0 for f in ALL_FEATURES} self.sumsq: Dict[str, float] = {f: 0.0 for f in ALL_FEATURES} self.mins: Dict[str, float] = {f: float("inf") for f in ALL_FEATURES} self.maxs: Dict[str, float] = {f: float("-inf") for f in ALL_FEATURES} self.dark_counts: Dict[str, int] = {f: 0 for f in ALL_FEATURES} self.sat_counts: Dict[str, int] = {f: 0 for f in ALL_FEATURES} def _merge_reservoir(self, feature_name: str, candidates: np.ndarray): """Mescla um lote no reservoir sem crescimento de memória. Usa uma amostragem hipergeométrica para decidir quantos elementos do lote novo devem entrar em uma amostra uniforme de tamanho K sobre todos os candidatos vistos até agora. """ if candidates.size == 0: return candidates = np.ascontiguousarray(candidates, dtype=np.float32) old_seen = int(self.sampled_seen[feature_name]) new_seen = int(candidates.size) total_seen = old_seen + new_seen target = min(self.reservoir_size, total_seen) old_res = self.reservoir[feature_name] if total_seen <= self.reservoir_size: if old_res.size == 0: merged = candidates.copy() else: merged = np.concatenate((old_res, candidates)).astype(np.float32, copy=False) else: # Quantos itens do lote novo pertencem a uma amostra uniforme de # tamanho ``target`` sobre old_seen + new_seen candidatos. take_new = int(np.random.hypergeometric(new_seen, old_seen, target)) take_old = target - take_new if take_old > 0: if take_old >= old_res.size: old_part = old_res else: idx_old = np.random.choice(old_res.size, size=take_old, replace=False) old_part = old_res[idx_old] else: old_part = np.empty((0,), dtype=np.float32) if take_new > 0: if take_new >= candidates.size: new_part = candidates else: idx_new = np.random.choice(candidates.size, size=take_new, replace=False) new_part = candidates[idx_new] else: new_part = np.empty((0,), dtype=np.float32) merged = np.concatenate((old_part, new_part)).astype(np.float32, copy=False) self.reservoir[feature_name] = np.ascontiguousarray(merged, dtype=np.float32) self.sampled_seen[feature_name] = total_seen def add(self, feature_name: str, values: np.ndarray, max_samples: int = 25000): v = np.asarray(values, dtype=np.float32).reshape(-1) finite = np.isfinite(v) if not np.any(finite): return v = v[finite] if v.size == 0: return # Estatísticas exatas sobre TODOS os pixels finitos. n = int(v.size) self.counts[feature_name] += n vf64 = v.astype(np.float64, copy=False) self.sums[feature_name] += float(np.sum(vf64, dtype=np.float64)) self.sumsq[feature_name] += float(np.dot(vf64, vf64)) self.mins[feature_name] = min(self.mins[feature_name], float(np.min(v))) self.maxs[feature_name] = max(self.maxs[feature_name], float(np.max(v))) if feature_name in CHANNELS: self.dark_counts[feature_name] += int(np.count_nonzero(v <= 0.01)) self.sat_counts[feature_name] += int(np.count_nonzero(v >= 0.99)) # Percentis usam somente um subconjunto por amostra e um reservoir # global fixo. Mantém a mesma filosofia da versão antiga, sem acumular. if max_samples > 0 and v.size > max_samples: idx = np.random.choice(v.size, size=max_samples, replace=False) candidates = v[idx] else: candidates = v self._merge_reservoir(feature_name, candidates) def summarize(self) -> Dict[str, Dict[str, float]]: out: Dict[str, Dict[str, float]] = {} for f in ALL_FEATURES: count = int(self.counts.get(f, 0)) arr = self.reservoir.get(f, np.empty((0,), dtype=np.float32)) if count <= 0: s = calc_stats(np.empty((0,), dtype=np.float32), raw01=(f in CHANNELS)) s["total_pixels_seen"] = 0 s["sampled_values"] = 0 out[f] = s continue # Percentis aproximados pelo reservoir. if arr.size > 0: p = np.percentile(arr, [1, 5, 25, 50, 75, 95, 99]) p01, p05, p25, p50, p75, p95, p99 = [float(x) for x in p] else: p01 = p05 = p25 = p50 = p75 = p95 = p99 = 0.0 mean = self.sums[f] / count var = max(0.0, self.sumsq[f] / count - mean * mean) std = math.sqrt(var) s = { "count": count, "mean": float(mean), "std": float(std), "min": float(self.mins[f]), "p01": p01, "p05": p05, "p25": p25, "p50": p50, "p75": p75, "p95": p95, "p99": p99, "max": float(self.maxs[f]), "iqr": float(p75 - p25), "p95_p05": float(p95 - p05), "dark_pct": 0.0, "sat_pct": 0.0, "total_pixels_seen": count, "sampled_values": int(arr.size), } if f in CHANNELS: s["dark_pct"] = 100.0 * self.dark_counts[f] / count s["sat_pct"] = 100.0 * self.sat_counts[f] / count out[f] = s return out # ============================================================ # Alinhamento por borda/correlação # ============================================================ def gradient_mag(x: np.ndarray) -> np.ndarray: u8 = normalize_to_u8(x) gx = cv2.Sobel(u8, cv2.CV_32F, 1, 0, ksize=3) gy = cv2.Sobel(u8, cv2.CV_32F, 0, 1, ksize=3) mag = cv2.magnitude(gx, gy) return mag.astype(np.float32) def estimate_shift_phase(a: np.ndarray, b: np.ndarray) -> Tuple[float, float, float]: # dx, dy estimados entre mapas. Valores grandes sugerem desalinhamento residual. aa = normalize_to_u8(a).astype(np.float32) bb = normalize_to_u8(b).astype(np.float32) try: (dx, dy), response = cv2.phaseCorrelate(aa, bb) return float(dx), float(dy), float(response) except Exception: return 0.0, 0.0, 0.0 def edge_agreement(a: np.ndarray, b: np.ndarray) -> Dict[str, float]: ga = gradient_mag(a) gb = gradient_mag(b) va = ga.reshape(-1) vb = gb.reshape(-1) if np.std(va) < EPS or np.std(vb) < EPS: corr = 0.0 else: corr = float(np.corrcoef(va, vb)[0, 1]) dx, dy, resp = estimate_shift_phase(ga, gb) return { "edge_corr": corr, "phase_dx": dx, "phase_dy": dy, "phase_response": resp, } def make_edge_overlay(tensor: np.ndarray) -> np.ndarray: r, g, b, re, nir = [tensor[i] for i in range(5)] rgb_gray = (0.299 * r + 0.587 * g + 0.114 * b).astype(np.float32) e_rgb = normalize_to_u8(gradient_mag(rgb_gray), 5, 99) e_re = normalize_to_u8(gradient_mag(re), 5, 99) e_nir = normalize_to_u8(gradient_mag(nir), 5, 99) # BGR: RGB edge em verde, RE em vermelho, NIR em azul. overlay = np.zeros((tensor.shape[1], tensor.shape[2], 3), dtype=np.uint8) overlay[:, :, 1] = e_rgb overlay[:, :, 2] = e_re overlay[:, :, 0] = e_nir return overlay # ============================================================ # Visualizações por amostra # ============================================================ def colorize_mask(mask: Optional[np.ndarray], class_map: Dict[int, str], target_hw: Tuple[int, int]) -> np.ndarray: h, w = target_hw out = np.zeros((h, w, 3), dtype=np.uint8) if mask is None: return out palette = dict(DEFAULT_VIS_PALETTE_BGR) for cls_id in np.unique(mask): cls_id = int(cls_id) if cls_id in palette: out[mask == cls_id] = palette[cls_id] elif cls_id in class_map: rng = np.random.default_rng(cls_id) out[mask == cls_id] = rng.integers(40, 220, size=3) return out def mask_edges_on_rgb(rgb_bgr: np.ndarray, mask: Optional[np.ndarray]) -> np.ndarray: out = rgb_bgr.copy() if mask is None: return out m = mask.astype(np.uint8) edges = cv2.Canny(m, 0, 1) out[edges > 0] = (0, 255, 255) return out def make_sample_visual( sample_name: str, tensor: np.ndarray, mask: Optional[np.ndarray], class_map: Dict[int, str], sample_stats: Dict[str, Any], ) -> np.ndarray: features = compute_feature_maps(tensor) h, w = tensor.shape[1], tensor.shape[2] rgb = rgb_from_tensor(tensor, stretch=False) rgb_stretch = rgb_from_tensor(tensor, stretch=True) re = apply_colormap_gray(features["RE"], stretch=True) nir = apply_colormap_gray(features["NIR"], stretch=True) ndvi = apply_colormap_gray(features["NDVI"], stretch=True) ndre = apply_colormap_gray(features["NDRE"], stretch=True) diff = apply_colormap_gray(features["NIR_minus_RE"], stretch=True) ratio = apply_colormap_gray(features["NIR_over_RE"], stretch=True) edge = make_edge_overlay(tensor) mask_bgr = colorize_mask(mask, class_map, (h, w)) overlay_mask = rgb.copy() if mask is not None: overlay_mask = cv2.addWeighted(rgb, 0.65, mask_bgr, 0.35, 0) mask_edges = mask_edges_on_rgb(rgb, mask) align = sample_stats.get("alignment", {}) re_align = align.get("RGBgray_vs_RE", {}) nir_align = align.get("RGBgray_vs_NIR", {}) panels = [ ("RGB tensor", rgb, sample_name), ("RGB stretch", rgb_stretch, "visual apenas para contraste"), ("GT mask", mask_bgr, f"unique={mask_unique_summary(mask)}"), ("Mask overlay", overlay_mask, "GT over RGB"), ("Mask edges", mask_edges, "GT edges over RGB"), ("RE", re, "canal 3 | stretch p1-p99"), ("NIR", nir, "canal 4 | stretch p1-p99"), ("NDVI", ndvi, "(NIR-R)/(NIR+R)"), ("NDRE", ndre, "(NIR-RE)/(NIR+RE)"), ("NIR - RE", diff, "diferença direta"), ("NIR / RE", ratio, "razão com eps"), ( "Edge overlay", edge, f"G=RGB | R=RE | B=NIR | REcorr={safe_float(re_align.get('edge_corr')):.3f} NIRcorr={safe_float(nir_align.get('edge_corr')):.3f}", ), ] return make_grid(panels, panel_w=380) # ============================================================ # Auditoria principal # ============================================================ def resolve_rejected_preview_root(fixed_root: Path) -> Path: # fixed_root normalmente é dataset/fixed/group # queremos dataset/fixed/rejected_previews if fixed_root.name == "group": return fixed_root.parent / "rejected_previews" return fixed_root / "_rejected_previews" def find_original_preview(dataset_root: Path, real_stem: str) -> Optional[Path]: previews_dir = dataset_root / "previews" if not previews_dir.is_dir(): return None for ext in (".png", ".jpg", ".jpeg", ".webp"): p = previews_dir / f"{real_stem}{ext}" if p.exists(): return p # fallback caso tenha sufixo no nome matches = [] for ext in (".png", ".jpg", ".jpeg", ".webp"): matches.extend(previews_dir.glob(f"{real_stem}*{ext}")) return matches[0] if matches else None def resize_preview_max_width(img: np.ndarray, max_width: int) -> np.ndarray: if img is None or img.size == 0: return img if max_width <= 0 or img.shape[1] <= max_width: return img scale = max_width / img.shape[1] new_w = int(img.shape[1] * scale) new_h = int(img.shape[0] * scale) return cv2.resize(img, (new_w, new_h), interpolation=cv2.INTER_AREA) def make_rejected_filename(row: Dict[str, Any]) -> str: sample = str(row.get("sample", "sample")) reasons = str(row.get("reject_reasons", "")) # Nome curto com principais motivos. tags = [] if "abs_shift" in reasons: tags.append("absShift") if "outlier" in reasons: tags.append("outlier") if "low_edge_corr" in reasons: tags.append("lowCorr") if "too_small" in reasons: tags.append("smallTarget") if "missing_mask" in reasons: tags.append("noMask") tag = "_".join(tags) if tags else "rejected" return f"{sample}__{tag}.png" def draw_rejected_header(img: np.ndarray, row: Dict[str, Any]) -> np.ndarray: if img is None or img.size == 0: return img out = img.copy() h, w = out.shape[:2] header_h = 92 canvas = np.zeros((h + header_h, w, 3), dtype=np.uint8) canvas[:header_h, :] = (20, 20, 20) canvas[header_h:, :] = out sample = cv_text(row.get("sample", "")) reasons = cv_text(row.get("reject_reasons", "")) re_mag = safe_float(row.get("re_shift_mag")) nir_mag = safe_float(row.get("nir_shift_mag")) max_dev = safe_float(row.get("max_shift_dev_px")) re_corr = safe_float(row.get("re_edge_corr")) nir_corr = safe_float(row.get("nir_edge_corr")) line1 = sample[:120] line2 = f"RE={re_mag:.1f}px NIR={nir_mag:.1f}px dev={max_dev:.1f}px corrRE={re_corr:.3f} corrNIR={nir_corr:.3f}" line3 = reasons[:150] cv2.putText(canvas, line1, (10, 24), cv2.FONT_HERSHEY_SIMPLEX, 0.58, (0, 255, 255), 2, cv2.LINE_AA) cv2.putText(canvas, line2, (10, 52), cv2.FONT_HERSHEY_SIMPLEX, 0.50, (255, 255, 255), 1, cv2.LINE_AA) cv2.putText(canvas, line3, (10, 78), cv2.FONT_HERSHEY_SIMPLEX, 0.43, (120, 220, 255), 1, cv2.LINE_AA) return canvas def save_rejected_preview( row: Dict[str, Any], fixed_root: Path, args, tensor: Optional[np.ndarray] = None, mask: Optional[np.ndarray] = None, class_map: Optional[Dict[int, str]] = None, ): rejected_root = resolve_rejected_preview_root(fixed_root) group = str(row.get("group", "unknown")) dataset_root = Path(str(row["dataset_root"])) real_stem = str(row["real_stem"]) dst_dir = ensure_dir(rejected_root / group) dst_path = dst_dir / make_rejected_filename(row) source_mode = str(args.rejected_preview_source) img = None # 1) preview original, se existir. if source_mode in ("auto", "preview"): preview_path = find_original_preview(dataset_root, real_stem) if preview_path is not None: img = cv2.imread(str(preview_path), cv2.IMREAD_COLOR) # 2) painel completo do audit, se já existir e o usuário pediu. # Aqui é útil quando --save-visuals também estiver ligado. if img is None and source_mode in ("auto", "audit_panel"): # Procura por qualquer painel visual que contenha o sample no nome. # Fica em out_dir/visuals, mas não temos out_dir aqui. Então esse modo # fica mais útil se você passar painel diretamente no futuro. pass # 3) RGB reconstruído do tensor, se disponível. if img is None and tensor is not None and source_mode in ("auto", "rgb_tensor", "audit_panel"): img = rgb_from_tensor(tensor, stretch=False) if mask is not None and class_map is not None: mask_bgr = colorize_mask(mask, class_map, (tensor.shape[1], tensor.shape[2])) img = cv2.addWeighted(img, 0.70, mask_bgr, 0.30, 0) if img is None: # Último fallback: imagem preta com cabeçalho, para não perder rastreabilidade. img = np.zeros((360, 640, 3), dtype=np.uint8) img = resize_preview_max_width(img, int(args.rejected_preview_max_width)) img = draw_rejected_header(img, row) cv2.imwrite(str(dst_path), img) def robust_median_mad(values: List[float]) -> Tuple[float, float]: arr = np.asarray([v for v in values if np.isfinite(v)], dtype=np.float32) if arr.size == 0: return 0.0, 1.0 med = float(np.median(arr)) mad = float(np.median(np.abs(arr - med))) # Evita divisão por zero em grupos muito estáveis. if mad < 0.5: mad = 0.5 return med, mad def safe_copy_file(src: Path, dst: Path): if not src.exists(): return ensure_dir(dst.parent) shutil.copy2(str(src), str(dst)) def resolve_fixed_out_root(dataset_roots: List[Path], args) -> Path: if args.fixed_out_root: return Path(args.fixed_out_root).resolve() # Esperado: # dataset/original/group/chao # dataset/original/group/chao_cana # # Queremos: # dataset/fixed/group first = dataset_roots[0].resolve() # Se o root é .../original/group/chao, sobe 3: chao -> group -> original -> dataset if first.parent.name == "group" and first.parent.parent.name == "original": dataset_base = first.parent.parent.parent return dataset_base / "fixed" / "group" # Fallback seguro. return first.parent / "fixed" / "group" def sample_shift_metrics(row: Dict[str, Any]) -> Dict[str, float]: re_dx = safe_float(row.get("re_phase_dx")) re_dy = safe_float(row.get("re_phase_dy")) nir_dx = safe_float(row.get("nir_phase_dx")) nir_dy = safe_float(row.get("nir_phase_dy")) return { "re_dx": re_dx, "re_dy": re_dy, "nir_dx": nir_dx, "nir_dy": nir_dy, "re_mag": math.hypot(re_dx, re_dy), "nir_mag": math.hypot(nir_dx, nir_dy), } def build_group_shift_baselines(sample_rows: List[Dict[str, Any]]) -> Dict[str, Dict[str, Tuple[float, float]]]: grouped: Dict[str, Dict[str, List[float]]] = {} for row in sample_rows: group = str(row.get("group", "unknown")) grouped.setdefault(group, { "re_dx": [], "re_dy": [], "nir_dx": [], "nir_dy": [], "re_mag": [], "nir_mag": [], }) m = sample_shift_metrics(row) for k, v in m.items(): grouped[group][k].append(v) baselines: Dict[str, Dict[str, Tuple[float, float]]] = {} for group, vals in grouped.items(): baselines[group] = {} for k, arr in vals.items(): baselines[group][k] = robust_median_mad(arr) return baselines def classify_sample_for_training( row: Dict[str, Any], baselines: Dict[str, Dict[str, Tuple[float, float]]], class_map: Dict[int, str], args, ) -> Tuple[bool, List[str], Dict[str, float]]: reasons: List[str] = [] metrics = sample_shift_metrics(row) group = str(row.get("group", "unknown")) base = baselines.get(group, {}) if not bool(row.get("mask_found")): reasons.append("missing_mask") # Shift absoluto extremo. if metrics["re_mag"] > args.clean_max_abs_shift_px: reasons.append(f"re_abs_shift_high:{metrics['re_mag']:.2f}") if metrics["nir_mag"] > args.clean_max_abs_shift_px: reasons.append(f"nir_abs_shift_high:{metrics['nir_mag']:.2f}") # Desvio robusto por grupo. max_dev = 0.0 for k in ("re_dx", "re_dy", "nir_dx", "nir_dy", "re_mag", "nir_mag"): med, mad = base.get(k, (0.0, 1.0)) dev_abs = abs(metrics[k] - med) max_dev = max(max_dev, dev_abs) if dev_abs > args.clean_max_dev_shift_px: reasons.append(f"{k}_outlier:val={metrics[k]:.2f},med={med:.2f},dev={dev_abs:.2f}") re_corr = safe_float(row.get("re_edge_corr")) nir_corr = safe_float(row.get("nir_edge_corr")) low_corr = re_corr < args.clean_min_edge_corr or nir_corr < args.clean_min_edge_corr shift_bad = max_dev > args.clean_max_dev_shift_px * 0.75 if low_corr: if args.clean_reject_low_corr_only_if_shift_bad: if shift_bad: reasons.append(f"low_edge_corr_with_shift:re={re_corr:.3f},nir={nir_corr:.3f}") else: reasons.append(f"low_edge_corr:re={re_corr:.3f},nir={nir_corr:.3f}") # Filtro semântico opcional por presença mínima da classe alvo. # Mantém chão puro sem exigir cana/erva. min_target_pct = float(args.clean_min_target_pct) if min_target_pct > 0: h = int(row.get("H", 0) or 0) w = int(row.get("W", 0) or 0) total_px = max(1, h * w) group_lower = group.lower() if "cana" in group_lower: pct_cana = safe_float(row.get("pixels_cana")) / total_px if pct_cana < min_target_pct: reasons.append(f"cana_too_small:{pct_cana:.5f}") if "erva" in group_lower: pct_erva = safe_float(row.get("pixels_erva")) / total_px if pct_erva < min_target_pct: reasons.append(f"erva_too_small:{pct_erva:.5f}") approved = len(reasons) == 0 extra = { "re_shift_mag": metrics["re_mag"], "nir_shift_mag": metrics["nir_mag"], "max_shift_dev_px": max_dev, "re_edge_corr": re_corr, "nir_edge_corr": nir_corr, } return approved, reasons, extra def copy_sample_to_fixed(row: Dict[str, Any], fixed_root: Path, copy_previews: bool): dataset_root = Path(str(row["dataset_root"])) group = str(row["group"]) real_stem = str(row["real_stem"]) meta_path = Path(str(row["meta_path"])) dst_group = fixed_root / group # Copia meta. safe_copy_file(meta_path, dst_group / "metas" / meta_path.name) # Copia máscara. mask_path = Path(str(row.get("mask_path", ""))) if str(mask_path) and mask_path.exists(): safe_copy_file(mask_path, dst_group / "masks" / mask_path.name) # Copia payloads do meta. meta = load_json(meta_path) # Caso raw_native_multi: vários payloads. saved_payload_paths = meta.get("saved_payload_paths", {}) or {} if saved_payload_paths: for _, fname in saved_payload_paths.items(): src = dataset_root / "bins" / Path(fname).name safe_copy_file(src, dst_group / "bins" / src.name) # Caso payload único. saved_payload_path = meta.get("saved_payload_path") if saved_payload_path: src = dataset_root / "bins" / Path(saved_payload_path).name safe_copy_file(src, dst_group / "bins" / src.name) # Fallback usando payload_path do CSV. payload_path = Path(str(row.get("payload_path", ""))) if str(payload_path) and payload_path.exists(): safe_copy_file(payload_path, dst_group / "bins" / payload_path.name) # Preview opcional. if copy_previews: previews_dir = dataset_root / "previews" if previews_dir.is_dir(): for ext in (".png", ".jpg", ".jpeg", ".webp"): src = previews_dir / f"{real_stem}{ext}" if src.exists(): safe_copy_file(src, dst_group / "previews" / src.name) def build_fixed_dataset_from_audit( sample_rows: List[Dict[str, Any]], dataset_roots: List[Path], class_map: Dict[int, str], args, ): fixed_root = resolve_fixed_out_root(dataset_roots, args) ensure_dir(fixed_root) baselines = build_group_shift_baselines(sample_rows) manifest_rows = [] approved_rows = [] rejected_rows = [] for row in sample_rows: approved, reasons, extra = classify_sample_for_training(row, baselines, class_map, args) out_row = dict(row) out_row.update(extra) out_row["approved_for_training"] = approved out_row["reject_reasons"] = " ; ".join(reasons) manifest_rows.append(out_row) if approved: approved_rows.append(out_row) copy_sample_to_fixed(row, fixed_root, copy_previews=args.clean_copy_previews) else: rejected_rows.append(out_row) if args.save_rejected_previews: save_rejected_preview( row=out_row, fixed_root=fixed_root, args=args, tensor=None, mask=None, class_map=class_map, ) summary = { "schema": "multispec_fixed_dataset_v1", "fixed_root": str(fixed_root), "samples_total": len(sample_rows), "samples_approved": len(approved_rows), "samples_rejected": len(rejected_rows), "approval_pct": 100.0 * len(approved_rows) / max(1, len(sample_rows)), "filters": { "clean_max_abs_shift_px": args.clean_max_abs_shift_px, "clean_max_dev_shift_px": args.clean_max_dev_shift_px, "clean_min_edge_corr": args.clean_min_edge_corr, "clean_min_target_pct": args.clean_min_target_pct, "clean_reject_low_corr_only_if_shift_bad": args.clean_reject_low_corr_only_if_shift_bad, }, "group_shift_baselines": { group: { k: { "median": float(v[0]), "mad": float(v[1]), } for k, v in vals.items() } for group, vals in baselines.items() }, "approved_by_group": {}, "rejected_by_group": {}, "rejected_previews": { "enabled": bool(args.save_rejected_previews), "root": str(resolve_rejected_preview_root(fixed_root)), "source": args.rejected_preview_source, "max_width": args.rejected_preview_max_width, }, } for row in approved_rows: g = str(row.get("group", "unknown")) summary["approved_by_group"][g] = summary["approved_by_group"].get(g, 0) + 1 for row in rejected_rows: g = str(row.get("group", "unknown")) summary["rejected_by_group"][g] = summary["rejected_by_group"].get(g, 0) + 1 write_csv(fixed_root / "fixed_manifest.csv", manifest_rows) write_csv(fixed_root / "fixed_approved.csv", approved_rows) write_csv(fixed_root / "fixed_rejected.csv", rejected_rows) write_json(fixed_root / "fixed_summary.json", summary) print("\n========== FIXED DATASET ==========") print(f"Saída fixed/group : {fixed_root}") print(f"Aprovadas : {len(approved_rows)} / {len(sample_rows)}") print(f"Rejeitadas : {len(rejected_rows)} / {len(sample_rows)}") print(f"Manifest : {fixed_root / 'fixed_manifest.csv'}") print("===================================\n") def audit_dataset(args): np.random.seed(args.seed) dataset_roots = find_dataset_roots(Path(args.input_path)) out_dir = ensure_dir(args.out_dir) visuals_dir = ensure_dir(out_dir / "visuals") class_map = parse_class_map(args.class_map) # Monta uma lista única de amostras: (dataset_root, meta_path) entries: List[Tuple[Path, Path]] = [] for root in dataset_roots: for meta_path in list_meta_files(root): entries.append((root, meta_path)) if args.limit and args.limit > 0: entries = entries[:args.limit] global_acc = RunningFeatureStats(reservoir_size=args.stats_reservoir_size) class_acc: Dict[int, RunningFeatureStats] = { cls: RunningFeatureStats(reservoir_size=args.stats_reservoir_size) for cls in class_map.keys() } # sample_rows continua em RAM porque o diagnóstico final e o modo fixed # precisam de uma linha leve por amostra. Isso representa poucos MB. # As linhas por classe/feature, muito mais numerosas, são gravadas direto. sample_rows = [] warnings = [] sample_csv = IncrementalCsvWriter(out_dir / "audit_samples.csv") by_class_sample_csv = IncrementalCsvWriter(out_dir / "audit_by_sample_class_channel.csv") core_cache: Dict[Tuple[int, int, str, str], Any] = {} print(f"[INFO] dataset_roots={len(dataset_roots)}") for root in dataset_roots: print(f" - {root}") print(f"[INFO] amostras={len(entries)}") print(f"[INFO] classes={class_map}") print(f"[INFO] out_dir={out_dir}") print(f"[INFO] module_params_override={args.module_params or 'AUTO/META'}") print(f"[INFO] stats_reservoir_size={args.stats_reservoir_size} por feature/acumulador") print(f"[INFO] CSV incremental ativo | flush_every={args.flush_every} | gc_every={args.gc_every}") for idx, (dataset_root, meta_path) in enumerate(entries, start=1): # Prefixa com o nome do grupo para evitar colisão de nomes entre subdatasets. group_name = dataset_root.name stem = f"{group_name}__{meta_path.stem}" try: tensor, meta, payload_path, fusion_result = load_multispec_tensor( meta_path, dataset_root, core_cache, cli_module_params=args.module_params or None, ) h, w = tensor.shape[1], tensor.shape[2] mask_path = resolve_mask_path(dataset_root, meta_path) mask = load_mask( mask_path, (h, w), args.ignore_index, fusion_result=fusion_result, ) features = compute_feature_maps(tensor) # Estatísticas globais da amostra. sample_feature_stats = {} for fname, fmap in features.items(): raw01 = fname in CHANNELS sample_feature_stats[fname] = calc_stats(fmap.reshape(-1), raw01=raw01) global_acc.add(fname, fmap.reshape(-1), max_samples=args.max_pixels_per_feature) # Estatísticas por classe da amostra. class_pixel_counts = {} if mask is not None: valid_mask = mask != args.ignore_index for cls_id, cls_name in class_map.items(): cm = (mask == cls_id) & valid_mask n = int(np.sum(cm)) class_pixel_counts[cls_name] = n if n < args.min_class_pixels: continue for fname, fmap in features.items(): vals = fmap[cm] class_acc[cls_id].add(fname, vals, max_samples=args.max_pixels_per_feature) s = calc_stats(vals, raw01=(fname in CHANNELS)) by_class_sample_csv.write({ "sample": stem, "class_id": cls_id, "class_name": cls_name, "feature": fname, **s, }) else: warnings.append({ "sample": stem, "type": "missing_mask", "message": "Mascara nao encontrada; estatistica por classe ignorada.", }) # Alinhamento por bordas. r, g, b, re, nir = [tensor[i] for i in range(5)] rgb_gray = (0.299 * r + 0.587 * g + 0.114 * b).astype(np.float32) alignment = { "RGBgray_vs_RE": edge_agreement(rgb_gray, re), "RGBgray_vs_NIR": edge_agreement(rgb_gray, nir), "RE_vs_NIR": edge_agreement(re, nir), } # Heurísticas de alerta. sample_warn = [] for ch in CHANNELS: st = sample_feature_stats[ch] if st["sat_pct"] > args.warn_sat_pct: sample_warn.append(f"{ch}: saturação alta {st['sat_pct']:.2f}%") if st["dark_pct"] > args.warn_dark_pct: sample_warn.append(f"{ch}: pixels escuros alto {st['dark_pct']:.2f}%") if st["p95_p05"] < args.warn_low_dynamic: sample_warn.append(f"{ch}: baixa dinâmica p95-p05={st['p95_p05']:.4f}") for pair_name, al in alignment.items(): dx = abs(safe_float(al.get("phase_dx"))) dy = abs(safe_float(al.get("phase_dy"))) corr = safe_float(al.get("edge_corr")) if dx > args.warn_shift_px or dy > args.warn_shift_px: sample_warn.append(f"{pair_name}: possível shift residual dx={dx:.2f}, dy={dy:.2f}") if corr < args.warn_edge_corr: sample_warn.append(f"{pair_name}: baixa correlação de borda {corr:.3f}") if sample_warn: warnings.append({ "sample": stem, "type": "sample_warning", "messages": sample_warn, }) row = { "idx": idx, "sample": stem, "group": group_name, "real_stem": meta_path.stem, "dataset_root": str(dataset_root), "meta_path": str(meta_path), "payload_path": str(payload_path), "mask_path": str(mask_path) if mask_path else "", "mask_found": bool(mask_path is not None), "mask_shape": str(mask.shape) if mask is not None else "", "mask_unique": mask_unique_summary(mask), "H": h, "W": w, "warnings_count": len(sample_warn), "warning_text": " ; ".join(sample_warn), "re_edge_corr": alignment["RGBgray_vs_RE"]["edge_corr"], "nir_edge_corr": alignment["RGBgray_vs_NIR"]["edge_corr"], "re_phase_dx": alignment["RGBgray_vs_RE"]["phase_dx"], "re_phase_dy": alignment["RGBgray_vs_RE"]["phase_dy"], "nir_phase_dx": alignment["RGBgray_vs_NIR"]["phase_dx"], "nir_phase_dy": alignment["RGBgray_vs_NIR"]["phase_dy"], } for fname in ALL_FEATURES: st = sample_feature_stats[fname] row[f"{fname}_mean"] = st["mean"] row[f"{fname}_std"] = st["std"] row[f"{fname}_p50"] = st["p50"] row[f"{fname}_p95"] = st["p95"] row[f"{fname}_p95_p05"] = st["p95_p05"] if fname in CHANNELS: row[f"{fname}_dark_pct"] = st["dark_pct"] row[f"{fname}_sat_pct"] = st["sat_pct"] for cls_name, count in class_pixel_counts.items(): row[f"pixels_{cls_name}"] = count sample_rows.append(row) sample_csv.write(row) sample_stats_for_visual = { "alignment": alignment, "features": sample_feature_stats, } should_save_visual = args.save_visuals and ( args.visual_every <= 1 or idx % args.visual_every == 0 or len(sample_warn) > 0 ) if should_save_visual: canvas = make_sample_visual(stem, tensor, mask, class_map, sample_stats_for_visual) cv2.imwrite(str(visuals_dir / f"{idx:05d}_{stem}.png"), canvas) if idx % args.print_every == 0 or idx == len(entries): print(f"[OK] {idx}/{len(entries)} | {stem} | warnings={len(sample_warn)}") if args.flush_every > 0 and idx % args.flush_every == 0: sample_csv.flush() by_class_sample_csv.flush() # Libera arrays grandes imediatamente. O coletor periódico ajuda # OpenCV/numpy a devolver referências temporárias mais cedo. del tensor, features, sample_feature_stats, alignment, rgb_gray if mask is not None: del mask if args.gc_every > 0 and idx % args.gc_every == 0: gc.collect() except Exception as e: msg = str(e) warnings.append({ "sample": stem, "type": "exception", "message": msg, }) print(f"[ERRO] {idx}/{len(entries)} | {stem}: {msg}") if args.stop_on_error: raise sample_csv.flush() by_class_sample_csv.flush() sample_csv.close() by_class_sample_csv.close() sample_csv.ensure_file() by_class_sample_csv.ensure_file() # Resumos finais. global_summary = global_acc.summarize() class_summary = { str(cls_id): { "class_name": class_map[cls_id], "features": acc.summarize(), } for cls_id, acc in class_acc.items() } diagnosis = build_diagnosis(global_summary, class_summary, sample_rows, warnings, args) summary = { "schema": "multispec_dataset_audit_v1", "dataset_roots": [str(r) for r in dataset_roots], "samples_processed": len(sample_rows), "channels": CHANNELS, "derived_features": DERIVED, "class_map": {str(k): v for k, v in class_map.items()}, "global_summary": global_summary, "class_summary": class_summary, "diagnosis": diagnosis, } write_json(out_dir / "audit_summary.json", summary) write_json(out_dir / "audit_warnings.json", warnings) # audit_samples.csv e audit_by_sample_class_channel.csv já foram # persistidos incrementalmente durante o loop. write_class_channel_csv(out_dir / "audit_by_class_channel.csv", class_summary) if args.build_fixed_dataset: build_fixed_dataset_from_audit( sample_rows=sample_rows, dataset_roots=dataset_roots, class_map=class_map, args=args, ) print("\n========== AUDITORIA FINALIZADA ==========") print(f"Amostras processadas : {len(sample_rows)}") print(f"Warnings : {len(warnings)}") print(f"Resumo : {out_dir / 'audit_summary.json'}") print(f"CSV amostras : {out_dir / 'audit_samples.csv'}") print(f"CSV classes/canais : {out_dir / 'audit_by_class_channel.csv'}") print(f"Visuais : {visuals_dir}") print("==========================================\n") def write_csv(path: Path, rows: List[Dict[str, Any]]): if not rows: with open(path, "w", encoding="utf-8", newline="") as f: f.write("") return keys = [] for row in rows: for k in row.keys(): if k not in keys: keys.append(k) with open(path, "w", encoding="utf-8", newline="") as f: writer = csv.DictWriter(f, fieldnames=keys, extrasaction="ignore") writer.writeheader() writer.writerows(rows) def write_class_channel_csv(path: Path, class_summary: Dict[str, Any]): rows = [] for cls_id, item in class_summary.items(): cls_name = item["class_name"] for feature, stats in item["features"].items(): rows.append({ "class_id": cls_id, "class_name": cls_name, "feature": feature, **stats, }) write_csv(path, rows) # ============================================================ # Diagnóstico automático simples # ============================================================ def build_diagnosis( global_summary: Dict[str, Dict[str, float]], class_summary: Dict[str, Any], sample_rows: List[Dict[str, Any]], warnings: List[Dict[str, Any]], args, ) -> Dict[str, Any]: notes = [] risks = [] positives = [] # Sanidade global dos canais. for ch in CHANNELS: st = global_summary.get(ch, {}) sat = safe_float(st.get("sat_pct")) dark = safe_float(st.get("dark_pct")) dyn = safe_float(st.get("p95_p05")) mean = safe_float(st.get("mean")) if sat > args.warn_sat_pct: risks.append(f"{ch}: saturação global alta ({sat:.2f}%).") if dark > args.warn_dark_pct: risks.append(f"{ch}: muitos pixels escuros globalmente ({dark:.2f}%).") if dyn < args.warn_low_dynamic: risks.append(f"{ch}: baixa dinâmica global p95-p05={dyn:.4f}.") if 0.02 < mean < 0.98 and dyn >= args.warn_low_dynamic: positives.append(f"{ch}: média/dinâmica globais parecem utilizáveis (mean={mean:.3f}, p95-p05={dyn:.3f}).") # Alinhamento médio. if sample_rows: re_corr = np.mean([safe_float(r.get("re_edge_corr")) for r in sample_rows]) nir_corr = np.mean([safe_float(r.get("nir_edge_corr")) for r in sample_rows]) re_shift = np.mean([math.hypot(safe_float(r.get("re_phase_dx")), safe_float(r.get("re_phase_dy"))) for r in sample_rows]) nir_shift = np.mean([math.hypot(safe_float(r.get("nir_phase_dx")), safe_float(r.get("nir_phase_dy"))) for r in sample_rows]) notes.append(f"Correlação média de bordas RGB-RE={re_corr:.3f}, RGB-NIR={nir_corr:.3f}.") notes.append(f"Shift médio estimado RGB-RE={re_shift:.2f}px, RGB-NIR={nir_shift:.2f}px.") if re_shift > args.warn_shift_px: risks.append(f"RE: shift médio estimado alto ({re_shift:.2f}px). Verificar homografia/fusão.") if nir_shift > args.warn_shift_px: risks.append(f"NIR: shift médio estimado alto ({nir_shift:.2f}px). Verificar homografia/fusão.") if re_corr < args.warn_edge_corr: risks.append(f"RE: correlação média de borda baixa ({re_corr:.3f}). Pode indicar desalinhamento ou textura espectral muito diferente.") if nir_corr < args.warn_edge_corr: risks.append(f"NIR: correlação média de borda baixa ({nir_corr:.3f}). Pode indicar desalinhamento ou textura espectral muito diferente.") # Separabilidade simples por classes. separability = {} class_items = list(class_summary.items()) for feature in ALL_FEATURES: vals = [] for cls_id, item in class_items: st = item["features"].get(feature, {}) vals.append((item["class_name"], safe_float(st.get("mean")), safe_float(st.get("std")))) sep_rows = [] for i in range(len(vals)): for j in range(i + 1, len(vals)): a_name, a_mean, a_std = vals[i] b_name, b_mean, b_std = vals[j] pooled = math.sqrt((a_std * a_std + b_std * b_std) / 2.0) + EPS d = abs(a_mean - b_mean) / pooled sep_rows.append({ "pair": f"{a_name}_vs_{b_name}", "effect_size_d": d, "mean_a": a_mean, "mean_b": b_mean, }) separability[feature] = sep_rows # Destaca features com maior separação média. ranking = [] for feature, rows in separability.items(): if rows: avg_d = float(np.mean([r["effect_size_d"] for r in rows])) ranking.append((feature, avg_d)) ranking.sort(key=lambda x: x[1], reverse=True) notes.append("Ranking simples de separabilidade média por feature: " + ", ".join([f"{f}={d:.2f}" for f, d in ranking[:8]])) for f, d in ranking[:5]: if d > 0.5: positives.append(f"{f}: mostra separabilidade média interessante entre classes (d≈{d:.2f}).") if warnings: risks.append(f"Foram gerados {len(warnings)} avisos/exceções. Ver audit_warnings.json.") return { "positives": positives, "risks": risks, "notes": notes, "feature_separability": separability, "feature_separability_ranking": [{"feature": f, "avg_effect_size_d": d} for f, d in ranking], } # ============================================================ # CLI # ============================================================ def main(): parser = argparse.ArgumentParser( description="Auditoria final do tensor multiespectral [R,G,B,RE,NIR] pronto para treinamento." ) parser.add_argument("--input_path", required=True, help="Caminho para dataset_root, metas, bins, masks ou arquivo dentro do dataset.") parser.add_argument("--out_dir", default="audit_multispec_out", help="Pasta de saída da auditoria.") parser.add_argument( "--module-params", default="", help=( "Override AUTORITATIVO do module_params para reconstruir RAW_BRUTO. " "Ex.: calibration/mp_ov9782/module_params.json. " "Se informado e não existir, a auditoria falha sem usar o caminho salvo no meta." ), ) parser.add_argument("--class-map", default="0:chao,1:cana,2:erva", help="Mapa de classes. Ex: 0:chao,1:cana,2:erva") parser.add_argument("--ignore-index", type=int, default=255, help="Valor ignorado na máscara.") parser.add_argument("--limit", type=int, default=0, help="Limita número de amostras para teste rápido. 0 = todas.") parser.add_argument("--save-visuals", action="store_true", help="Salva painéis visuais por amostra.") parser.add_argument("--visual-every", type=int, default=10, help="Salva visual a cada N amostras. Amostras com warning sempre são salvas.") parser.add_argument("--print-every", type=int, default=10, help="Mostra progresso a cada N amostras.") parser.add_argument("--min-class-pixels", type=int, default=50, help="Mínimo de pixels por classe para estatística por amostra.") parser.add_argument("--max-pixels-per-feature", type=int, default=25000, help="Amostragem máxima de pixels por feature/amostra para acumuladores.") parser.add_argument( "--stats-reservoir-size", type=int, default=200000, help=( "Máximo TOTAL de valores float32 mantidos em RAM por feature e por acumulador " "para estimar percentis. Média/std/min/max continuam exatos. Default=200000." ), ) parser.add_argument( "--flush-every", type=int, default=25, help="Faz flush dos CSVs incrementais a cada N amostras. 0 desativa flush periódico.", ) parser.add_argument( "--gc-every", type=int, default=100, help="Executa gc.collect() a cada N amostras para reduzir picos de memória. 0 desativa.", ) parser.add_argument("--warn-sat-pct", type=float, default=1.0, help="Alerta se canal tiver saturação acima deste percentual.") parser.add_argument("--warn-dark-pct", type=float, default=35.0, help="Alerta se canal tiver pixels <=0.01 acima deste percentual.") parser.add_argument("--warn-low-dynamic", type=float, default=0.03, help="Alerta se p95-p05 do canal for menor que isso.") parser.add_argument("--warn-shift-px", type=float, default=3.0, help="Alerta se shift estimado por phase correlation passar disso.") parser.add_argument("--warn-edge-corr", type=float, default=0.08, help="Alerta se correlação de borda for menor que isso.") parser.add_argument("--seed", type=int, default=42) parser.add_argument("--stop-on-error", action="store_true") parser.add_argument("--build-fixed-dataset", action="store_true", help="Cria uma cópia filtrada do dataset em fixed/group, mantendo apenas amostras aprovadas para treino.") parser.add_argument("--fixed-out-root", default="", help="Raiz de saída do dataset filtrado. Se vazio, usa dataset_root/../../fixed/group.") parser.add_argument("--clean-max-abs-shift-px", type=float, default=22.0, help="Reprova amostras com shift absoluto extremo em RE ou NIR.") parser.add_argument("--clean-max-dev-shift-px", type=float, default=7.0, help="Reprova amostras cujo shift foge muito da mediana robusta do grupo.") parser.add_argument("--clean-min-edge-corr", type=float, default=0.045, help="Correlação mínima de borda para aprovar. Valor baixo porque canais espectrais naturalmente diferem do RGB.") parser.add_argument("--clean-reject-low-corr-only-if-shift-bad", action="store_true", help="Só reprova baixa correlação se também houver desvio geométrico alto.") parser.add_argument("--clean-min-target-pct", type=float, default=0.0025, help="Percentual mínimo da classe alvo em grupos com cana/erva. 0.0025 = 0.25%%. Use 0 para desativar.") parser.add_argument("--clean-copy-previews", action="store_true", help="Copia previews quando existirem.") parser.add_argument("--save-rejected-previews", action="store_true", help="Salva PNGs leves das amostras rejeitadas em fixed/rejected_previews para inspeção visual.") parser.add_argument("--rejected-preview-source", default="auto", choices=["auto", "preview", "audit_panel", "rgb_tensor"], help="Fonte do preview dos rejeitados: preview original, painel audit, RGB reconstruído ou automático.") parser.add_argument("--rejected-preview-max-width", type=int, default=640, help="Largura máxima dos previews rejeitados salvos.") args = parser.parse_args() audit_dataset(args) if __name__ == "__main__": main()