"""Root-cause analysis for ref_04 thermal absent false positives. Measurement only. Reads archived data from the modality-limited routing v2 run, extracts TP/FP populations for ref_04, generates composites, and writes a small report bundle under _logs/ref04_fp_analysis/. """ from __future__ import annotations import json from pathlib import Path from statistics import mean, median from typing import Any import cv2 import numpy as np from src.evaluation.task3_manifest_eval import load_task3_manifest ARCHIVE_DIR = Path("_logs/reports_generated/task3_manifest/2026-04-20_per_reference_routing_modality_v2") DEBUG_DIR = Path("_logs/debug/task3_per_reference_routing_modality_v2") OUTPUT_DIR = Path("_logs/ref04_fp_analysis") REFERENCE_PATH = Path("data/references/2026_baseline/ref_04.jpg") MANIFEST_PATH = Path("data/task3_eval_manifest.json") def _load_json(path: Path) -> dict[str, Any]: return json.loads(path.read_text(encoding="utf-8")) def _load_jsonl(path: Path) -> list[dict[str, Any]]: if not path.exists(): return [] return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()] def _scenario_map() -> dict[str, dict[str, Any]]: manifest = load_task3_manifest(MANIFEST_PATH) return {str(item["id"]): item for item in manifest["scenarios"]} def _accepted_frames() -> tuple[set[int], set[int]]: summary = _load_json(ARCHIVE_DIR / "per_reference_routing_summary.json") tp_frames = set(int(value) for value in summary["thermal_cross_sensor_proxy"]["accepted_frames_by_ref"]["ref_04"]) fp_frames = set(int(value) for value in summary["thermal_absent_target_proxy_2025"]["accepted_frames_by_ref"]["ref_04"]) return tp_frames, fp_frames def _pick_population(rows: list[dict[str, Any]], accepted_frames: set[int]) -> list[dict[str, Any]]: grouped: dict[int, list[dict[str, Any]]] = {} for row in rows: if row.get("object_id") != "ref_04": continue frame_idx = int(row["frame_idx"]) if frame_idx not in accepted_frames: continue grouped.setdefault(frame_idx, []).append(row) selected: list[dict[str, Any]] = [] for frame_idx in sorted(grouped): best = max(grouped[frame_idx], key=lambda item: float(item.get("score", 0.0))) selected.append(best) selected.sort(key=lambda item: float(item.get("score", 0.0)), reverse=True) return selected def _range(values: list[float]) -> tuple[float, float]: return (min(values), max(values)) if values else (0.0, 0.0) def _format_range(values: list[float]) -> str: if not values: return "-" low, high = _range(values) return f"{low:.4f}-{high:.4f}" def _overlap(tp_values: list[float], fp_values: list[float]) -> str: if not tp_values or not fp_values: return "-" tp_low, tp_high = _range(tp_values) fp_low, fp_high = _range(fp_values) low = max(tp_low, fp_low) high = min(tp_high, fp_high) if high < low: return "none" return f"{low:.4f}-{high:.4f}" def _ascii_scatter(tp: list[dict[str, Any]], fp: list[dict[str, Any]], *, width: int = 28, height: int = 12) -> str: grid = [["." for _ in range(width)] for _ in range(height)] def _plot(points: list[dict[str, Any]], marker: str) -> None: for item in points: x = max(0.0, min(float(item["inlier_ratio"]), 1.0)) y = max(0.0, min(float(item["score"]), 1.0)) col = min(int(round(x * (width - 1))), width - 1) row = min(int(round((1.0 - y) * (height - 1))), height - 1) current = grid[row][col] if current != "." and current != marker: grid[row][col] = "*" else: grid[row][col] = marker _plot(tp, "T") _plot(fp, "F") lines = ["score^"] for row in grid: lines.append("".join(row)) lines.append("+" + "-" * width + "> inlier_ratio") lines.append("Legend: T=true positive, F=false positive, *=overlap") return "\n".join(lines) def _extract_frame(video_path: Path, frame_idx: int) -> np.ndarray: cap = cv2.VideoCapture(str(video_path)) try: cap.set(cv2.CAP_PROP_POS_FRAMES, frame_idx) ok, frame = cap.read() if not ok or frame is None: raise RuntimeError(f"cannot read frame {frame_idx} from {video_path}") return frame finally: cap.release() def _ensure_bgr(image: np.ndarray) -> np.ndarray: if image.ndim == 2: return cv2.cvtColor(image, cv2.COLOR_GRAY2BGR) return image def _resize_fit(image: np.ndarray, max_side: int = 400) -> np.ndarray: image = _ensure_bgr(image) height, width = image.shape[:2] scale = min(max_side / max(height, width), 1.0) if scale >= 1.0: return image return cv2.resize(image, (int(round(width * scale)), int(round(height * scale))), interpolation=cv2.INTER_AREA) def _pad_to_height(image: np.ndarray, target_height: int) -> np.ndarray: if image.shape[0] >= target_height: return image bottom = target_height - image.shape[0] return cv2.copyMakeBorder(image, 0, bottom, 0, 0, cv2.BORDER_CONSTANT, value=(0, 0, 0)) def _annotate(image: np.ndarray, lines: list[str]) -> np.ndarray: canvas = image.copy() y = 22 for line in lines: cv2.putText(canvas, line, (8, y), cv2.FONT_HERSHEY_SIMPLEX, 0.55, (0, 255, 0), 2, cv2.LINE_AA) y += 22 return canvas def _write_composite(reference_bgr: np.ndarray, crop_bgr: np.ndarray, *, out_path: Path, label: str, frame_idx: int, score: float, inlier_ratio: float) -> None: ref_panel = _resize_fit(reference_bgr) crop_panel = _resize_fit(crop_bgr) target_height = max(ref_panel.shape[0], crop_panel.shape[0]) ref_panel = _pad_to_height(ref_panel, target_height) crop_panel = _pad_to_height(crop_panel, target_height) ref_panel = _annotate(ref_panel, ["REFERENCE", f"{REFERENCE_PATH.name}", f"{reference_bgr.shape[1]}x{reference_bgr.shape[0]}"]) crop_panel = _annotate(crop_panel, [label, f"frame={frame_idx}", f"score={score:.4f} inlier={inlier_ratio:.4f}"]) composite = cv2.hconcat([ref_panel, crop_panel]) out_path.parent.mkdir(parents=True, exist_ok=True) cv2.imwrite(str(out_path), composite) def _reference_inspection(reference_gray: np.ndarray) -> dict[str, Any]: blurred = cv2.GaussianBlur(reference_gray, (5, 5), 0) _threshold, mask = cv2.threshold(blurred, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU) components = cv2.connectedComponentsWithStats(mask, connectivity=8) stats = components[2] image_area = reference_gray.shape[0] * reference_gray.shape[1] largest_area = 0 largest_bbox = [0, 0, 0, 0] for label_idx in range(1, stats.shape[0]): x, y, w, h, area = [int(value) for value in stats[label_idx]] if area > largest_area: largest_area = area largest_bbox = [x, y, w, h] area_ratio = largest_area / max(image_area, 1) return { "image_width": int(reference_gray.shape[1]), "image_height": int(reference_gray.shape[0]), "largest_bright_component_bbox_xywh": largest_bbox, "largest_bright_component_area_ratio": round(float(area_ratio), 4), } def main() -> None: OUTPUT_DIR.mkdir(parents=True, exist_ok=True) scenarios = _scenario_map() tp_frames, fp_frames = _accepted_frames() tp_rows = _pick_population(_load_jsonl(DEBUG_DIR / "thermal_cross_sensor_proxy" / "post_gate.jsonl"), tp_frames) fp_rows = _pick_population(_load_jsonl(DEBUG_DIR / "thermal_absent_target_proxy_2025" / "post_gate.jsonl"), fp_frames) reference_bgr = cv2.imread(str(REFERENCE_PATH)) if reference_bgr is None: raise RuntimeError(f"cannot load reference image: {REFERENCE_PATH}") tp_video = Path(scenarios["thermal_cross_sensor_proxy"]["video"]) fp_video = Path(scenarios["thermal_absent_target_proxy_2025"]["video"]) visual_dir = OUTPUT_DIR / "composites" for index, row in enumerate(tp_rows, start=1): frame = _extract_frame(tp_video, int(row["frame_idx"])) x1, y1, x2, y2 = [int(value) for value in row["bbox"]] crop = frame[y1:y2, x1:x2] _write_composite( reference_bgr, crop, out_path=visual_dir / f"tp_{index:02d}_frame_{int(row['frame_idx']):04d}.png", label="TP", frame_idx=int(row["frame_idx"]), score=float(row["score"]), inlier_ratio=float(row["inlier_ratio"]), ) for index, row in enumerate(fp_rows, start=1): frame = _extract_frame(fp_video, int(row["frame_idx"])) x1, y1, x2, y2 = [int(value) for value in row["bbox"]] crop = frame[y1:y2, x1:x2] _write_composite( reference_bgr, crop, out_path=visual_dir / f"fp_{index:02d}_frame_{int(row['frame_idx']):04d}.png", label="FP", frame_idx=int(row["frame_idx"]), score=float(row["score"]), inlier_ratio=float(row["inlier_ratio"]), ) # contact sheet ordered_paths = sorted(visual_dir.glob("*.png")) thumbs = [] for path in ordered_paths: image = cv2.imread(str(path)) thumbs.append(_resize_fit(image, max_side=260)) rows: list[np.ndarray] = [] for start in range(0, len(thumbs), 2): pair = thumbs[start : start + 2] max_h = max(item.shape[0] for item in pair) padded = [_pad_to_height(item, max_h) for item in pair] if len(padded) == 1: padded.append(np.zeros_like(padded[0])) rows.append(cv2.hconcat(padded)) sheet = rows[0] for row in rows[1:]: max_w = max(sheet.shape[1], row.shape[1]) if sheet.shape[1] < max_w: sheet = cv2.copyMakeBorder(sheet, 0, 0, 0, max_w - sheet.shape[1], cv2.BORDER_CONSTANT, value=(0, 0, 0)) if row.shape[1] < max_w: row = cv2.copyMakeBorder(row, 0, 0, 0, max_w - row.shape[1], cv2.BORDER_CONSTANT, value=(0, 0, 0)) sheet = cv2.vconcat([sheet, row]) cv2.imwrite(str(OUTPUT_DIR / "contact_sheet.png"), sheet) tp_conf = [float(item["yoloe_confidence"]) for item in tp_rows] fp_conf = [float(item["yoloe_confidence"]) for item in fp_rows] tp_matches = [float(item["match_count"]) for item in tp_rows] fp_matches = [float(item["match_count"]) for item in fp_rows] tp_inlier = [float(item["inlier_ratio"]) for item in tp_rows] fp_inlier = [float(item["inlier_ratio"]) for item in fp_rows] tp_score = [float(item["score"]) for item in tp_rows] fp_score = [float(item["score"]) for item in fp_rows] feature_summary = { "yoloe_confidence": {"tp_range": _format_range(tp_conf), "fp_range": _format_range(fp_conf), "overlap": _overlap(tp_conf, fp_conf)}, "match_count": {"tp_range": _format_range(tp_matches), "fp_range": _format_range(fp_matches), "overlap": _overlap(tp_matches, fp_matches)}, "inlier_ratio": {"tp_range": _format_range(tp_inlier), "fp_range": _format_range(fp_inlier), "overlap": _overlap(tp_inlier, fp_inlier)}, "score_3term": {"tp_range": _format_range(tp_score), "fp_range": _format_range(fp_score), "overlap": _overlap(tp_score, fp_score)}, } reference_notes = _reference_inspection(cv2.cvtColor(reference_bgr, cv2.COLOR_BGR2GRAY)) scatter = _ascii_scatter(tp_rows, fp_rows) payload = { "tp_population": tp_rows, "fp_population": fp_rows, "feature_summary": feature_summary, "ascii_scatter": scatter, "reference_inspection": reference_notes, "composite_dir": str(visual_dir), } (OUTPUT_DIR / "analysis.json").write_text(json.dumps(payload, indent=2), encoding="utf-8") lines = [ "# ref_04 TP vs FP Analysis", "", "## Feature Summary", f"- YOLOE confidence: TP {feature_summary['yoloe_confidence']['tp_range']}, FP {feature_summary['yoloe_confidence']['fp_range']}, overlap {feature_summary['yoloe_confidence']['overlap']}", f"- match_count: TP {feature_summary['match_count']['tp_range']}, FP {feature_summary['match_count']['fp_range']}, overlap {feature_summary['match_count']['overlap']}", f"- inlier_ratio: TP {feature_summary['inlier_ratio']['tp_range']}, FP {feature_summary['inlier_ratio']['fp_range']}, overlap {feature_summary['inlier_ratio']['overlap']}", f"- score_3term: TP {feature_summary['score_3term']['tp_range']}, FP {feature_summary['score_3term']['fp_range']}, overlap {feature_summary['score_3term']['overlap']}", "", "## ASCII Scatter", "```text", scatter, "```", "", "## Reference Inspection", f"- Largest bright component bbox xywh: {reference_notes['largest_bright_component_bbox_xywh']}", f"- Largest bright component area ratio: {reference_notes['largest_bright_component_area_ratio']}", "", "## TP Population", "| frame_idx | conf | match_count | inlier_count | inlier_ratio | score | bbox |", "| --- | --- | --- | --- | --- | --- | --- |", ] for row in tp_rows: lines.append( f"| {row['frame_idx']} | {float(row['yoloe_confidence']):.6f} | {int(row['match_count'])} | {int(row['inlier_count'])} | {float(row['inlier_ratio']):.6f} | {float(row['score']):.6f} | {row['bbox']} |" ) lines.extend( [ "", "## FP Population", "| frame_idx | conf | match_count | inlier_count | inlier_ratio | score | bbox |", "| --- | --- | --- | --- | --- | --- | --- |", ] ) for row in fp_rows: lines.append( f"| {row['frame_idx']} | {float(row['yoloe_confidence']):.6f} | {int(row['match_count'])} | {int(row['inlier_count'])} | {float(row['inlier_ratio']):.6f} | {float(row['score']):.6f} | {row['bbox']} |" ) (OUTPUT_DIR / "analysis.md").write_text("\n".join(lines) + "\n", encoding="utf-8") print(json.dumps(payload, indent=2)) if __name__ == "__main__": main()