from __future__ import annotations import argparse import csv import html as html_lib import json import math import os import random import sys import time from collections import Counter, defaultdict from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timedelta from pathlib import Path from typing import Any, Iterable, Optional import cv2 import numpy as np import torch import yaml FILE = Path(__file__).resolve() ROOT = FILE.parents[2] if str(ROOT) not in sys.path: sys.path.append(str(ROOT)) from ultralytics.data.ground3d_augment import normalize_roi_depth, parse_ground_3d_label_file, read_calib_from_path, remap_labels_to_roi from ultralytics.engine.validator import BaseValidator from ultralytics.utils.plotting_3d import ( EDGE_YAW_VALID_VISIBILITY_SCORE_THRESH, FACE_COLORS, classify_edge_yaw_prediction_bucket, collect_face_bottom_edges, create_bev_image, decode_3d_target, decode_multi_visible_face_yaw_from_gt, draw_3d_box, extract_3d_attrs_from_gt, extract_3d_attrs_from_prediction, face_center_from_corners, get_gt_cut_state, get_pred_cut_state, is_gt_cut_object, project_3d_to_2d_with_distortion, rebuild_box_corners_for_visualization, select_gt_visible_faces, ) from ultralytics.utils.metrics import ConfusionMatrix, ap_per_class, smooth from tools.pdcl_inference.two_roi_inference import ( filter_prediction_outputs, iter_batches, prepare_roi_image, decode_prepared_roi_predictions, run_model_for_prepared_roi_batch, ) from tools.pdcl_inference.run_batch_two_roi_infer import ( add_two_roi_inference_args, build_two_roi_inference_context_from_args, populate_two_roi_inference_args, ) DEFAULT_OUTPUT_ROOT = FILE.parent / "validation_analysis" / "report_{}".format(datetime.now().strftime("%Y%m%d_%H%M%S")) FACE_NAMES = {0: "front", 1: "rear", 2: "left", 3: "right"} FACE_SELECTION_LABEL_ORDER = ("cut_in", "cut_out", "front", "rear", "left", "right") FAKE_CLASS_LABEL_ORDER = ("non_fake", "car_fake", "bicyclist_fake", "pedestrian_fake", "car", "bicycle", "pedestrian") OCCLUSION_BINARY_LABEL_ORDER = ("visible", "occluded") HORIZONTAL_LATERAL_BIN_M = 5.0 HORIZONTAL_LATERAL_RANGE_M = 30.0 VERTICAL_DEPTH_BIN_M = 5.0 YAW_DEPTH_BIN_M = 10.0 YAW_HEADING_BIN_DEG = 10.0 ERROR_DISTANCE_BIN_M = 0.5 ERROR_YAW_BIN_DEG = 1.0 DEFAULT_ERROR_BIN_BADCASES = 50 DEFAULT_ERROR_BIN_SAMPLES_PER_BIN = 10 ROI1_MIN_Z_ERROR_DEPTH_M = 10.0 DEFAULT_YAW_COMPARE_MAX_LATERAL_DIST_M = HORIZONTAL_LATERAL_RANGE_M DEFAULT_YAW_COMPARE_MAX_LONGITUDINAL_DIST_M = 50.0 FACE_VISIBILITY_BUCKET_ORDER = ("front_rear_only", "side only", "two-face") FOCUSED_CONFUSION_MAX_ABS_LATERAL_M = 5.0 FOCUSED_CONFUSION_MAX_ABS_LONGITUDINAL_M = 80.0 FOCUSED_CONFUSION_REQUIRED_DIFFICULTY = 0 ROI0_FOCUSED_CONFUSION_MAX_ABS_LONGITUDINAL_M = 30.0 LARGE_VEHICLE_CLASS_IDS = frozenset({5, 6, 7}) LARGE_VEHICLE_CLASS_SCOPE_TEXT = "5=bus, 6=truck/tanker/large_truck/construction_vehicle, 7=special_vehicle" PORTRAIT_HEADING_BIN_DEG = 10.0 PORTRAIT_LATERAL_BIN_M = 5.0 PORTRAIT_LONGITUDINAL_BIN_M = 10.0 DEFAULT_DATA_PORTRAIT_WORKERS = max(1, os.cpu_count() or 1) DATA_PORTRAIT_CHUNK_SIZE = 256 MIN_CONFIDENCE_FOR_2D_THRESHOLD_SEARCH = 0.001 BADCASE_FIELDS = [ "sample_index", "roi", "frame_name", "image_path", "label_path", "cls_id", "cls_name", "gt_index", "pred_index", "confidence", "match_iou", "bbox_diag_px", "bbox_diag_bin", "distance_bin", "is_cut_object", "position_eligible", "position_error_basis", "yaw_compare_eligible", "visible_face_count", "visible_faces", "has_side_face_visible", "yaw_abs_deg", "direct_visible_yaw_abs_deg", "edge_visible_yaw_abs_deg", "direct_minus_edge_visible_yaw_abs_deg", "direct_length_abs_err_m", "edge_length_abs_err_m", "direct_minus_edge_length_abs_err_m", "gt_lateral_abs_m", "x_abs_m", "y_abs_m", "z_abs_m", "center_error_m", "position_gt_x_m", "position_pred_x_m", "position_gt_y_m", "position_pred_y_m", "position_gt_z_m", "position_pred_z_m", "gt_x_m", "pred_x_m", "gt_y_m", "pred_y_m", "gt_z_m", "pred_z_m", "gt_depth_m", "pred_depth_m", "gt_yaw_deg", "pred_yaw_deg", "gt_visible_yaw_deg", "pred_edge_yaw_deg", "gt_face_selection_label", "pred_face_selection_label", "face_selection_correct", "gt_bbox_xyxy", "pred_bbox_xyxy", ] BADCASE_2D_FIELDS = [ "sample_index", "roi", "frame_name", "image_path", "label_path", "kind", "cls_id", "cls_name", "gt_index", "pred_index", "confidence", "max_iou_any", "max_iou_same_class", "bbox_xyxy", ] FAKE_CLASS_BADCASE_FIELDS = [ "sample_index", "roi", "frame_name", "image_path", "label_path", "kind", "gt_index", "pred_index", "gt_cls_id", "gt_cls_name", "pred_cls_id", "pred_cls_name", "gt_fake_class_label", "pred_fake_class_label", "gt_occlusion_label", "pred_occlusion_label", "match_iou", "confidence", "gt_bbox_xyxy", "pred_bbox_xyxy", ] FACE_SELECTION_BADCASE_FIELDS = [ "sample_index", "roi", "frame_name", "image_path", "label_path", "gt_index", "pred_index", "gt_cls_id", "gt_cls_name", "pred_cls_id", "pred_cls_name", "gt_face_selection_label", "pred_face_selection_label", "match_iou", "confidence", "gt_bbox_xyxy", "pred_bbox_xyxy", ] def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Run validation-set error analysis for the two-ROI Detect3D models and save bad-case visualizations." ) parser.add_argument( "--data", type=str, default=str(ROOT / "ultralytics" / "cfg" / "datasets" / "mono3d_ground.yaml"), help="Dataset YAML path used to resolve the validation split.", ) parser.add_argument("--split", type=str, default="val", choices=("train", "val", "test"), help="Dataset split to analyze.") parser.add_argument( "--output-root", type=str, default=str(DEFAULT_OUTPUT_ROOT), help="Directory where summaries, CSVs, and bad-case visualizations will be written.", ) parser.add_argument( "--analyze-rois", nargs="+", default=["roi0", "roi1"], choices=("roi0", "roi1"), help="ROI models to analyze.", ) parser.add_argument("--max-samples", type=int, default=0, help="Optional cap on analyzed samples. 0 means full split.") parser.add_argument( "--sample-selection", type=str, default="head", choices=("head", "random"), help="How to choose the first --max-samples entries when a cap is set.", ) parser.add_argument("--sample-random-seed", type=int, default=20260327, help="Random seed used when --sample-selection=random.") parser.add_argument("--log-every", type=int, default=100, help="Progress log interval in samples.") parser.add_argument("--topk-badcases", type=int, default=100, help="How many worst cases to save per category and ROI.") parser.add_argument("--per-class-badcases", type=int, default=50, help="How many bad cases to save per class/category and ROI.") parser.add_argument( "--error-bin-badcases", type=int, default=DEFAULT_ERROR_BIN_BADCASES, help="How many object-level candidates to keep per error bin before frame-level aggregation for report sections.", ) parser.add_argument( "--error-bin-samples-per-bin", type=int, default=DEFAULT_ERROR_BIN_SAMPLES_PER_BIN, help="How many frame-level sample images to export per error bin in the report sections.", ) parser.add_argument("--badcase-random-seed", type=int, default=20260326, help="Random seed for bad-case visualization sampling.") parser.add_argument("--yaw-bad-threshold-deg", type=float, default=5.0, help="Threshold for yaw bad-case CSV export.") parser.add_argument( "--yaw-compare-max-lateral-dist", type=float, default=DEFAULT_YAW_COMPARE_MAX_LATERAL_DIST_M, help="Signed lateral range limit for yaw comparison bins; compares paired samples within [-limit, limit) meters.", ) parser.add_argument("--horizontal-bad-threshold-m", type=float, default=0.5, help="Threshold for bad horizontal X error.") parser.add_argument("--vertical-bad-threshold-m", type=float, default=0.5, help="Threshold for bad vertical Y error.") parser.add_argument("--face-visibility-score-thresh", type=float, default=0.05, help="Visible-face score threshold.") parser.add_argument("--torch-threads", type=int, default=0, help="Optional torch CPU thread count override.") parser.add_argument("--skip-data-portrait", action="store_true", help="Skip dataset portrait generation.") parser.add_argument( "--data-portrait-split", type=str, default="train", choices=("train", "val", "test"), help="Dataset split used for the frame/object portrait statistics.", ) parser.add_argument( "--data-portrait-max-samples", type=int, default=0, help="Optional cap on portrait samples. 0 means the full portrait split.", ) parser.add_argument( "--data-portrait-sample-selection", type=str, default="head", choices=("head", "random"), help="How to choose the first --data-portrait-max-samples entries when a cap is set.", ) parser.add_argument( "--data-portrait-random-seed", type=int, default=20260330, help="Random seed used when --data-portrait-sample-selection=random.", ) parser.add_argument( "--data-portrait-workers", type=int, default=0, help="Worker threads used for portrait GT/calibration scanning. 0 uses all detected CPU threads.", ) add_two_roi_inference_args(parser, include_output_dir=False) return parser.parse_args() def load_yaml(path: str | Path) -> dict[str, Any]: with Path(path).open("r", encoding="utf-8") as file: return yaml.safe_load(file) or {} def resolve_data_path(data_yaml: Path, dataset_root: Optional[str], value: Any) -> Path: if value is None: raise ValueError(f"Missing split path in {data_yaml}") path = Path(str(value)) if path.is_absolute(): return path.resolve() candidates = [] if dataset_root: candidates.append((Path(dataset_root) / path).resolve()) candidates.append((data_yaml.parent / path).resolve()) for candidate in candidates: if candidate.exists(): return candidate return candidates[0] if candidates else path.resolve() def load_split_entries( split_file: Path, max_samples: int = 0, sample_selection: str = "head", sample_random_seed: int = 20260327, ) -> list[tuple[str, str]]: entries: list[tuple[str, str]] = [] with split_file.open("r", encoding="utf-8") as file: for line in file: entry = line.strip() if not entry or entry.lstrip().startswith("#"): continue if Path(entry).suffix.lower() != ".txt": raise ValueError(f"Ground3D split entries must point to label .txt files, but got: {entry}") entries.append((str(split_file.parent.resolve()), entry)) if not entries: raise FileNotFoundError(f"No usable entries found in {split_file}") if max_samples > 0 and len(entries) > max_samples: if sample_selection == "random": rng = random.Random(int(sample_random_seed)) indices = sorted(rng.sample(range(len(entries)), int(max_samples))) entries = [entries[idx] for idx in indices] else: entries = entries[:max_samples] return entries _IMAGE_EXTENSIONS = (".png", ".jpg", ".jpeg") def label_rel_to_image_rel(rel_label: Path) -> Path: parts = list(rel_label.parts) if "labels" in parts: parts[len(parts) - 1 - parts[::-1].index("labels")] = "images" return Path(*parts).with_suffix(".png") def resolve_image_path(image_path: Path) -> Path: if image_path.is_file(): return image_path stem = image_path.stem parent = image_path.parent for ext in _IMAGE_EXTENSIONS: if ext == image_path.suffix.lower(): continue candidate = parent / (stem + ext) if candidate.is_file(): return candidate return image_path def label_rel_to_calib_rel(rel_label: Path) -> Path: parts = list(rel_label.parts) if "labels" not in parts: raise ValueError(f"Expected a Ground3D label path containing `labels`, but got: {rel_label}") parts[len(parts) - 1 - parts[::-1].index("labels")] = "calib" return Path(*parts).with_suffix(".json") def entry_to_label_file(entry: tuple[str, str]) -> Path: list_root, rel_label = entry return (Path(list_root) / rel_label).resolve() def label_path_to_calib_file(label_path: Path) -> Path: return label_rel_to_calib_rel(label_path).resolve() def entry_to_image_file(entry: tuple[str, str], image_root: Path) -> Path: rel_image = label_rel_to_image_rel(Path(entry[1])) return resolve_image_path((image_root / rel_image).resolve()) def read_raw_calib_from_label(image_path: Path, label_path: Path) -> Optional[dict[str, Any]]: calib_path = label_path_to_calib_file(label_path) return read_calib_from_path(str(image_path), extra_calib_candidates=[calib_path]) def infer_class_name_map(class_map: dict[str, int], preferred_names: Optional[dict[int, str]] = None) -> dict[int, str]: names = dict(preferred_names or {}) for raw_name, cls_id in class_map.items(): names.setdefault(int(cls_id), str(raw_name)) return {int(cls_id): str(names[cls_id]) for cls_id in sorted(names)} def read_mapped_label_class_names(label_file: Path, class_map: dict[str, int]) -> list[str]: if not label_file.is_file(): return [] names: list[str] = [] with label_file.open("r", encoding="utf-8") as file: for line in file: parts = line.split() if not parts: continue cls_name = str(parts[0]) if cls_name not in class_map: continue names.append(cls_name) return names def parse_label_frame_metadata(label_path: Path) -> dict[str, Optional[str]]: stem = label_path.stem parts = stem.split("_") timestamp_index = next((index for index, token in enumerate(parts) if len(token) == 14 and token.isdigit()), None) date_index = next((index for index, token in enumerate(parts) if len(token) == 8 and token.isdigit() and token.startswith("20")), None) vehicle_alias_from_path = next((part for part in label_path.parts[:-1] if str(part).startswith("G1M3_")), None) if vehicle_alias_from_path: vehicle_alias = str(vehicle_alias_from_path) elif timestamp_index is not None and timestamp_index > 0: vehicle_alias = "_".join(parts[:timestamp_index]) elif date_index is not None and date_index > 0: vehicle_alias = "_".join(parts[:date_index]) else: vehicle_alias = "unknown" timestamp_text = parts[timestamp_index] if timestamp_index is not None else None frame_dt = None if timestamp_text: try: frame_dt = datetime.strptime(timestamp_text, "%Y%m%d%H%M%S") except ValueError: frame_dt = None if frame_dt is None and date_index is not None: date_text = parts[date_index] try: frame_dt = datetime.strptime(date_text, "%Y%m%d") except ValueError: frame_dt = None if frame_dt is None: fallback_day = next((part for part in label_path.parts if len(part) == 8 and part.isdigit()), None) if fallback_day: try: frame_dt = datetime.strptime(fallback_day, "%Y%m%d") except ValueError: frame_dt = None return { "frame_name": stem, "vehicle_alias": vehicle_alias or "unknown", "timestamp_text": timestamp_text, "day": frame_dt.strftime("%Y-%m-%d") if frame_dt else None, "hour": frame_dt.strftime("%H") if frame_dt else None, } def build_day_axis(day_keys: Iterable[str]) -> list[str]: parsed_days = [] for key in {str(value) for value in day_keys if value}: try: parsed_days.append(datetime.strptime(key, "%Y-%m-%d")) except ValueError: continue if not parsed_days: return sorted({str(value) for value in day_keys if value}) start_day = min(parsed_days) end_day = max(parsed_days) axis = [] current = start_day while current <= end_day: axis.append(current.strftime("%Y-%m-%d")) current += timedelta(days=1) extras = sorted({str(value) for value in day_keys if value} - set(axis)) return axis + extras def build_hour_axis(hour_keys: Iterable[str]) -> list[str]: axis = [f"{hour:02d}" for hour in range(24)] extras = sorted({str(value) for value in hour_keys if value} - set(axis)) return axis + extras def build_counter_rows(counter: Counter[str], axis: list[str], key_name: str, count_name: str) -> list[dict[str, Any]]: rows = [] seen = set() for label in axis: rows.append({key_name: label, count_name: int(counter.get(label, 0))}) seen.add(label) for label, value in sorted(counter.items()): if label in seen: continue rows.append({key_name: str(label), count_name: int(value)}) return rows def wrap_heading_deg(angle_deg: float) -> float: wrapped = ((float(angle_deg) + 180.0) % 360.0) - 180.0 return -180.0 if math.isclose(wrapped, 180.0, rel_tol=0.0, abs_tol=1e-9) else float(wrapped) def heading_interval_start(angle_deg: Optional[float], bin_width_deg: float = PORTRAIT_HEADING_BIN_DEG) -> Optional[float]: if angle_deg is None or not math.isfinite(angle_deg) or bin_width_deg <= 0: return None heading = wrap_heading_deg(float(angle_deg)) return math.floor((heading + 180.0) / bin_width_deg) * bin_width_deg - 180.0 def build_heading_rows(counter: Counter[float], bin_width_deg: float = PORTRAIT_HEADING_BIN_DEG) -> list[dict[str, Any]]: if bin_width_deg <= 0: return [] bin_count = max(1, int(round(360.0 / bin_width_deg))) rows = [] for index in range(bin_count): start_deg = -180.0 + index * bin_width_deg rows.append( { "heading_bin_start_deg": float(start_deg), "heading_bin_end_deg": float(start_deg + bin_width_deg), "heading_bin_label": interval_label(float(start_deg), float(bin_width_deg), unit="deg"), "count": int(counter.get(float(start_deg), 0)), } ) return rows def build_nonnegative_interval_rows(counter: Counter[float], bin_width_m: float, prefix: str) -> list[dict[str, Any]]: if bin_width_m <= 0: return [] max_start = max((float(start) for start, count in counter.items() if int(count) > 0), default=0.0) num_bins = max(1, int(math.floor(max_start / bin_width_m)) + 1) rows = [] for index in range(num_bins): start_m = float(index * bin_width_m) rows.append( { f"{prefix}_start_m": start_m, f"{prefix}_end_m": float(start_m + bin_width_m), f"{prefix}_label": interval_label(start_m, float(bin_width_m)), "count": int(counter.get(start_m, 0)), } ) return rows def build_class_count_rows( frame_counter: Counter[int], object_counter: Counter[int], class_names: dict[int, str], class_order: list[int], include_zero_rows: bool, ) -> list[dict[str, Any]]: rows = [] for cls_id in class_order: row = { "cls_id": int(cls_id), "cls_name": get_cls_name(class_names, int(cls_id)), "frame_count": int(frame_counter.get(int(cls_id), 0)), "object_count": int(object_counter.get(int(cls_id), 0)), } if include_zero_rows or row["frame_count"] > 0 or row["object_count"] > 0: rows.append(row) return rows def build_full_image_plot_calib(raw_calib: Optional[dict[str, Any]], ori_w: int, ori_h: int) -> Optional[dict[str, Any]]: if raw_calib is None: return None fx = float(raw_calib.get("fx", raw_calib.get("focal_u", ori_w))) fy = float(raw_calib.get("fy", raw_calib.get("focal_v", ori_h))) cx = float(raw_calib.get("cx", raw_calib.get("cu", ori_w / 2.0))) cy = float(raw_calib.get("cy", raw_calib.get("cv", ori_h / 2.0))) return { "fx": fx, "fy": fy, "cx": cx, "cy": cy, "distort_coeffs": list(raw_calib.get("distort_coeffs", [])), "depth_scale": float(raw_calib.get("depth_scale", 1.0)), } def create_data_portrait_accumulators() -> dict[str, Any]: return { "frame_counts_by_vehicle": Counter(), "day_counts_by_vehicle": defaultdict(Counter), "hour_counts_by_vehicle": defaultdict(Counter), "class_frame_counts_by_vehicle": defaultdict(Counter), "class_object_counts_by_vehicle": defaultdict(Counter), "heading_counts_by_vehicle": defaultdict(Counter), "lateral_counts_by_class": defaultdict(Counter), "longitudinal_counts_by_class": defaultdict(Counter), "total_day_counts": Counter(), "total_hour_counts": Counter(), "total_class_frame_counts": Counter(), "total_class_object_counts": Counter(), "total_heading_counts": Counter(), "all_day_keys": set(), "all_hour_keys": set(), "frames_with_mapped_objects": 0, "frames_with_valid_3d": 0, "processed_samples": 0, } def merge_counter_maps(dst: defaultdict[Any, Counter], src: defaultdict[Any, Counter]) -> None: for key, counter in src.items(): dst[key].update(counter) def merge_data_portrait_accumulators(dst: dict[str, Any], src: dict[str, Any]) -> None: dst["frame_counts_by_vehicle"].update(src["frame_counts_by_vehicle"]) merge_counter_maps(dst["day_counts_by_vehicle"], src["day_counts_by_vehicle"]) merge_counter_maps(dst["hour_counts_by_vehicle"], src["hour_counts_by_vehicle"]) merge_counter_maps(dst["class_frame_counts_by_vehicle"], src["class_frame_counts_by_vehicle"]) merge_counter_maps(dst["class_object_counts_by_vehicle"], src["class_object_counts_by_vehicle"]) merge_counter_maps(dst["heading_counts_by_vehicle"], src["heading_counts_by_vehicle"]) merge_counter_maps(dst["lateral_counts_by_class"], src["lateral_counts_by_class"]) merge_counter_maps(dst["longitudinal_counts_by_class"], src["longitudinal_counts_by_class"]) dst["total_day_counts"].update(src["total_day_counts"]) dst["total_hour_counts"].update(src["total_hour_counts"]) dst["total_class_frame_counts"].update(src["total_class_frame_counts"]) dst["total_class_object_counts"].update(src["total_class_object_counts"]) dst["total_heading_counts"].update(src["total_heading_counts"]) dst["all_day_keys"].update(src["all_day_keys"]) dst["all_hour_keys"].update(src["all_hour_keys"]) dst["frames_with_mapped_objects"] += int(src["frames_with_mapped_objects"]) dst["frames_with_valid_3d"] += int(src["frames_with_valid_3d"]) dst["processed_samples"] += int(src["processed_samples"]) def resolve_data_portrait_workers(requested_workers: int, num_entries: int) -> int: if num_entries <= 1: return 1 if requested_workers and requested_workers > 0: return max(1, min(int(requested_workers), int(num_entries))) return max(1, min(int(DEFAULT_DATA_PORTRAIT_WORKERS), int(num_entries))) def iter_data_portrait_chunks(entries: list[tuple[str, str]], chunk_size: int) -> Iterable[list[tuple[str, str]]]: step = max(1, int(chunk_size)) for index in range(0, len(entries), step): yield entries[index : index + step] def process_data_portrait_chunk( entries: list[tuple[str, str]], image_root: Path, class_map: dict[str, int], difficulty_weights: list[float], face_3d_classes: set[int], complete_3d_classes: set[int], img_w: int, img_h: int, face_visibility_score_thresh: float, ) -> dict[str, Any]: chunk_stats = create_data_portrait_accumulators() for entry in entries: label_path = entry_to_label_file(entry) image_path = entry_to_image_file(entry, image_root) frame_meta = parse_label_frame_metadata(label_path) vehicle_alias = str(frame_meta.get("vehicle_alias") or "unknown") day_key = frame_meta.get("day") hour_key = frame_meta.get("hour") chunk_stats["frame_counts_by_vehicle"][vehicle_alias] += 1 if day_key: chunk_stats["day_counts_by_vehicle"][vehicle_alias][day_key] += 1 chunk_stats["total_day_counts"][day_key] += 1 chunk_stats["all_day_keys"].add(day_key) if hour_key: chunk_stats["hour_counts_by_vehicle"][vehicle_alias][hour_key] += 1 chunk_stats["total_hour_counts"][hour_key] += 1 chunk_stats["all_hour_keys"].add(hour_key) lb_2d, lb_3d = parse_ground_3d_label_file( str(label_path), class_map, difficulty_weights, face_3d_classes, complete_3d_classes, min_wh=0.0, ) cls_ids = lb_2d["cls"].reshape(-1).astype(np.int32) if len(lb_2d["cls"]) else np.zeros((0,), dtype=np.int32) frame_cls_ids = {int(cls_id) for cls_id in cls_ids.tolist()} if frame_cls_ids: chunk_stats["frames_with_mapped_objects"] += 1 for cls_id in cls_ids.tolist(): cls_id_int = int(cls_id) chunk_stats["class_object_counts_by_vehicle"][vehicle_alias][cls_id_int] += 1 chunk_stats["total_class_object_counts"][cls_id_int] += 1 for cls_id in frame_cls_ids: chunk_stats["class_frame_counts_by_vehicle"][vehicle_alias][cls_id] += 1 chunk_stats["total_class_frame_counts"][cls_id] += 1 frame_has_valid_3d = False if lb_3d is not None and len(lb_3d) and len(cls_ids): raw_calib = read_raw_calib_from_label(image_path, label_path) gt_calib = build_full_image_plot_calib(raw_calib, img_w, img_h) if gt_calib is not None: for gt_index, cls_id in enumerate(cls_ids.tolist()): if gt_index >= len(lb_3d): break gt_attrs = extract_3d_attrs_from_gt( lb_3d[gt_index], int(cls_id), gt_calib, img_w, img_h, face_3d_classes, complete_3d_classes, score_thr=face_visibility_score_thresh, ) if gt_attrs is None: continue frame_has_valid_3d = True yaw_rad = to_float(gt_attrs.get("yaw")) if yaw_rad is not None: heading_start = heading_interval_start(math.degrees(yaw_rad), PORTRAIT_HEADING_BIN_DEG) if heading_start is not None: chunk_stats["heading_counts_by_vehicle"][vehicle_alias][heading_start] += 1 chunk_stats["total_heading_counts"][heading_start] += 1 center = gt_attrs.get("center") lateral_m = None longitudinal_m = None if center is not None and len(center) >= 3: lateral_m = to_float(center[0]) longitudinal_m = to_float(center[2]) if lateral_m is not None: lateral_start = depth_interval_start(abs(float(lateral_m)), PORTRAIT_LATERAL_BIN_M) if lateral_start is not None: chunk_stats["lateral_counts_by_class"][int(cls_id)][float(lateral_start)] += 1 if longitudinal_m is not None and longitudinal_m >= 0.0: longitudinal_start = depth_interval_start(float(longitudinal_m), PORTRAIT_LONGITUDINAL_BIN_M) if longitudinal_start is not None: chunk_stats["longitudinal_counts_by_class"][int(cls_id)][float(longitudinal_start)] += 1 if frame_has_valid_3d: chunk_stats["frames_with_valid_3d"] += 1 chunk_stats["processed_samples"] = len(entries) return chunk_stats def maybe_log_data_portrait_progress( split_name: str, processed_samples: int, total_samples: int, frame_counts_by_vehicle: Counter[str], total_class_object_counts: Counter[int], start_time: float, ) -> None: elapsed_minutes = (time.time() - start_time) / 60.0 print( f"[portrait:{split_name}] {processed_samples}/{total_samples} samples " f"vehicles={len(frame_counts_by_vehicle)} mapped_objects={sum(total_class_object_counts.values())} " f"elapsed={elapsed_minutes:.2f}m" ) def build_data_portrait( entries: list[tuple[str, str]], split_name: str, split_path: Path, image_root: Path, output_root: Path, class_map: dict[str, int], class_names: dict[int, str], difficulty_weights: list[float], face_3d_classes: set[int], complete_3d_classes: set[int], ori_img_size: tuple[int, int], face_visibility_score_thresh: float, log_every: int, workers: int = 0, ) -> dict[str, Any]: portrait_root = output_root / f"{str(split_name).lower()}_portrait" portrait_root.mkdir(parents=True, exist_ok=True) img_w = int(ori_img_size[0]) if ori_img_size and len(ori_img_size) > 0 else 1920 img_h = int(ori_img_size[1]) if ori_img_size and len(ori_img_size) > 1 else 1080 portrait_stats = create_data_portrait_accumulators() start_time = time.time() worker_count = resolve_data_portrait_workers(workers, len(entries)) next_log_sample = int(log_every) if log_every > 0 else None chunks = list(iter_data_portrait_chunks(entries, DATA_PORTRAIT_CHUNK_SIZE)) if worker_count > 1 and len(chunks) > 1: print( f"[portrait:{split_name}] scanning GT/calib with {worker_count} worker threads " f"(chunk_size={DATA_PORTRAIT_CHUNK_SIZE})..." ) with ThreadPoolExecutor(max_workers=worker_count) as executor: futures = [ executor.submit( process_data_portrait_chunk, entries=chunk, image_root=image_root, class_map=class_map, difficulty_weights=difficulty_weights, face_3d_classes=face_3d_classes, complete_3d_classes=complete_3d_classes, img_w=img_w, img_h=img_h, face_visibility_score_thresh=face_visibility_score_thresh, ) for chunk in chunks ] for future in as_completed(futures): chunk_stats = future.result() merge_data_portrait_accumulators(portrait_stats, chunk_stats) while next_log_sample is not None and portrait_stats["processed_samples"] >= next_log_sample: maybe_log_data_portrait_progress( split_name, next_log_sample, len(entries), portrait_stats["frame_counts_by_vehicle"], portrait_stats["total_class_object_counts"], start_time, ) next_log_sample += int(log_every) else: for chunk in chunks: chunk_stats = process_data_portrait_chunk( entries=chunk, image_root=image_root, class_map=class_map, difficulty_weights=difficulty_weights, face_3d_classes=face_3d_classes, complete_3d_classes=complete_3d_classes, img_w=img_w, img_h=img_h, face_visibility_score_thresh=face_visibility_score_thresh, ) merge_data_portrait_accumulators(portrait_stats, chunk_stats) while next_log_sample is not None and portrait_stats["processed_samples"] >= next_log_sample: maybe_log_data_portrait_progress( split_name, next_log_sample, len(entries), portrait_stats["frame_counts_by_vehicle"], portrait_stats["total_class_object_counts"], start_time, ) next_log_sample += int(log_every) if log_every > 0: last_logged_sample = max(0, (next_log_sample or int(log_every)) - int(log_every)) else: last_logged_sample = 0 if log_every > 0 and portrait_stats["processed_samples"] > 0 and last_logged_sample < len(entries): maybe_log_data_portrait_progress( split_name, len(entries), len(entries), portrait_stats["frame_counts_by_vehicle"], portrait_stats["total_class_object_counts"], start_time, ) frame_counts_by_vehicle = portrait_stats["frame_counts_by_vehicle"] day_counts_by_vehicle = portrait_stats["day_counts_by_vehicle"] hour_counts_by_vehicle = portrait_stats["hour_counts_by_vehicle"] class_frame_counts_by_vehicle = portrait_stats["class_frame_counts_by_vehicle"] class_object_counts_by_vehicle = portrait_stats["class_object_counts_by_vehicle"] heading_counts_by_vehicle = portrait_stats["heading_counts_by_vehicle"] lateral_counts_by_class = portrait_stats["lateral_counts_by_class"] longitudinal_counts_by_class = portrait_stats["longitudinal_counts_by_class"] total_day_counts = portrait_stats["total_day_counts"] total_hour_counts = portrait_stats["total_hour_counts"] total_class_frame_counts = portrait_stats["total_class_frame_counts"] total_class_object_counts = portrait_stats["total_class_object_counts"] total_heading_counts = portrait_stats["total_heading_counts"] all_day_keys = portrait_stats["all_day_keys"] all_hour_keys = portrait_stats["all_hour_keys"] frames_with_mapped_objects = portrait_stats["frames_with_mapped_objects"] frames_with_valid_3d = portrait_stats["frames_with_valid_3d"] vehicle_order = [alias for alias, _ in sorted(frame_counts_by_vehicle.items(), key=lambda item: (-int(item[1]), str(item[0])))] class_order = sorted(class_names, key=lambda cls_id: (-int(total_class_object_counts.get(int(cls_id), 0)), int(cls_id))) if not class_order: class_order = sorted({int(cls_id) for cls_id in total_class_object_counts}) day_axis = build_day_axis(all_day_keys) hour_axis = build_hour_axis(all_hour_keys) vehicle_rows = [{"vehicle_alias": alias, "frame_count": int(frame_counts_by_vehicle[alias])} for alias in vehicle_order] daily_total_rows = build_counter_rows(total_day_counts, day_axis, "day", "frame_count") hourly_total_rows = build_counter_rows(total_hour_counts, hour_axis, "hour", "frame_count") heading_total_rows = build_heading_rows(total_heading_counts, PORTRAIT_HEADING_BIN_DEG) class_total_rows = build_class_count_rows(total_class_frame_counts, total_class_object_counts, class_names, class_order, include_zero_rows=False) daily_cards = [ { "title": "All Vehicles", "subtitle": f"frames={len(entries)}", "rows": daily_total_rows, } ] hourly_cards = [ { "title": "All Vehicles", "subtitle": f"frames={len(entries)}", "rows": hourly_total_rows, } ] heading_cards = [ { "title": "All Vehicles", "subtitle": f"objects={sum(int(row.get('count', 0)) for row in heading_total_rows)}", "rows": heading_total_rows, } ] class_vehicle_groups = [] for alias in vehicle_order: alias_daily_rows = build_counter_rows(day_counts_by_vehicle[alias], day_axis, "day", "frame_count") alias_hourly_rows = build_counter_rows(hour_counts_by_vehicle[alias], hour_axis, "hour", "frame_count") alias_heading_rows = build_heading_rows(heading_counts_by_vehicle[alias], PORTRAIT_HEADING_BIN_DEG) alias_class_rows = build_class_count_rows( class_frame_counts_by_vehicle[alias], class_object_counts_by_vehicle[alias], class_names, class_order, include_zero_rows=False, ) daily_cards.append({"title": alias, "subtitle": f"frames={frame_counts_by_vehicle[alias]}", "rows": alias_daily_rows}) hourly_cards.append({"title": alias, "subtitle": f"frames={frame_counts_by_vehicle[alias]}", "rows": alias_hourly_rows}) heading_cards.append( { "title": alias, "subtitle": f"objects={sum(int(row.get('count', 0)) for row in alias_heading_rows)}", "rows": alias_heading_rows, } ) class_vehicle_groups.append( { "vehicle_alias": alias, "summary": ( f"{alias} " f"(frames={frame_counts_by_vehicle[alias]}, " f"class_frames={sum(int(row.get('frame_count', 0)) for row in alias_class_rows)}, " f"objects={sum(int(row.get('object_count', 0)) for row in alias_class_rows)})" ), "rows": alias_class_rows, } ) lateral_cards = [] longitudinal_cards = [] lateral_rows_flat = [] longitudinal_rows_flat = [] for cls_id in class_order: cls_name = get_cls_name(class_names, int(cls_id)) lateral_rows = build_nonnegative_interval_rows(lateral_counts_by_class[int(cls_id)], PORTRAIT_LATERAL_BIN_M, "lateral_bin") lateral_rows = [ {"cls_id": int(cls_id), "cls_name": cls_name, **row} for row in lateral_rows if int(row.get("count", 0)) > 0 ] if lateral_rows: lateral_cards.append( { "title": cls_name, "subtitle": f"id={int(cls_id)} objects={sum(int(row.get('count', 0)) for row in lateral_rows)}", "rows": lateral_rows, } ) lateral_rows_flat.extend(lateral_rows) longitudinal_rows = build_nonnegative_interval_rows( longitudinal_counts_by_class[int(cls_id)], PORTRAIT_LONGITUDINAL_BIN_M, "longitudinal_bin" ) longitudinal_rows = [ {"cls_id": int(cls_id), "cls_name": cls_name, **row} for row in longitudinal_rows if int(row.get("count", 0)) > 0 ] if longitudinal_rows: longitudinal_cards.append( { "title": cls_name, "subtitle": f"id={int(cls_id)} objects={sum(int(row.get('count', 0)) for row in longitudinal_rows)}", "rows": longitudinal_rows, } ) longitudinal_rows_flat.extend(longitudinal_rows) frames_by_vehicle_csv = portrait_root / "frames_by_vehicle.csv" frames_by_day_csv = portrait_root / "frames_by_day.csv" frames_by_hour_csv = portrait_root / "frames_by_hour.csv" class_counts_csv = portrait_root / "class_counts.csv" heading_counts_csv = portrait_root / "heading_counts.csv" lateral_counts_csv = portrait_root / "class_lateral_distance_5m.csv" longitudinal_counts_csv = portrait_root / "class_longitudinal_distance_10m.csv" write_rows_csv(frames_by_vehicle_csv, vehicle_rows, fallback_fields=["vehicle_alias", "frame_count"]) write_rows_csv( frames_by_day_csv, [{"scope": "total", "vehicle_alias": "all", **row} for row in daily_total_rows] + [ {"scope": "vehicle", "vehicle_alias": card["title"], **row} for card in daily_cards[1:] for row in card["rows"] ], fallback_fields=["scope", "vehicle_alias", "day", "frame_count"], ) write_rows_csv( frames_by_hour_csv, [{"scope": "total", "vehicle_alias": "all", **row} for row in hourly_total_rows] + [ {"scope": "vehicle", "vehicle_alias": card["title"], **row} for card in hourly_cards[1:] for row in card["rows"] ], fallback_fields=["scope", "vehicle_alias", "hour", "frame_count"], ) write_rows_csv( class_counts_csv, [{"scope": "total", "vehicle_alias": "all", **row} for row in class_total_rows] + [ {"scope": "vehicle", "vehicle_alias": str(group.get("vehicle_alias", "unknown")), **row} for group in class_vehicle_groups for row in group["rows"] ], fallback_fields=["scope", "vehicle_alias", "cls_id", "cls_name", "frame_count", "object_count"], ) write_rows_csv( heading_counts_csv, [{"scope": "total", "vehicle_alias": "all", **row} for row in heading_total_rows] + [ {"scope": "vehicle", "vehicle_alias": card["title"], **row} for card in heading_cards[1:] for row in card["rows"] ], fallback_fields=["scope", "vehicle_alias", "heading_bin_start_deg", "heading_bin_end_deg", "heading_bin_label", "count"], ) write_rows_csv( lateral_counts_csv, lateral_rows_flat, fallback_fields=["cls_id", "cls_name", "lateral_bin_start_m", "lateral_bin_end_m", "lateral_bin_label", "count"], ) write_rows_csv( longitudinal_counts_csv, longitudinal_rows_flat, fallback_fields=[ "cls_id", "cls_name", "longitudinal_bin_start_m", "longitudinal_bin_end_m", "longitudinal_bin_label", "count", ], ) summary_payload = { "split": str(split_name), "split_path": str(split_path), "image_root": str(image_root), "artifact_root": str(portrait_root), "summary": { "num_entries": len(entries), "vehicles": len(vehicle_order), "frames_with_mapped_objects": int(frames_with_mapped_objects), "frames_with_valid_3d": int(frames_with_valid_3d), "mapped_objects": int(sum(total_class_object_counts.values())), "heading_objects": int(sum(total_heading_counts.values())), "lateral_objects": int(sum(int(row.get("count", 0)) for row in lateral_rows_flat)), "longitudinal_objects": int(sum(int(row.get("count", 0)) for row in longitudinal_rows_flat)), }, "vehicle_rows": vehicle_rows, "daily_cards": daily_cards, "hourly_cards": hourly_cards, "class_total_rows": class_total_rows, "class_vehicle_groups": class_vehicle_groups, "heading_cards": heading_cards, "lateral_cards": lateral_cards, "longitudinal_cards": longitudinal_cards, "artifact_paths": { "frames_by_vehicle_csv": str(frames_by_vehicle_csv), "frames_by_day_csv": str(frames_by_day_csv), "frames_by_hour_csv": str(frames_by_hour_csv), "class_counts_csv": str(class_counts_csv), "heading_counts_csv": str(heading_counts_csv), "lateral_counts_csv": str(lateral_counts_csv), "longitudinal_counts_csv": str(longitudinal_counts_csv), }, } summary_path = portrait_root / "summary.json" write_json(summary_path, summary_payload) summary_payload["summary_path"] = str(summary_path) return summary_payload def load_existing_data_portrait(output_root: Path, split_name: str) -> Optional[dict[str, Any]]: portrait_root = output_root / f"{str(split_name).lower()}_portrait" summary_path = portrait_root / "summary.json" if not summary_path.is_file(): return None payload = json.loads(summary_path.read_text(encoding="utf-8")) if not isinstance(payload, dict): return None payload = dict(payload) payload["summary_path"] = str(summary_path) return payload def filter_labels_by_classes( lb_2d: dict[str, Any], lb_3d: Optional[np.ndarray], include_classes: Optional[set[int]] ) -> tuple[dict[str, Any], Optional[np.ndarray]]: if include_classes is None or len(lb_2d["cls"]) == 0: return lb_2d, lb_3d cls = lb_2d["cls"].reshape(-1).astype(np.int32) keep = np.isin(cls, np.asarray(sorted(include_classes), dtype=np.int32)) filtered_2d = { **lb_2d, "cls": lb_2d["cls"][keep], "bboxes": lb_2d["bboxes"][keep], "difficulties": lb_2d["difficulties"][keep], } if lb_3d is None: return filtered_2d, None return filtered_2d, lb_3d[keep] def filter_label_names_by_classes(class_names: list[str], cls_ids: np.ndarray, include_classes: Optional[set[int]]) -> list[str]: if include_classes is None or not class_names: return class_names cls = np.asarray(cls_ids, dtype=np.int32).reshape(-1) keep = np.isin(cls, np.asarray(sorted(include_classes), dtype=np.int32)) return [name for name, keep_item in zip(class_names, keep.tolist()) if keep_item] def filter_labels_by_min_wh( lb_2d: dict[str, Any], lb_3d: Optional[np.ndarray], img_w: int, img_h: int, min_wh_px: float, ) -> tuple[dict[str, Any], Optional[np.ndarray]]: if min_wh_px <= 0 or len(lb_2d["bboxes"]) == 0: return lb_2d, lb_3d bboxes = np.asarray(lb_2d["bboxes"], dtype=np.float32) w_px = bboxes[:, 2] * float(img_w) h_px = bboxes[:, 3] * float(img_h) keep = (w_px >= float(min_wh_px)) | (h_px >= float(min_wh_px)) if keep.all(): return lb_2d, lb_3d filtered_2d = { **lb_2d, "cls": lb_2d["cls"][keep], "bboxes": lb_2d["bboxes"][keep], "difficulties": lb_2d["difficulties"][keep], } if lb_3d is None: return filtered_2d, None return filtered_2d, lb_3d[keep] def filter_label_names_by_min_wh(class_names: list[str], bboxes: np.ndarray, img_w: int, img_h: int, min_wh_px: float) -> list[str]: if min_wh_px <= 0 or not class_names: return class_names boxes = np.asarray(bboxes, dtype=np.float32) if boxes.size == 0: return [] w_px = boxes[:, 2] * float(img_w) h_px = boxes[:, 3] * float(img_h) keep = (w_px >= float(min_wh_px)) | (h_px >= float(min_wh_px)) return [name for name, keep_item in zip(class_names, keep.tolist()) if keep_item] def xywhn_to_xyxy(boxes_xywhn: np.ndarray, img_w: int, img_h: int) -> np.ndarray: if boxes_xywhn is None or len(boxes_xywhn) == 0: return np.zeros((0, 4), dtype=np.float32) boxes = np.asarray(boxes_xywhn, dtype=np.float32).copy() boxes[:, 0] = (boxes_xywhn[:, 0] - boxes_xywhn[:, 2] * 0.5) * img_w boxes[:, 1] = (boxes_xywhn[:, 1] - boxes_xywhn[:, 3] * 0.5) * img_h boxes[:, 2] = (boxes_xywhn[:, 0] + boxes_xywhn[:, 2] * 0.5) * img_w boxes[:, 3] = (boxes_xywhn[:, 1] + boxes_xywhn[:, 3] * 0.5) * img_h return boxes def box_iou_matrix(gt_boxes: np.ndarray, pred_boxes: np.ndarray) -> np.ndarray: if len(gt_boxes) == 0 or len(pred_boxes) == 0: return np.zeros((len(gt_boxes), len(pred_boxes)), dtype=np.float32) gt = gt_boxes.astype(np.float32) pred = pred_boxes.astype(np.float32) inter_x1 = np.maximum(gt[:, None, 0], pred[None, :, 0]) inter_y1 = np.maximum(gt[:, None, 1], pred[None, :, 1]) inter_x2 = np.minimum(gt[:, None, 2], pred[None, :, 2]) inter_y2 = np.minimum(gt[:, None, 3], pred[None, :, 3]) inter_w = np.clip(inter_x2 - inter_x1, 0.0, None) inter_h = np.clip(inter_y2 - inter_y1, 0.0, None) inter = inter_w * inter_h gt_area = np.clip(gt[:, 2] - gt[:, 0], 0.0, None) * np.clip(gt[:, 3] - gt[:, 1], 0.0, None) pred_area = np.clip(pred[:, 2] - pred[:, 0], 0.0, None) * np.clip(pred[:, 3] - pred[:, 1], 0.0, None) union = gt_area[:, None] + pred_area[None, :] - inter return inter / np.clip(union, 1e-6, None) def greedy_match_indices(gt_cls: np.ndarray, gt_boxes: np.ndarray, pred_cls: np.ndarray, pred_boxes: np.ndarray, iou_thr: float = 0.5): if len(gt_boxes) == 0 or len(pred_boxes) == 0: return np.zeros((0, 2), dtype=np.int32), np.zeros((len(gt_boxes), len(pred_boxes)), dtype=np.float32) iou = box_iou_matrix(gt_boxes, pred_boxes) class_mask = gt_cls[:, None] == pred_cls[None, :] iou = iou * class_mask.astype(np.float32) matches = np.array(np.nonzero(iou >= iou_thr)).T if matches.shape[0] == 0: return matches.astype(np.int32), iou if matches.shape[0] > 1: scores = iou[matches[:, 0], matches[:, 1]] matches = matches[scores.argsort()[::-1]] matches = matches[np.unique(matches[:, 1], return_index=True)[1]] matches = matches[np.unique(matches[:, 0], return_index=True)[1]] return matches.astype(np.int32), iou def greedy_match_indices_any_class(gt_boxes: np.ndarray, pred_boxes: np.ndarray, iou_thr: float = 0.5): if len(gt_boxes) == 0 or len(pred_boxes) == 0: return np.zeros((0, 2), dtype=np.int32), np.zeros((len(gt_boxes), len(pred_boxes)), dtype=np.float32) iou = box_iou_matrix(gt_boxes, pred_boxes) matches = np.array(np.nonzero(iou >= iou_thr)).T if matches.shape[0] == 0: return matches.astype(np.int32), iou if matches.shape[0] > 1: scores = iou[matches[:, 0], matches[:, 1]] matches = matches[scores.argsort()[::-1]] matches = matches[np.unique(matches[:, 1], return_index=True)[1]] matches = matches[np.unique(matches[:, 0], return_index=True)[1]] return matches.astype(np.int32), iou def is_focused_confusion_spatial_object( lateral_m: Optional[float], longitudinal_m: Optional[float], max_abs_lateral_m: float = FOCUSED_CONFUSION_MAX_ABS_LATERAL_M, max_abs_longitudinal_m: float = FOCUSED_CONFUSION_MAX_ABS_LONGITUDINAL_M, ) -> bool: if lateral_m is None or longitudinal_m is None: return False return abs(float(lateral_m)) < float(max_abs_lateral_m) and abs(float(longitudinal_m)) < float(max_abs_longitudinal_m) def is_focused_confusion_gt_object( lateral_m: Optional[float], longitudinal_m: Optional[float], difficulty: Any, max_abs_lateral_m: float = FOCUSED_CONFUSION_MAX_ABS_LATERAL_M, max_abs_longitudinal_m: float = FOCUSED_CONFUSION_MAX_ABS_LONGITUDINAL_M, required_difficulty: int = FOCUSED_CONFUSION_REQUIRED_DIFFICULTY, ) -> bool: difficulty_value = to_float(difficulty) if difficulty_value is None or int(round(float(difficulty_value))) != int(required_difficulty): return False return is_focused_confusion_spatial_object( lateral_m=lateral_m, longitudinal_m=longitudinal_m, max_abs_lateral_m=max_abs_lateral_m, max_abs_longitudinal_m=max_abs_longitudinal_m, ) def focused_confusion_max_abs_longitudinal_m_for_roi(roi_name: str) -> float: return float(ROI0_FOCUSED_CONFUSION_MAX_ABS_LONGITUDINAL_M if str(roi_name).lower() == "roi0" else FOCUSED_CONFUSION_MAX_ABS_LONGITUDINAL_M) def resolve_face_center_eval_face_type( gt_visible_faces: list[tuple[int, np.ndarray]], gt_decoded: Optional[dict[str, Any]], pred_decoded: Optional[dict[str, Any]], ) -> Optional[int]: gt_visible_face_types = {int(face_type) for face_type, _ in gt_visible_faces} pred_visible_face_type = None if pred_decoded is None else pred_decoded.get("visible_face_type") if pred_visible_face_type is not None and int(pred_visible_face_type) in gt_visible_face_types: return int(pred_visible_face_type) gt_visible_face_type = None if gt_decoded is None else gt_decoded.get("visible_face_type") if gt_visible_face_type is not None and int(gt_visible_face_type) in gt_visible_face_types: return int(gt_visible_face_type) return None def update_subset_confusion_matrix( confusion_matrix: ConfusionMatrix, gt_cls: np.ndarray, gt_boxes: np.ndarray, pred_cls: np.ndarray, pred_boxes: np.ndarray, pred_conf: np.ndarray, gt_mask: np.ndarray, pred_mask: np.ndarray, conf: float = 0.0, iou_thres: float = 0.5, ) -> None: gt_cls = np.asarray(gt_cls, dtype=np.int32).reshape(-1) gt_boxes = np.asarray(gt_boxes, dtype=np.float32).reshape(-1, 4) pred_cls = np.asarray(pred_cls, dtype=np.int32).reshape(-1) pred_boxes = np.asarray(pred_boxes, dtype=np.float32).reshape(-1, 4) pred_conf = np.asarray(pred_conf, dtype=np.float32).reshape(-1) gt_mask = np.asarray(gt_mask, dtype=bool).reshape(-1) pred_mask = np.asarray(pred_mask, dtype=bool).reshape(-1) if gt_mask.shape[0] != gt_cls.shape[0] or pred_mask.shape[0] != pred_cls.shape[0]: raise ValueError("Focused confusion matrix masks must align with GT/pred arrays.") if pred_conf.size: keep = pred_conf > float(conf) pred_cls = pred_cls[keep] pred_boxes = pred_boxes[keep] pred_mask = pred_mask[keep] else: pred_cls = np.zeros((0,), dtype=np.int32) pred_boxes = np.zeros((0, 4), dtype=np.float32) pred_mask = np.zeros((0,), dtype=bool) matches, _ = greedy_match_indices_any_class(gt_boxes, pred_boxes, iou_thr=iou_thres) matched_gt_to_pred = {int(gt_index): int(pred_index) for gt_index, pred_index in matches.tolist()} matched_pred = {int(pred_index) for _, pred_index in matches.tolist()} for gt_index, gt_class in enumerate(gt_cls.tolist()): if not bool(gt_mask[gt_index]): continue pred_index = matched_gt_to_pred.get(int(gt_index)) if pred_index is None: confusion_matrix.matrix[confusion_matrix.nc, int(gt_class)] += 1 continue confusion_matrix.matrix[int(pred_cls[pred_index]), int(gt_class)] += 1 for pred_index, pred_class in enumerate(pred_cls.tolist()): if int(pred_index) in matched_pred or not bool(pred_mask[pred_index]): continue confusion_matrix.matrix[int(pred_class), confusion_matrix.nc] += 1 def compute_2d_tp_matrix( gt_cls: np.ndarray, gt_boxes: np.ndarray, pred_cls: np.ndarray, pred_boxes: np.ndarray, iou_thresholds: np.ndarray, ) -> np.ndarray: correct = np.zeros((len(pred_cls), len(iou_thresholds)), dtype=bool) if len(gt_boxes) == 0 or len(pred_boxes) == 0: return correct iou = box_iou_matrix(gt_boxes, pred_boxes) correct_class = gt_cls[:, None] == pred_cls[None, :] iou = iou * correct_class.astype(np.float32) for idx, threshold in enumerate(iou_thresholds.tolist()): matches = np.array(np.nonzero(iou >= threshold)).T if matches.shape[0]: if matches.shape[0] > 1: matches = matches[iou[matches[:, 0], matches[:, 1]].argsort()[::-1]] matches = matches[np.unique(matches[:, 1], return_index=True)[1]] matches = matches[np.unique(matches[:, 0], return_index=True)[1]] correct[matches[:, 1].astype(int), idx] = True return correct def make_2d_ap_store() -> dict[str, list[np.ndarray]]: return {"tp": [], "conf": [], "pred_cls": [], "target_cls": []} def add_2d_ap_sample( store: dict[str, list[np.ndarray]], gt_cls: np.ndarray, gt_boxes: np.ndarray, pred_cls: np.ndarray, pred_boxes: np.ndarray, pred_conf: np.ndarray, iou_thresholds: np.ndarray, ) -> None: store["target_cls"].append(np.asarray(gt_cls, dtype=np.int32)) if len(pred_cls) == 0: return tp = compute_2d_tp_matrix( np.asarray(gt_cls, dtype=np.int32), np.asarray(gt_boxes, dtype=np.float32), np.asarray(pred_cls, dtype=np.int32), np.asarray(pred_boxes, dtype=np.float32), iou_thresholds=np.asarray(iou_thresholds, dtype=np.float32), ) store["tp"].append(tp.astype(np.float32)) store["conf"].append(np.asarray(pred_conf, dtype=np.float32)) store["pred_cls"].append(np.asarray(pred_cls, dtype=np.int32)) def summarize_2d_ap_store( store: dict[str, list[np.ndarray]], plot: bool = False, save_dir: Optional[Path] = None, names: Optional[dict[int, str]] = None, prefix: str = "", ) -> dict[str, Any]: target_cls = np.concatenate(store["target_cls"], axis=0) if store["target_cls"] else np.zeros((0,), dtype=np.int32) fallback_threshold = float(MIN_CONFIDENCE_FOR_2D_THRESHOLD_SEARCH) if not store["tp"] or target_cls.size == 0: unique_classes = np.unique(target_cls) return { "map50": None, "map50_95": None, "threshold_advice": { "recommended_confidence": fallback_threshold, "mean_precision": None, "mean_recall": None, "mean_f1": None, "source": "fallback_no_detections", "curve_point_count": 0, }, "curve_paths": {}, "per_class": { int(cls_id): { "map50": None, "map50_95": None, "precision_ap": None, "recall_ap": None, "f1_at_recommended_confidence": None, "optimal_confidence": fallback_threshold, "optimal_precision": None, "optimal_recall": None, "optimal_f1": None, } for cls_id in unique_classes }, } tp = np.concatenate(store["tp"], axis=0) conf = np.concatenate(store["conf"], axis=0) pred_cls = np.concatenate(store["pred_cls"], axis=0) resolved_save_dir = Path() if save_dir is None else Path(save_dir) ( _tp, _fp, p, r, f1, ap, unique_classes, p_curve, r_curve, f1_curve, x, _prec_values, ) = ap_per_class( tp, conf, pred_cls, target_cls, plot=plot, save_dir=resolved_save_dir, names=names or {}, prefix=prefix, ) best_index = int(smooth(f1_curve.mean(0), 0.1).argmax()) if f1_curve.size else 0 recommended_confidence = float(x[best_index]) if x.size else fallback_threshold mean_precision = float(p.mean()) if p.size else None mean_recall = float(r.mean()) if r.size else None mean_f1 = float(f1.mean()) if f1.size else None curve_paths = {} if save_dir is not None: candidate_paths = { "pr_curve": resolved_save_dir / f"{prefix}PR_curve.png", "f1_curve": resolved_save_dir / f"{prefix}F1_curve.png", "precision_curve": resolved_save_dir / f"{prefix}P_curve.png", "recall_curve": resolved_save_dir / f"{prefix}R_curve.png", } curve_paths = {key: str(path) for key, path in candidate_paths.items() if path.is_file()} per_class = {} for class_index, cls_id in enumerate(unique_classes.tolist()): class_best_index = int(np.argmax(f1_curve[class_index])) if f1_curve.shape[1] else 0 per_class[int(cls_id)] = { "map50": float(ap[class_index, 0]), "map50_95": float(ap[class_index].mean()), "precision_ap": float(p[class_index]), "recall_ap": float(r[class_index]), "f1_at_recommended_confidence": float(f1[class_index]), "optimal_confidence": float(x[class_best_index]) if x.size else fallback_threshold, "optimal_precision": float(p_curve[class_index, class_best_index]) if p_curve.shape[1] else None, "optimal_recall": float(r_curve[class_index, class_best_index]) if r_curve.shape[1] else None, "optimal_f1": float(f1_curve[class_index, class_best_index]) if f1_curve.shape[1] else None, } return { "map50": float(ap[:, 0].mean()) if ap.size else None, "map50_95": float(ap.mean()) if ap.size else None, "threshold_advice": { "recommended_confidence": recommended_confidence, "mean_precision": mean_precision, "mean_recall": mean_recall, "mean_f1": mean_f1, "source": "max_mean_f1", "curve_point_count": int(x.size), }, "curve_paths": curve_paths, "per_class": per_class, } def summarize_2d_classification_from_confusion(confusion_matrix: ConfusionMatrix) -> dict[str, Any]: nc = int(confusion_matrix.nc) matrix = np.asarray(confusion_matrix.matrix, dtype=np.float64) if matrix.shape[0] < nc or matrix.shape[1] < nc: return {"overall_accuracy": None, "overall_pairs": 0, "per_class": {}, "matrix": matrix.tolist()} matched_matrix = matrix[:nc, :nc] overall_pairs = int(np.sum(matched_matrix)) overall_correct = float(np.trace(matched_matrix)) per_class: dict[int, dict[str, Any]] = {} for cls_id in range(nc): pairs = int(np.sum(matched_matrix[:, cls_id])) correct = float(matched_matrix[cls_id, cls_id]) per_class[int(cls_id)] = { "cls_eval_pairs_2d": pairs, "cls_correct_2d": int(correct), "cls_acc_2d": (correct / float(pairs)) if pairs > 0 else None, } return { "overall_accuracy": (overall_correct / float(overall_pairs)) if overall_pairs > 0 else None, "overall_pairs": overall_pairs, "per_class": per_class, "matrix": matrix.tolist(), } def make_label_accuracy_store(label_order: Iterable[str]) -> dict[str, Any]: return { "label_order": [str(label) for label in label_order], "total": 0, "correct": 0, "per_label": defaultdict(lambda: {"total": 0, "correct": 0}), "matrix": defaultdict(lambda: defaultdict(int)), } def add_label_accuracy_sample(store: dict[str, Any], gt_label: Optional[str], pred_label: Optional[str]) -> None: if gt_label is None or pred_label is None: return gt_label = str(gt_label) pred_label = str(pred_label) store["total"] += 1 store["per_label"][gt_label]["total"] += 1 store["matrix"][gt_label][pred_label] += 1 if gt_label == pred_label: store["correct"] += 1 store["per_label"][gt_label]["correct"] += 1 def summarize_label_accuracy_store(store: dict[str, Any]) -> dict[str, Any]: label_order = [str(label) for label in store.get("label_order", [])] seen_labels = set(label_order) seen_labels.update(str(label) for label in store.get("per_label", {}).keys()) for gt_label, pred_counts in store.get("matrix", {}).items(): seen_labels.add(str(gt_label)) seen_labels.update(str(label) for label in pred_counts.keys()) ordered_labels = [label for label in label_order if label in seen_labels] + sorted(seen_labels - set(label_order)) per_label = {} accuracies = [] for label in ordered_labels: counts = store.get("per_label", {}).get(label, {}) total = int(counts.get("total", 0) or 0) correct = int(counts.get("correct", 0) or 0) accuracy = (correct / float(total)) if total > 0 else None per_label[label] = {"total": total, "correct": correct, "accuracy": accuracy} if accuracy is not None: accuracies.append(float(accuracy)) matrix = { gt_label: {pred_label: int(store.get("matrix", {}).get(gt_label, {}).get(pred_label, 0) or 0) for pred_label in ordered_labels} for gt_label in ordered_labels } total = int(store.get("total", 0) or 0) correct = int(store.get("correct", 0) or 0) return { "label_order": ordered_labels, "overall_accuracy": (correct / float(total)) if total > 0 else None, "mean_accuracy": mean_or_none(accuracies), "total": total, "correct": correct, "per_label": per_label, "matrix": matrix, } def label_accuracy_rows(summary: dict[str, Any], label_key: str = "label") -> list[dict[str, Any]]: per_label = summary.get("per_label") or {} return [ { label_key: label, "total": int((per_label.get(label) or {}).get("total", 0) or 0), "correct": int((per_label.get(label) or {}).get("correct", 0) or 0), "accuracy": (per_label.get(label) or {}).get("accuracy"), } for label in summary.get("label_order", []) ] def label_confusion_matrix_rows(summary: dict[str, Any], label_key: str = "gt_label") -> list[dict[str, Any]]: labels = [str(label) for label in summary.get("label_order", [])] matrix = summary.get("matrix") or {} rows = [] for gt_label in labels: row = {label_key: gt_label} pred_counts = matrix.get(gt_label) or {} for pred_label in labels: row[pred_label] = int(pred_counts.get(pred_label, 0) or 0) rows.append(row) return rows def face_type_label(face_type: Any) -> Optional[str]: try: face_type_int = int(face_type) except (TypeError, ValueError): return None return FACE_NAMES.get(face_type_int) def cut_state_label(cut_state: Any) -> Optional[str]: try: cut_state_int = int(cut_state) except (TypeError, ValueError): return None if cut_state_int == 1: return "cut_in" if cut_state_int == 2: return "cut_out" return None def gt_face_selection_label(target_42: np.ndarray, gt_decoded: Optional[dict[str, Any]]) -> Optional[str]: cut_label = cut_state_label(get_gt_cut_state(target_42)) if cut_label is not None: return cut_label if gt_decoded is None: return None return face_type_label(gt_decoded.get("visible_face_type")) def pred_face_selection_label(pred_41: np.ndarray, pred_decoded: Optional[dict[str, Any]]) -> Optional[str]: cut_label = cut_state_label(get_pred_cut_state(pred_41)) if cut_label is not None: return cut_label if pred_decoded is None: return None return face_type_label(pred_decoded.get("visible_face_type")) def is_occluded_class_name(cls_name: Any) -> bool: return str(cls_name).endswith("_fake") def fake_class_label(cls_name: Any) -> str: name = str(cls_name) if name in {"car_fake", "bicyclist_fake", "pedestrian_fake", "car", "bicycle", "pedestrian"}: return name return "non_fake" def occlusion_binary_label(cls_name: Any) -> str: return "occluded" if is_occluded_class_name(cls_name) else "visible" def angle_abs_deg(pred_rad: float, gt_rad: float) -> Optional[float]: if pred_rad is None or gt_rad is None: return None if not (math.isfinite(pred_rad) and math.isfinite(gt_rad)): return None diff = (float(pred_rad) - float(gt_rad) + math.pi) % (2.0 * math.pi) - math.pi diff = abs(diff) diff = min(diff, abs(math.pi - diff)) return math.degrees(diff) def depth_bin(depth: Optional[float]) -> str: if depth is None or not math.isfinite(depth): return "unknown" if depth < 20.0: return "<20m" if depth < 40.0: return "20-40m" if depth < 60.0: return "40-60m" return ">=60m" def bbox_diag_bin(diag_px: Optional[float]) -> str: if diag_px is None or not math.isfinite(diag_px): return "unknown" if diag_px < 32.0: return "<32px" if diag_px < 64.0: return "32-64px" if diag_px < 128.0: return "64-128px" return ">=128px" def sanitize_name(value: str) -> str: safe = [] for char in str(value): if char.isalnum() or char in ("-", "_", "."): safe.append(char) else: safe.append("_") return "".join(safe).strip("_") or "item" def to_float(value: Any) -> Optional[float]: if value is None: return None if isinstance(value, (np.floating, np.integer)): value = value.item() try: value = float(value) except (TypeError, ValueError): return None if not math.isfinite(value): return None return value def to_list(values: Any) -> list[float]: arr = np.asarray(values, dtype=np.float32).reshape(-1) return [float(v) for v in arr.tolist()] def to_serializable_array(values: Any) -> Any: if values is None: return None arr = np.asarray(values) if arr.ndim == 0: return arr.item() return arr.tolist() def get_cls_name(names: Any, cls_id: int) -> str: if isinstance(names, dict): return str(names.get(cls_id, cls_id)) if isinstance(names, (list, tuple)) and 0 <= cls_id < len(names): return str(names[cls_id]) return str(cls_id) def names_to_dict(names: Any) -> dict[int, str]: if isinstance(names, dict): return {int(key): str(value) for key, value in names.items()} if isinstance(names, (list, tuple)): return {int(index): str(value) for index, value in enumerate(names)} return {} def format_float(value: Optional[float], digits: int = 3) -> str: if value is None or not math.isfinite(value): return "n/a" return f"{value:.{digits}f}" def format_percent(value: Optional[float], digits: int = 1) -> str: if value is None or not math.isfinite(value): return "n/a" return f"{value * 100:.{digits}f}%" def depth_interval_start(depth_m: Optional[float], bin_width_m: float) -> Optional[float]: if depth_m is None or not math.isfinite(depth_m) or bin_width_m <= 0: return None depth = max(float(depth_m), 0.0) return math.floor(depth / bin_width_m) * bin_width_m def lateral_interval_start(lateral_m: Optional[float], bin_width_m: float, lateral_limit_m: float) -> Optional[float]: if lateral_m is None or not math.isfinite(lateral_m) or bin_width_m <= 0: return None lateral = float(lateral_m) limit = float(lateral_limit_m) if lateral < -limit or lateral >= limit: return None return math.floor((lateral + limit) / bin_width_m) * bin_width_m - limit def metric_interval_start(value: Optional[float], bin_width: float) -> Optional[float]: if value is None or not math.isfinite(value) or bin_width <= 0: return None metric = max(float(value), 0.0) return math.floor(metric / bin_width) * bin_width def format_interval_bound(value: float) -> str: if float(value).is_integer(): return str(int(value)) half_step = round(float(value) * 2.0) if math.isclose(float(value), half_step / 2.0, rel_tol=0.0, abs_tol=1e-9): return f"{float(value):.1f}" return f"{float(value):.2f}" def interval_label(start_m: float, bin_width_m: float, unit: str = "m") -> str: end_m = start_m + bin_width_m return f"[{format_interval_bound(start_m)},{format_interval_bound(end_m)}){unit}" def interval_slug(start: float, bin_width: float, unit: str) -> str: start_text = format_interval_bound(start).replace("-", "neg").replace(".", "p") end_text = format_interval_bound(start + bin_width).replace("-", "neg").replace(".", "p") return sanitize_name(f"{start_text}_{end_text}{unit}") def make_interval_store() -> defaultdict[tuple[int, str, float], list[tuple[float, Optional[float]]]]: return defaultdict(list) def add_interval_store_sample( store: defaultdict[tuple[int, str, float], list[tuple[float, Optional[float]]]], cls_id: int, cls_name: str, start_value: Optional[float], value: Optional[float], relative_reference_value: Optional[float] = None, relative_reference_abs: bool = False, ) -> None: if start_value is None or value is None or not math.isfinite(value): return relative_reference = to_float(relative_reference_value) if relative_reference is not None and math.isfinite(relative_reference): relative_reference = abs(float(relative_reference)) if relative_reference_abs else float(relative_reference) else: relative_reference = None store[(int(cls_id), str(cls_name), float(start_value))].append((float(value), relative_reference)) def add_interval_sample( store: defaultdict[tuple[int, str, float], list[tuple[float, Optional[float]]]], cls_id: int, cls_name: str, depth_m: Optional[float], value: Optional[float], bin_width_m: float, relative_reference_value: Optional[float] = None, relative_reference_abs: bool = False, ) -> None: start_m = depth_interval_start(depth_m, bin_width_m) add_interval_store_sample( store, cls_id=cls_id, cls_name=cls_name, start_value=start_m, value=value, relative_reference_value=relative_reference_value, relative_reference_abs=relative_reference_abs, ) def add_lateral_interval_sample( store: defaultdict[tuple[int, str, float], list[tuple[float, Optional[float]]]], cls_id: int, cls_name: str, lateral_m: Optional[float], value: Optional[float], bin_width_m: float, lateral_limit_m: float, relative_reference_value: Optional[float] = None, relative_reference_abs: bool = False, ) -> None: start_m = lateral_interval_start(lateral_m, bin_width_m, lateral_limit_m) add_interval_store_sample( store, cls_id=cls_id, cls_name=cls_name, start_value=start_m, value=value, relative_reference_value=relative_reference_value, relative_reference_abs=relative_reference_abs, ) def add_heading_interval_sample( store: defaultdict[tuple[int, str, float], list[tuple[float, Optional[float]]]], cls_id: int, cls_name: str, heading_deg: Optional[float], value: Optional[float], bin_width_deg: float, relative_reference_value: Optional[float] = None, relative_reference_abs: bool = False, ) -> None: start_deg = heading_interval_start(heading_deg, bin_width_deg) add_interval_store_sample( store, cls_id=cls_id, cls_name=cls_name, start_value=start_deg, value=value, relative_reference_value=relative_reference_value, relative_reference_abs=relative_reference_abs, ) def build_interval_rows( store: defaultdict[tuple[int, str, float], list[tuple[float, Optional[float]]]], bin_width_m: float, value_name: str, threshold: Optional[float] = None, relative_value_name: Optional[str] = None, relative_reference_abs: bool = False, interval_prefix: str = "depth_bin", interval_unit: str = "m", relative_reference_label: str = "gt_bin", relative_reference_unit: str = "m", relative_reference_field: str = "relative_gt_bin_m", ) -> list[dict[str, Any]]: rows = [] for (cls_id, cls_name, start_m), values in sorted(store.items(), key=lambda item: (item[0][0], item[0][2])): sample_values: list[float] = [] relative_values: list[float] = [] relative_references: list[float] = [] bin_center_m = float(start_m + bin_width_m / 2.0) for value_item in values: if isinstance(value_item, tuple): sample_value = to_float(value_item[0]) explicit_relative_reference = to_float(value_item[1]) else: sample_value = to_float(value_item) explicit_relative_reference = None if sample_value is None or not math.isfinite(sample_value): continue sample_values.append(float(sample_value)) if relative_value_name: effective_relative_reference = explicit_relative_reference if effective_relative_reference is None: effective_relative_reference = abs(bin_center_m) if relative_reference_abs else bin_center_m if effective_relative_reference is not None and math.isfinite(effective_relative_reference): effective_relative_reference = float(effective_relative_reference) relative_references.append(effective_relative_reference) if not math.isclose(effective_relative_reference, 0.0, rel_tol=0.0, abs_tol=1e-9): relative_values.append(float(sample_value) * 100.0 / effective_relative_reference) if not sample_values: continue arr = np.asarray(sample_values, dtype=np.float64) bin_center_m = float(start_m + bin_width_m / 2.0) interval_label_text = interval_label(float(start_m), float(bin_width_m), unit=interval_unit) row = { "cls_id": int(cls_id), "cls_name": str(cls_name), "interval_start": float(start_m), "interval_end": float(start_m + bin_width_m), "interval_center": bin_center_m, "interval_label": interval_label_text, f"{interval_prefix}_start_{interval_unit}": float(start_m), f"{interval_prefix}_end_{interval_unit}": float(start_m + bin_width_m), f"{interval_prefix}_center_{interval_unit}": bin_center_m, f"{interval_prefix}_label": interval_label_text, "count": int(arr.size), f"mean_{value_name}": float(np.mean(arr)), f"median_{value_name}": float(np.median(arr)), f"p90_{value_name}": float(np.percentile(arr, 90)), f"max_{value_name}": float(np.max(arr)), } if relative_value_name: relative_arr = np.asarray(relative_values, dtype=np.float64) relative_reference_value = mean_or_none(relative_references) row["relative_reference_value"] = relative_reference_value row["relative_reference_label"] = str(relative_reference_label) row["relative_reference_unit"] = str(relative_reference_unit) row[relative_reference_field] = relative_reference_value if relative_arr.size > 0: row[f"mean_{relative_value_name}"] = float(np.mean(relative_arr)) row[f"median_{relative_value_name}"] = float(np.median(relative_arr)) row[f"p90_{relative_value_name}"] = float(np.percentile(relative_arr, 90)) row[f"max_{relative_value_name}"] = float(np.max(relative_arr)) else: row[f"mean_{relative_value_name}"] = None row[f"median_{relative_value_name}"] = None row[f"p90_{relative_value_name}"] = None row[f"max_{relative_value_name}"] = None if threshold is not None: row[f"rate_gt_{str(threshold).replace('.', 'p')}"] = float(np.mean(arr > threshold)) row[f"count_gt_{str(threshold).replace('.', 'p')}"] = int(np.sum(arr > threshold)) rows.append(row) return rows def build_metric_bucket_interval_rows( store: defaultdict[float, dict[str, Any]], bin_width_m: float, prefix: str = "lateral_bin", ) -> list[dict[str, Any]]: rows = [] for start_m, bucket in sorted(store.items(), key=lambda item: float(item[0])): summary = summarize_metric_bucket(bucket) rows.append( { f"{prefix}_start_m": float(start_m), f"{prefix}_end_m": float(start_m + bin_width_m), f"{prefix}_label": interval_label(float(start_m), float(bin_width_m)), **summary, } ) return rows def build_grouped_metric_bucket_interval_rows( stores_by_group: dict[str, defaultdict[float, dict[str, Any]]], bin_width_m: float, prefix: str = "lateral_bin", group_order: Optional[Iterable[str]] = None, ) -> dict[str, list[dict[str, Any]]]: ordered_groups = [str(group) for group in (group_order or stores_by_group.keys())] rows_by_group: dict[str, list[dict[str, Any]]] = {} for group_name in ordered_groups: rows_by_group[group_name] = build_metric_bucket_interval_rows( stores_by_group.get(group_name, defaultdict(make_metric_bucket)), bin_width_m=bin_width_m, prefix=prefix, ) return rows_by_group def flatten_grouped_metric_bucket_interval_rows( rows_by_group: dict[str, list[dict[str, Any]]], group_key: str = "face_visibility_bucket", ) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for group_name, group_rows in rows_by_group.items(): for row in group_rows: rows.append({group_key: str(group_name), **row}) return rows def write_rows_csv(path: Path, rows: list[dict[str, Any]], fallback_fields: Optional[list[str]] = None) -> None: path.parent.mkdir(parents=True, exist_ok=True) if not rows: with path.open("w", encoding="utf-8", newline="") as file: writer = csv.writer(file) writer.writerow(fallback_fields or []) return helper_fields = { "interval_start", "interval_end", "interval_center", "interval_label", "relative_reference_value", "relative_reference_label", "relative_reference_unit", } fieldnames = [field for field in rows[0].keys() if field not in helper_fields] with path.open("w", encoding="utf-8", newline="") as file: writer = csv.DictWriter(file, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() writer.writerows(rows) def group_rows_by_class(rows: list[dict[str, Any]]) -> dict[tuple[int, str], list[dict[str, Any]]]: grouped: dict[tuple[int, str], list[dict[str, Any]]] = defaultdict(list) for row in rows: grouped[(int(row["cls_id"]), str(row["cls_name"]))].append(row) return grouped def top_interval_rows(rows: list[dict[str, Any]], metric_key: str, topn: int = 3, min_count: int = 30) -> list[dict[str, Any]]: filtered = [row for row in rows if row.get(metric_key) is not None and int(row.get("count", 0)) >= min_count] if not filtered: filtered = [row for row in rows if row.get(metric_key) is not None] return sorted(filtered, key=lambda row: (float(row.get(metric_key) or 0.0), int(row.get("count", 0))), reverse=True)[:topn] def make_metric_bucket() -> dict[str, Any]: return { "gt_total": 0, "pred_total": 0, "matched_2d": 0, "matched_3d": 0, "matched_pos": 0, "yaw_compare_eligible_count": 0, "yaw_compare_pair_count": 0, "confidence": [], "match_iou": [], "yaw_abs_deg": [], "x_abs_m": [], "y_abs_m": [], "z_abs_m": [], "center_error_m": [], "yaw_compare_yaw_abs_deg": [], "direct_visible_yaw_abs_deg": [], "edge_visible_yaw_abs_deg": [], "direct_minus_edge_visible_yaw_abs_deg": [], "length_compare_pair_count": 0, "direct_length_abs_err_m": [], "edge_length_abs_err_m": [], "direct_minus_edge_length_abs_err_m": [], "length_compare_edge_better_count": 0, "length_compare_direct_better_count": 0, "length_compare_tie_count": 0, "yaw_compare_edge_better_count": 0, "yaw_compare_direct_better_count": 0, "yaw_compare_tie_count": 0, "yaw_bad_count": 0, "x_bad_count": 0, "y_bad_count": 0, "z_bad_count": 0, "yaw_abs_sum": 0.0, } def add_prediction_count(bucket: dict[str, Any], count_value: int = 1) -> None: bucket["pred_total"] += int(count_value) def add_gt_count(bucket: dict[str, Any], count_value: int = 1) -> None: bucket["gt_total"] += int(count_value) def add_2d_match(bucket: dict[str, Any]) -> None: bucket["matched_2d"] += 1 def add_3d_record( bucket: dict[str, Any], record: dict[str, Any], yaw_bad_threshold_deg: float, horizontal_bad_threshold_m: float, vertical_bad_threshold_m: float, edge_visible_key: str = "edge_visible_yaw_abs_deg", direct_minus_edge_key: str = "direct_minus_edge_visible_yaw_abs_deg", ) -> None: bucket["matched_3d"] += 1 bucket["confidence"].append(record["confidence"]) bucket["match_iou"].append(record["match_iou"]) yaw_abs = record["yaw_abs_deg"] if yaw_abs is not None: bucket["yaw_abs_deg"].append(yaw_abs) bucket["yaw_abs_sum"] += yaw_abs if yaw_abs >= yaw_bad_threshold_deg: bucket["yaw_bad_count"] += 1 direct_visible = record["direct_visible_yaw_abs_deg"] edge_visible = record.get(edge_visible_key) if record["yaw_compare_eligible"]: bucket["yaw_compare_eligible_count"] += 1 if direct_visible is not None and edge_visible is not None: bucket["yaw_compare_pair_count"] += 1 if yaw_abs is not None: bucket["yaw_compare_yaw_abs_deg"].append(yaw_abs) bucket["direct_visible_yaw_abs_deg"].append(direct_visible) bucket["edge_visible_yaw_abs_deg"].append(edge_visible) direct_minus_edge = record.get(direct_minus_edge_key) if direct_minus_edge is None: direct_minus_edge = float(direct_visible) - float(edge_visible) bucket["direct_minus_edge_visible_yaw_abs_deg"].append(direct_minus_edge) eps = 1e-6 if edge_visible + eps < direct_visible: bucket["yaw_compare_edge_better_count"] += 1 elif direct_visible + eps < edge_visible: bucket["yaw_compare_direct_better_count"] += 1 else: bucket["yaw_compare_tie_count"] += 1 direct_length_err = record.get("direct_length_abs_err_m") edge_length_err = record.get("edge_length_abs_err_m") if direct_length_err is not None and edge_length_err is not None: bucket["length_compare_pair_count"] += 1 bucket["direct_length_abs_err_m"].append(float(direct_length_err)) bucket["edge_length_abs_err_m"].append(float(edge_length_err)) direct_minus_edge_length = record.get("direct_minus_edge_length_abs_err_m") if direct_minus_edge_length is None: direct_minus_edge_length = float(direct_length_err) - float(edge_length_err) bucket["direct_minus_edge_length_abs_err_m"].append(float(direct_minus_edge_length)) eps = 1e-6 if edge_length_err + eps < direct_length_err: bucket["length_compare_edge_better_count"] += 1 elif direct_length_err + eps < edge_length_err: bucket["length_compare_direct_better_count"] += 1 else: bucket["length_compare_tie_count"] += 1 if record["position_eligible"]: bucket["matched_pos"] += 1 x_abs = record["x_abs_m"] y_abs = record["y_abs_m"] z_abs = record["z_abs_m"] center = record["center_error_m"] if x_abs is not None: bucket["x_abs_m"].append(x_abs) if x_abs > horizontal_bad_threshold_m: bucket["x_bad_count"] += 1 if y_abs is not None: bucket["y_abs_m"].append(y_abs) if z_abs is not None: bucket["z_abs_m"].append(z_abs) if z_abs > vertical_bad_threshold_m: bucket["z_bad_count"] += 1 if center is not None: bucket["center_error_m"].append(center) def mean_or_none(values: list[float]) -> Optional[float]: if not values: return None return float(np.mean(np.asarray(values, dtype=np.float64))) def percentile_or_none(values: list[float], q: float) -> Optional[float]: if not values: return None return float(np.percentile(np.asarray(values, dtype=np.float64), q)) def rate_or_none(numer: int, denom: int) -> Optional[float]: if denom <= 0: return None return float(numer) / float(denom) def f1_or_none(precision: Optional[float], recall: Optional[float]) -> Optional[float]: if precision is None or recall is None: return None denom = float(precision) + float(recall) if denom <= 0.0: return None return (2.0 * float(precision) * float(recall)) / denom def summarize_metric_bucket(bucket: dict[str, Any]) -> dict[str, Any]: matched_3d = int(bucket["matched_3d"]) matched_pos = int(bucket["matched_pos"]) yaw_compare_count = int(bucket["yaw_compare_pair_count"]) length_compare_count = int(bucket["length_compare_pair_count"]) direct_regression_mae = mean_or_none(bucket["direct_visible_yaw_abs_deg"]) direct_regression_p90 = percentile_or_none(bucket["direct_visible_yaw_abs_deg"], 90.0) edge_based_mae = mean_or_none(bucket["edge_visible_yaw_abs_deg"]) edge_based_p90 = percentile_or_none(bucket["edge_visible_yaw_abs_deg"], 90.0) mean_direct_minus_edge = mean_or_none(bucket["direct_minus_edge_visible_yaw_abs_deg"]) median_direct_minus_edge = percentile_or_none(bucket["direct_minus_edge_visible_yaw_abs_deg"], 50.0) direct_length_mae = mean_or_none(bucket["direct_length_abs_err_m"]) edge_length_mae = mean_or_none(bucket["edge_length_abs_err_m"]) mean_direct_minus_edge_length = mean_or_none(bucket["direct_minus_edge_length_abs_err_m"]) precision_2d = rate_or_none(bucket["matched_2d"], bucket["pred_total"]) recall_2d = rate_or_none(bucket["matched_2d"], bucket["gt_total"]) return { "gt_total": int(bucket["gt_total"]), "pred_total": int(bucket["pred_total"]), "matched_2d": int(bucket["matched_2d"]), "matched_3d": matched_3d, "matched_pos": matched_pos, "yaw_compare_eligible_count": int(bucket["yaw_compare_eligible_count"]), "yaw_compare_count": yaw_compare_count, "length_compare_count": length_compare_count, "precision_2d": precision_2d, "recall_2d": recall_2d, "f1_2d": f1_or_none(precision_2d, recall_2d), "mean_confidence": mean_or_none(bucket["confidence"]), "mean_match_iou": mean_or_none(bucket["match_iou"]), "yaw_mae_deg": mean_or_none(bucket["yaw_abs_deg"]), "yaw_p90_deg": percentile_or_none(bucket["yaw_abs_deg"], 90.0), "x_abs_mae_m": mean_or_none(bucket["x_abs_m"]), "y_abs_mae_m": mean_or_none(bucket["y_abs_m"]), "z_abs_mae_m": mean_or_none(bucket["z_abs_m"]), "horizontal_abs_mae_m": mean_or_none(bucket["x_abs_m"]), "vertical_abs_mae_m": mean_or_none(bucket["z_abs_m"]), "center_error_mae_m": mean_or_none(bucket["center_error_m"]), "yaw_compare_subset_yaw_mae_deg": mean_or_none(bucket["yaw_compare_yaw_abs_deg"]), "yaw_compare_subset_yaw_p90_deg": percentile_or_none(bucket["yaw_compare_yaw_abs_deg"], 90.0), "direct_visible_yaw_mae_deg": direct_regression_mae, "direct_visible_yaw_p90_deg": direct_regression_p90, "edge_visible_yaw_mae_deg": edge_based_mae, "edge_visible_yaw_p90_deg": edge_based_p90, "direct_regression_yaw_mae_deg": direct_regression_mae, "direct_regression_yaw_p90_deg": direct_regression_p90, "edge_based_yaw_mae_deg": edge_based_mae, "edge_based_yaw_p90_deg": edge_based_p90, "mean_direct_minus_edge_visible_yaw_deg": mean_direct_minus_edge, "median_direct_minus_edge_visible_yaw_deg": median_direct_minus_edge, "mean_direct_minus_edge_yaw_deg": mean_direct_minus_edge, "median_direct_minus_edge_yaw_deg": median_direct_minus_edge, "direct_regression_length_mae_m": direct_length_mae, "side_edge_length_mae_m": edge_length_mae, "mean_direct_minus_side_edge_length_m": mean_direct_minus_edge_length, "length_compare_edge_better_count": int(bucket["length_compare_edge_better_count"]), "length_compare_direct_better_count": int(bucket["length_compare_direct_better_count"]), "length_compare_tie_count": int(bucket["length_compare_tie_count"]), "yaw_compare_edge_better_count": int(bucket["yaw_compare_edge_better_count"]), "yaw_compare_direct_better_count": int(bucket["yaw_compare_direct_better_count"]), "yaw_compare_tie_count": int(bucket["yaw_compare_tie_count"]), "yaw_bad_count": int(bucket["yaw_bad_count"]), "x_bad_count": int(bucket["x_bad_count"]), "y_bad_count": int(bucket["y_bad_count"]), "z_bad_count": int(bucket["z_bad_count"]), "horizontal_bad_count": int(bucket["x_bad_count"]), "vertical_bad_count": int(bucket["z_bad_count"]), "yaw_bad_rate": rate_or_none(bucket["yaw_bad_count"], matched_3d), "x_bad_rate": rate_or_none(bucket["x_bad_count"], matched_pos), "y_bad_rate": rate_or_none(bucket["y_bad_count"], matched_pos), "z_bad_rate": rate_or_none(bucket["z_bad_count"], matched_pos), "horizontal_bad_rate": rate_or_none(bucket["x_bad_count"], matched_pos), "vertical_bad_rate": rate_or_none(bucket["z_bad_count"], matched_pos), "length_compare_edge_better_rate": rate_or_none(bucket["length_compare_edge_better_count"], length_compare_count), "length_compare_direct_better_rate": rate_or_none(bucket["length_compare_direct_better_count"], length_compare_count), "length_compare_tie_rate": rate_or_none(bucket["length_compare_tie_count"], length_compare_count), "yaw_compare_edge_better_rate": rate_or_none(bucket["yaw_compare_edge_better_count"], yaw_compare_count), "yaw_compare_direct_better_rate": rate_or_none(bucket["yaw_compare_direct_better_count"], yaw_compare_count), "yaw_compare_tie_rate": rate_or_none(bucket["yaw_compare_tie_count"], yaw_compare_count), "yaw_abs_sum": float(bucket["yaw_abs_sum"]), } def make_breakdown_buckets() -> dict[str, dict[str, dict[str, Any]]]: return { "cut_status": defaultdict(make_metric_bucket), "face_visibility": defaultdict(make_metric_bucket), "distance_bin": defaultdict(make_metric_bucket), "bbox_diag_bin": defaultdict(make_metric_bucket), "class_group": defaultdict(make_metric_bucket), } def is_signed_lateral_yaw_compare_longitudinal_eligible( record: dict[str, Any], max_longitudinal_dist_m: float = DEFAULT_YAW_COMPARE_MAX_LONGITUDINAL_DIST_M, ) -> bool: longitudinal_m = to_float(record.get("gt_z_m")) if longitudinal_m is None: return False return 0.0 <= float(longitudinal_m) < float(max_longitudinal_dist_m) def make_reservoir_store() -> dict[str, Any]: return {"seen": 0, "records": []} def reservoir_add( store: dict[str, Any], record: dict[str, Any], capacity: int, rng: random.Random, ) -> None: if capacity <= 0: return store["seen"] = int(store.get("seen", 0)) + 1 copied = dict(record) records = store.setdefault("records", []) if len(records) < capacity: records.append(copied) return index = rng.randrange(int(store["seen"])) if index < capacity: records[index] = copied def reservoir_records(store: dict[str, Any], rng: random.Random) -> list[dict[str, Any]]: records = [dict(record) for record in store.get("records", [])] rng.shuffle(records) return records def make_interval_visual_reservoirs() -> defaultdict[float, dict[str, Any]]: return defaultdict(make_reservoir_store) def add_interval_visual_record( stores: defaultdict[float, dict[str, Any]], value: Optional[float], record: dict[str, Any], bin_width: float, per_bin_capacity: int, rng: random.Random, ) -> None: start = metric_interval_start(value, bin_width) if start is None: return reservoir_add(stores[float(start)], record, per_bin_capacity, rng) def make_writer(path: Path, fieldnames: Optional[list[str]] = None) -> tuple[Any, csv.DictWriter]: path.parent.mkdir(parents=True, exist_ok=True) handle = path.open("w", encoding="utf-8", newline="") writer = csv.DictWriter(handle, fieldnames=fieldnames or BADCASE_FIELDS) writer.writeheader() return handle, writer def record_to_csv_row(record: dict[str, Any]) -> dict[str, Any]: row = {key: record.get(key) for key in BADCASE_FIELDS} row["gt_bbox_xyxy"] = json.dumps(record.get("gt_bbox_xyxy")) row["pred_bbox_xyxy"] = json.dumps(record.get("pred_bbox_xyxy")) return row def record_2d_to_csv_row(record: dict[str, Any]) -> dict[str, Any]: row = {key: record.get(key) for key in BADCASE_2D_FIELDS} row["bbox_xyxy"] = json.dumps(record.get("bbox_xyxy")) return row def fake_class_record_to_csv_row(record: dict[str, Any]) -> dict[str, Any]: row = {key: record.get(key) for key in FAKE_CLASS_BADCASE_FIELDS} row["gt_bbox_xyxy"] = json.dumps(record.get("gt_bbox_xyxy")) row["pred_bbox_xyxy"] = json.dumps(record.get("pred_bbox_xyxy")) return row def face_selection_record_to_csv_row(record: dict[str, Any]) -> dict[str, Any]: row = {key: record.get(key) for key in FACE_SELECTION_BADCASE_FIELDS} row["gt_bbox_xyxy"] = json.dumps(record.get("gt_bbox_xyxy")) row["pred_bbox_xyxy"] = json.dumps(record.get("pred_bbox_xyxy")) return row def build_prediction_focus_mask( detections: np.ndarray, preds_3d: np.ndarray, anchors: np.ndarray, strides: np.ndarray, calib: dict[str, Any], max_abs_longitudinal_m: float, ) -> np.ndarray: if len(detections) == 0: return np.zeros((0,), dtype=bool) focus_mask = np.zeros((len(detections),), dtype=bool) for pred_index in range(len(detections)): attrs = extract_3d_attrs_from_prediction( preds_3d[pred_index], anchors[:, pred_index], float(strides[pred_index]), calib, ) center = None if attrs is None else attrs.get("center") if center is None: continue center = np.asarray(center, dtype=np.float32).reshape(-1) lateral_m = to_float(center[0]) if center.size > 0 else None longitudinal_m = to_float(center[2]) if center.size > 2 else None focus_mask[pred_index] = is_focused_confusion_spatial_object( lateral_m=lateral_m, longitudinal_m=longitudinal_m, max_abs_longitudinal_m=max_abs_longitudinal_m, ) return focus_mask def iter_roi_analysis_samples( bundle, entries: list[tuple[str, str]], image_root: Path, class_map: dict[str, int], difficulty_weights: list[float], face_3d_classes: set[int], complete_3d_classes: set[int], classes: Optional[set[int]], min_wh_px: float, inference_batch_size: int, ) -> Iterable[tuple[dict[str, Any], Any]]: batch_size = max(1, int(inference_batch_size)) for indexed_entry_batch in iter_batches(enumerate(entries), batch_size): batch_items = [] for sample_index, entry in indexed_entry_batch: image_path = entry_to_image_file(entry, image_root) label_path = entry_to_label_file(entry) image_bgr = cv2.imread(str(image_path), cv2.IMREAD_COLOR) if image_bgr is None: continue ori_h, ori_w = image_bgr.shape[:2] raw_calib = read_raw_calib_from_label(image_path, label_path) prepared = prepare_roi_image(image_bgr, raw_calib, bundle.spec, bundle.imgsz) gt = prepare_gt_for_roi( label_file=label_path, ori_w=ori_w, ori_h=ori_h, prepared=prepared, bundle=bundle, class_map=class_map, difficulty_weights=difficulty_weights, face_3d_classes=face_3d_classes, complete_3d_classes=complete_3d_classes, include_classes=classes, min_wh_px=min_wh_px, ) batch_items.append( { "sample_index": int(sample_index), "image_path": image_path, "label_path": label_path, "prepared": prepared, "gt": gt, } ) if not batch_items: continue raw_outputs_batch = run_model_for_prepared_roi_batch( bundle=bundle, prepared_batch=[item["prepared"] for item in batch_items], ) for batch_item, raw_outputs in zip(batch_items, raw_outputs_batch): yield batch_item, raw_outputs def collect_2d_confidence_advice( bundle, entries: list[tuple[str, str]], image_root: Path, class_map: dict[str, int], difficulty_weights: list[float], face_3d_classes: set[int], complete_3d_classes: set[int], classes: Optional[set[int]], min_wh_px: float, names_dict: dict[int, str], roi_name: str, roi_output: Path, topk_badcases: int, badcase_random_seed: int, focused_confusion_max_abs_longitudinal_m: float, threshold_search_conf: float, configured_confidence_2d: float, inference_batch_size: int, log_every: int, ) -> tuple[dict[str, Any], dict[str, Any], float]: ap_store_2d = make_2d_ap_store() total_samples = len(entries) start_time = time.time() for batch_item, raw_outputs in iter_roi_analysis_samples( bundle=bundle, entries=entries, image_root=image_root, class_map=class_map, difficulty_weights=difficulty_weights, face_3d_classes=face_3d_classes, complete_3d_classes=complete_3d_classes, classes=classes, min_wh_px=min_wh_px, inference_batch_size=inference_batch_size, ): sample_index = int(batch_item["sample_index"]) prepared = batch_item["prepared"] gt = batch_item["gt"] analysis_outputs = filter_prediction_outputs( raw_outputs=raw_outputs, conf_thres=float(threshold_search_conf), max_det=bundle.spec.max_det, classes=classes, ) analysis_detections = analysis_outputs.detections analysis_pred_boxes = ( np.asarray(analysis_detections[:, :4], dtype=np.float32) if len(analysis_detections) else np.zeros((0, 4), dtype=np.float32) ) analysis_pred_cls = ( np.asarray(analysis_detections[:, 5], dtype=np.int32).reshape(-1) if len(analysis_detections) else np.zeros((0,), dtype=np.int32) ) analysis_pred_conf = ( np.asarray(analysis_detections[:, 4], dtype=np.float32).reshape(-1) if len(analysis_detections) else np.zeros((0,), dtype=np.float32) ) add_2d_ap_sample( ap_store_2d, gt_cls=gt["classes"], gt_boxes=gt["boxes_xyxy"], pred_cls=analysis_pred_cls, pred_boxes=analysis_pred_boxes, pred_conf=analysis_pred_conf, iou_thresholds=np.linspace(0.5, 0.95, 10, dtype=np.float32), ) if (sample_index + 1) % max(1, log_every) == 0 or (sample_index + 1) == total_samples: elapsed = time.time() - start_time per_sample = elapsed / max(sample_index + 1, 1) remaining = total_samples - (sample_index + 1) eta = remaining * per_sample print( f"[{roi_name}:advice] {sample_index + 1}/{total_samples} " f"elapsed={elapsed / 60:.1f}m eta={eta / 60:.1f}m search_conf={threshold_search_conf:.3f}" ) confidence_curve_output = roi_output / "2d_confidence_curves" confidence_curve_output.mkdir(parents=True, exist_ok=True) ap_summary_2d = summarize_2d_ap_store( ap_store_2d, plot=True, save_dir=confidence_curve_output, names=names_dict, prefix="2d_", ) threshold_advice_2d = ap_summary_2d.get("threshold_advice") or {} report_confidence_2d = to_float(threshold_advice_2d.get("recommended_confidence")) if report_confidence_2d is None: report_confidence_2d = float(configured_confidence_2d) print( f"[{roi_name}:advice] configured_conf={configured_confidence_2d:.3f} " f"recommended_conf={float(report_confidence_2d):.3f}" ) return ap_summary_2d, threshold_advice_2d, float(report_confidence_2d) def prepare_gt_for_roi( label_file: Path, ori_w: int, ori_h: int, prepared, bundle, class_map: dict[str, int], difficulty_weights: list[float], face_3d_classes: set[int], complete_3d_classes: set[int], include_classes: Optional[set[int]], min_wh_px: float, ) -> dict[str, Any]: detailed_class_names = read_mapped_label_class_names(label_file, class_map) lb_2d, lb_3d = parse_ground_3d_label_file( str(label_file), class_map, difficulty_weights, face_3d_classes, complete_3d_classes, min_wh=0.0, ) detailed_class_names = filter_label_names_by_classes(detailed_class_names, lb_2d["cls"], include_classes) lb_2d, lb_3d = filter_labels_by_classes(lb_2d, lb_3d, include_classes) lb_2d, lb_3d = remap_labels_to_roi(lb_2d, lb_3d, ori_w, ori_h, prepared.crop_bounds) detailed_class_names = filter_label_names_by_min_wh( detailed_class_names, lb_2d["bboxes"], img_w=prepared.image.shape[1], img_h=prepared.image.shape[0], min_wh_px=float(min_wh_px), ) lb_2d, lb_3d = filter_labels_by_min_wh( lb_2d, lb_3d, img_w=prepared.image.shape[1], img_h=prepared.image.shape[0], min_wh_px=float(min_wh_px), ) lb_3d = normalize_roi_depth(lb_3d, prepared.calib["fx"], bundle.spec.virtual_fx) gt_boxes_xyxy = xywhn_to_xyxy(lb_2d["bboxes"], prepared.image.shape[1], prepared.image.shape[0]) gt_classes = lb_2d["cls"].reshape(-1).astype(np.int32) if len(detailed_class_names) != len(gt_classes): detailed_class_names = [get_cls_name(bundle.names, int(cls_id)) for cls_id in gt_classes.tolist()] return { "lb_2d": lb_2d, "lb_3d": lb_3d, "boxes_xyxy": gt_boxes_xyxy, "classes": gt_classes, "class_names": detailed_class_names, } def make_record( sample_index: int, roi_name: str, image_path: Path, label_path: Path, cls_name: str, gt_index: int, pred_index: int, gt_box: np.ndarray, pred_box: np.ndarray, match_iou: float, prediction: dict[str, Any], gt_attrs: dict[str, Any], pred_attrs: dict[str, Any], gt_position_attrs: Optional[dict[str, Any]], pred_position_attrs: Optional[dict[str, Any]], gt_decoded: dict[str, Any], pred_decoded: dict[str, Any], gt_visible_faces: list[tuple[int, np.ndarray]], gt_visible_yaw: Optional[float], pred_edge_visible_yaw: Optional[float], pred_edge_bucket_visible_yaw: Optional[float] = None, pred_edge_decoded: Optional[dict[str, Any]] = None, pred_edge_box: Optional[dict[str, Any]] = None, is_cut_object_flag: bool = False, position_eligible: bool = False, position_error_basis: Optional[str] = None, yaw_compare_max_lateral_dist_m: float = 5.0, pred_yaw_compare_face_types: Optional[Iterable[int]] = None, pred_yaw_compare_valid: bool = False, ) -> dict[str, Any]: gt_center = np.asarray(gt_attrs["center"], dtype=np.float32) pred_center = np.asarray(pred_attrs["center"], dtype=np.float32) gt_position_source = gt_position_attrs or gt_attrs pred_position_source = pred_position_attrs or pred_attrs gt_position_center = np.asarray(gt_position_source["center"], dtype=np.float32) pred_position_center = np.asarray(pred_position_source["center"], dtype=np.float32) position_error_basis = str( position_error_basis or ( "face_center" if gt_position_source.get("face_center") is not None and pred_position_source.get("face_center") is not None else "box_center" ) ) gt_lateral_abs = to_float(abs(gt_center[0])) x_abs = to_float(abs(pred_position_center[0] - gt_position_center[0])) y_abs = to_float(abs(pred_position_center[1] - gt_position_center[1])) z_abs = to_float(abs(pred_position_center[2] - gt_position_center[2])) center_error = None if np.all(np.isfinite(gt_position_center)) and np.all(np.isfinite(pred_position_center)): center_error = float(np.linalg.norm(pred_position_center - gt_position_center)) gt_yaw = to_float(gt_attrs["yaw"]) pred_yaw = to_float(pred_attrs["yaw"]) gt_visible_yaw = to_float(gt_visible_yaw) pred_edge_visible_yaw = to_float(pred_edge_visible_yaw) pred_edge_bucket_visible_yaw = to_float(pred_edge_bucket_visible_yaw) direct_visible_yaw_abs_deg = angle_abs_deg(pred_yaw, gt_yaw) edge_visible_yaw_abs_deg = angle_abs_deg(pred_edge_visible_yaw, gt_yaw) direct_minus_edge_visible_yaw_abs_deg = None if direct_visible_yaw_abs_deg is not None and edge_visible_yaw_abs_deg is not None: direct_minus_edge_visible_yaw_abs_deg = float(direct_visible_yaw_abs_deg) - float(edge_visible_yaw_abs_deg) edge_bucket_visible_yaw_abs_deg = angle_abs_deg(pred_edge_bucket_visible_yaw, gt_yaw) direct_minus_edge_bucket_visible_yaw_abs_deg = None if direct_visible_yaw_abs_deg is not None and edge_bucket_visible_yaw_abs_deg is not None: direct_minus_edge_bucket_visible_yaw_abs_deg = float(direct_visible_yaw_abs_deg) - float(edge_bucket_visible_yaw_abs_deg) yaw_compare_eligible = bool(gt_lateral_abs is not None and gt_lateral_abs < float(yaw_compare_max_lateral_dist_m)) visible_face_names = [FACE_NAMES.get(face_type, str(face_type)) for face_type, _ in gt_visible_faces] pred_compare_face_types = tuple(int(face_type) for face_type in (pred_yaw_compare_face_types or ())) pred_compare_face_names = [FACE_NAMES.get(face_type, str(face_type)) for face_type in pred_compare_face_types] pred_compare_has_side_face_visible = bool(any(face_type in (2, 3) for face_type in pred_compare_face_types)) pred_compare_bucket = classify_edge_yaw_prediction_bucket(pred_compare_face_types, pred_yaw_compare_valid) bbox_w = float(max(gt_box[2] - gt_box[0], 0.0)) bbox_h = float(max(gt_box[3] - gt_box[1], 0.0)) bbox_diag = math.hypot(bbox_w, bbox_h) gt_length_m = to_float(np.asarray(gt_attrs["dims"], dtype=np.float32)[0]) pred_direct_length_m = to_float(np.asarray(pred_attrs["dims"], dtype=np.float32)[0]) pred_edge_length_m = None pred_edge_box_mode = None if pred_edge_box is None else pred_edge_box.get("mode") if pred_edge_box is not None and pred_edge_box.get("dims") is not None and pred_edge_box_mode in ("two-face", "side"): pred_edge_length_m = to_float(np.asarray(pred_edge_box["dims"], dtype=np.float32)[0]) direct_length_abs_err_m = None edge_length_abs_err_m = None direct_minus_edge_length_abs_err_m = None if gt_length_m is not None and pred_direct_length_m is not None: direct_length_abs_err_m = abs(float(pred_direct_length_m) - float(gt_length_m)) if gt_length_m is not None and pred_edge_length_m is not None: edge_length_abs_err_m = abs(float(pred_edge_length_m) - float(gt_length_m)) if direct_length_abs_err_m is not None and edge_length_abs_err_m is not None: direct_minus_edge_length_abs_err_m = float(direct_length_abs_err_m) - float(edge_length_abs_err_m) gt_visible_face_type = None if gt_position_source.get("visible_face_type") is None else int(gt_position_source["visible_face_type"]) pred_visible_face_type = None if pred_position_source.get("visible_face_type") is None else int(pred_position_source["visible_face_type"]) gt_face_center = gt_position_source.get("face_center") if gt_position_source.get("face_center") is not None else gt_attrs.get("face_center") pred_face_center = ( pred_position_source.get("face_center") if pred_position_source.get("face_center") is not None else pred_attrs.get("face_center") ) gt_visible_face_types = () if gt_decoded is None else tuple(gt_decoded.get("visible_face_types", ()) or ()) pred_visible_face_types = () if pred_decoded is None else tuple(pred_decoded.get("visible_face_types", ()) or ()) pred_edge_visible_face_type = None if pred_edge_decoded is None else pred_edge_decoded.get("visible_face_type") pred_edge_visible_face_types = () if pred_edge_decoded is None else tuple(pred_edge_decoded.get("visible_face_types", ()) or ()) pred_edge_corners = None if pred_edge_decoded is None else pred_edge_decoded.get("corners_3d") pred_edge_points_2d = None if pred_edge_decoded is None else pred_edge_decoded.get("edge_points_2d") pred_edge_dims = None if pred_edge_box is None else pred_edge_box.get("dims") gt_decoded_visible_face_type = None if gt_decoded is None else gt_decoded.get("visible_face_type") pred_decoded_visible_face_type = None if pred_decoded is None else pred_decoded.get("visible_face_type") reuse_gt_decoded_face_overlay = gt_visible_face_type == gt_decoded_visible_face_type reuse_pred_decoded_face_overlay = pred_visible_face_type == pred_decoded_visible_face_type return { "sample_index": int(sample_index), "roi": roi_name, "frame_name": image_path.name, "image_path": str(image_path), "label_path": str(label_path), "cls_id": int(prediction["cls_id"]), "cls_name": cls_name, "gt_index": int(gt_index), "pred_index": int(pred_index), "confidence": float(prediction["confidence"]), "match_iou": float(match_iou), "bbox_diag_px": float(bbox_diag), "bbox_diag_bin": bbox_diag_bin(bbox_diag), "distance_bin": depth_bin(to_float(gt_center[2])), "is_cut_object": bool(is_cut_object_flag), "position_eligible": bool(position_eligible), "position_error_basis": position_error_basis, "yaw_compare_eligible": yaw_compare_eligible, "visible_face_count": int(len(gt_visible_faces)), "visible_faces": ",".join(visible_face_names) if visible_face_names else "none", "has_side_face_visible": bool(any(face_type in (2, 3) for face_type, _ in gt_visible_faces)), "yaw_compare_visible_face_count": int(len(pred_compare_face_types)), "yaw_compare_visible_faces": ",".join(pred_compare_face_names) if pred_compare_face_names else "none", "yaw_compare_has_side_face_visible": pred_compare_has_side_face_visible, "yaw_compare_face_bucket": pred_compare_bucket, "yaw_compare_pred_valid": bool(pred_yaw_compare_valid), "yaw_abs_deg": angle_abs_deg(pred_yaw, gt_yaw), "direct_visible_yaw_abs_deg": direct_visible_yaw_abs_deg, "edge_visible_yaw_abs_deg": edge_visible_yaw_abs_deg, "direct_minus_edge_visible_yaw_abs_deg": direct_minus_edge_visible_yaw_abs_deg, "edge_bucket_visible_yaw_abs_deg": edge_bucket_visible_yaw_abs_deg, "direct_minus_edge_bucket_visible_yaw_abs_deg": direct_minus_edge_bucket_visible_yaw_abs_deg, "gt_length_m": gt_length_m, "pred_direct_length_m": pred_direct_length_m, "pred_edge_length_m": pred_edge_length_m, "pred_edge_box_mode": pred_edge_box_mode, "direct_length_abs_err_m": direct_length_abs_err_m, "edge_length_abs_err_m": edge_length_abs_err_m, "direct_minus_edge_length_abs_err_m": direct_minus_edge_length_abs_err_m, "gt_lateral_abs_m": gt_lateral_abs, "x_abs_m": x_abs, "y_abs_m": y_abs, "z_abs_m": z_abs, "center_error_m": to_float(center_error), "position_gt_x_m": to_float(gt_position_center[0]), "position_pred_x_m": to_float(pred_position_center[0]), "position_gt_y_m": to_float(gt_position_center[1]), "position_pred_y_m": to_float(pred_position_center[1]), "position_gt_z_m": to_float(gt_position_center[2]), "position_pred_z_m": to_float(pred_position_center[2]), "gt_x_m": to_float(gt_center[0]), "pred_x_m": to_float(pred_center[0]), "gt_y_m": to_float(gt_center[1]), "pred_y_m": to_float(pred_center[1]), "gt_z_m": to_float(gt_center[2]), "pred_z_m": to_float(pred_center[2]), "gt_depth_m": to_float(gt_attrs["depth"]), "pred_depth_m": to_float(pred_attrs["depth"]), "gt_yaw_deg": to_float(math.degrees(gt_yaw)) if gt_yaw is not None else None, "pred_yaw_deg": to_float(math.degrees(pred_yaw)) if pred_yaw is not None else None, "gt_visible_yaw_deg": to_float(math.degrees(gt_visible_yaw)) if gt_visible_yaw is not None else None, "pred_edge_yaw_deg": to_float(math.degrees(pred_edge_visible_yaw)) if pred_edge_visible_yaw is not None else None, "pred_edge_bucket_yaw_deg": to_float(math.degrees(pred_edge_bucket_visible_yaw)) if pred_edge_bucket_visible_yaw is not None else None, "gt_bbox_xyxy": [float(v) for v in gt_box.tolist()], "pred_bbox_xyxy": [float(v) for v in pred_box.tolist()], "gt_center": to_list(gt_center), "pred_center": to_list(pred_center), "gt_position_center": to_list(gt_position_center), "pred_position_center": to_list(pred_position_center), "gt_dims": to_list(gt_attrs["dims"]), "pred_dims": to_list(pred_attrs["dims"]), "gt_face_center": None if gt_face_center is None else to_list(gt_face_center), "pred_face_center": None if pred_face_center is None else to_list(pred_face_center), "gt_visible_face_type": gt_visible_face_type, "pred_visible_face_type": pred_visible_face_type, "gt_visible_face_types": [int(v) for v in gt_visible_face_types], "pred_visible_face_types": [int(v) for v in pred_visible_face_types], "gt_face_center_2d": ( None if gt_decoded is None or not reuse_gt_decoded_face_overlay else to_serializable_array(gt_decoded.get("face_center_2d")) ), "pred_face_center_2d": ( None if pred_decoded is None or not reuse_pred_decoded_face_overlay else to_serializable_array(pred_decoded.get("face_center_2d")) ), "gt_face_color": None if gt_decoded is None or not reuse_gt_decoded_face_overlay else to_serializable_array(gt_decoded.get("face_color")), "pred_face_color": ( None if pred_decoded is None or not reuse_pred_decoded_face_overlay else to_serializable_array(pred_decoded.get("face_color")) ), "gt_edge_points_2d": None if gt_decoded is None else to_serializable_array(gt_decoded.get("edge_points_2d")), "pred_edge_points_2d": None if pred_decoded is None else to_serializable_array(pred_decoded.get("edge_points_2d")), "gt_corners": to_serializable_array(gt_attrs["corners_3d"]), "pred_corners": to_serializable_array(pred_attrs["corners_3d"]), "pred_edge_corners": None if pred_edge_corners is None else to_serializable_array(pred_edge_corners), "pred_edge_dims": None if pred_edge_dims is None else to_list(pred_edge_dims), "pred_edge_visible_face_type": None if pred_edge_visible_face_type is None else int(pred_edge_visible_face_type), "pred_edge_visible_face_types": [int(v) for v in pred_edge_visible_face_types], "pred_edge_edge_points_2d": None if pred_edge_points_2d is None else to_serializable_array(pred_edge_points_2d), "gt_yaw_rad": gt_yaw, "pred_yaw_rad": pred_yaw, "gt_visible_yaw_rad": gt_visible_yaw, "pred_edge_yaw_rad": pred_edge_visible_yaw, "pred_edge_bucket_yaw_rad": pred_edge_bucket_visible_yaw, } def annotate_panel_title(image: np.ndarray, title: str) -> np.ndarray: panel = image.copy() cv2.putText(panel, title, (10, 24), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 255), 2, cv2.LINE_AA) return panel def draw_dashed_rectangle( image: np.ndarray, p1: tuple[int, int], p2: tuple[int, int], color: tuple[int, int, int], thickness: int = 2, dash: int = 8, gap: int = 5, ) -> None: x1, y1 = p1 x2, y2 = p2 for x in range(x1, x2, dash + gap): cv2.line(image, (x, y1), (min(x + dash, x2), y1), color, thickness, cv2.LINE_AA) cv2.line(image, (x, y2), (min(x + dash, x2), y2), color, thickness, cv2.LINE_AA) for y in range(y1, y2, dash + gap): cv2.line(image, (x1, y), (x1, min(y + dash, y2)), color, thickness, cv2.LINE_AA) cv2.line(image, (x2, y), (x2, min(y + dash, y2)), color, thickness, cv2.LINE_AA) def draw_box_with_label(image: np.ndarray, xyxy: list[float], color: tuple[int, int, int], label: str, dashed: bool = False) -> np.ndarray: drawn = image.copy() x1, y1, x2, y2 = [int(round(float(v))) for v in xyxy] if dashed: draw_dashed_rectangle(drawn, (x1, y1), (x2, y2), color, thickness=2) else: cv2.rectangle(drawn, (x1, y1), (x2, y2), color, 2, cv2.LINE_AA) (tw, th), _ = cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, 0.5, 1) cv2.rectangle(drawn, (x1, max(0, y1 - th - 6)), (x1 + tw + 4, y1), color, -1) cv2.putText(drawn, label, (x1 + 2, max(th + 1, y1 - 3)), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 255, 255), 1, cv2.LINE_AA) return drawn def parse_visible_face_names(visible_faces: Any) -> list[int]: if visible_faces is None: return [] if isinstance(visible_faces, str): parts = [part.strip() for part in visible_faces.split(",") if part.strip() and part.strip() != "none"] elif isinstance(visible_faces, (list, tuple)): parts = [str(part).strip() for part in visible_faces if str(part).strip() and str(part).strip() != "none"] else: parts = [str(visible_faces).strip()] inverse = {str(name): int(face_type) for face_type, name in FACE_NAMES.items()} return [inverse[part] for part in parts if part in inverse] def build_badcase_draw_kwargs( corners_3d: Any, calib: dict[str, Any], visible_face_type: Optional[int] = None, visible_face_types: Optional[list[int]] = None, face_center_3d: Any = None, ) -> dict[str, Any]: kwargs: dict[str, Any] = {} corners_arr = np.asarray(corners_3d, dtype=np.float32) if corners_arr.shape != (8, 3) or not np.isfinite(corners_arr).all(): return kwargs edge_face_types = [int(face_type) for face_type in (visible_face_types or []) if face_type in range(4)] if not edge_face_types and visible_face_type in range(4): edge_face_types = [int(visible_face_type)] if edge_face_types: _, edge_points_2d = collect_face_bottom_edges(corners_arr, edge_face_types, calib, num_samples=5) kwargs["edge_points_2d"] = edge_points_2d face_type = int(visible_face_type) if visible_face_type in range(4) else None if face_type is not None: if face_center_3d is None: face_center_3d = face_center_from_corners(corners_arr, face_type) face_center_arr = None if face_center_3d is None else np.asarray(face_center_3d, dtype=np.float32) if face_center_arr is not None and face_center_arr.shape == (3,) and np.isfinite(face_center_arr).all(): projected = project_3d_to_2d_with_distortion(face_center_arr[None, :], calib) if projected is not None and len(projected) > 0 and np.isfinite(projected[0]).all(): kwargs["face_center_2d"] = tuple(float(v) for v in projected[0].tolist()) kwargs["face_color"] = FACE_COLORS[face_type] return kwargs def draw_3d_panel_object( panel: np.ndarray, calib: dict[str, Any], corners_3d: Any, dims: list[float], yaw_rad: Optional[float], visible_face_type: Optional[int] = None, face_center: Any = None, visible_face_types: Optional[list[int]] = None, face_center_2d: Any = None, face_color: Any = None, edge_points_2d: Any = None, rebuild: bool = False, thickness: int = 2, ) -> np.ndarray: dims_arr = np.asarray(dims, dtype=np.float32) corners_arr = np.asarray(corners_3d, dtype=np.float32) if corners_arr.shape != (8, 3) or not np.all(np.isfinite(corners_arr)): return panel draw_corners = corners_arr if rebuild and yaw_rad is not None and np.all(np.isfinite(dims_arr)): rebuilt_corners = rebuild_box_corners_for_visualization( corners_arr, dims_arr, float(yaw_rad), visible_face_type=visible_face_type, face_center_3d=face_center, ) if rebuilt_corners is not None: draw_corners = np.asarray(rebuilt_corners, dtype=np.float32) draw_kwargs = build_badcase_draw_kwargs( draw_corners, calib, visible_face_type=visible_face_type, visible_face_types=visible_face_types, face_center_3d=face_center, ) if face_center_2d is not None: face_center_2d_arr = np.asarray(face_center_2d, dtype=np.float32).reshape(-1) if face_center_2d_arr.size >= 2 and np.isfinite(face_center_2d_arr[:2]).all(): draw_kwargs["face_center_2d"] = tuple(float(v) for v in face_center_2d_arr[:2].tolist()) if face_color is not None: face_color_arr = np.asarray(face_color, dtype=np.int32).reshape(-1) if face_color_arr.size >= 3: draw_kwargs["face_color"] = tuple(int(v) for v in face_color_arr[:3].tolist()) if edge_points_2d is not None: draw_kwargs["edge_points_2d"] = np.asarray(edge_points_2d, dtype=np.float32) draw_3d_box(panel, draw_corners, calib, thickness=thickness, **draw_kwargs) return panel def make_3d_panel( image: np.ndarray, calib: dict[str, Any], corners_3d: Any, dims: list[float], yaw_rad: Optional[float], title: str, visible_face_type: Optional[int] = None, face_center: Any = None, visible_face_types: Optional[list[int]] = None, face_center_2d: Any = None, face_color: Any = None, edge_points_2d: Any = None, rebuild: bool = False, ) -> np.ndarray: panel = image.copy() draw_3d_panel_object( panel, calib, corners_3d, dims, yaw_rad, visible_face_type=visible_face_type, face_center=face_center, visible_face_types=visible_face_types, face_center_2d=face_center_2d, face_color=face_color, edge_points_2d=edge_points_2d, rebuild=rebuild, thickness=2, ) return annotate_panel_title(panel, title) def make_text_panel(shape: tuple[int, int, int], title: str, record: dict[str, Any]) -> np.ndarray: panel = np.full(shape, 28, dtype=np.uint8) lines = [ title, f"{record['cls_name']} conf={record['confidence']:.2f} iou={record['match_iou']:.2f}", f"face_select gt={record.get('gt_face_selection_label', 'n/a')} pred={record.get('pred_face_selection_label', 'n/a')}" if record.get("gt_face_selection_label") is not None or record.get("pred_face_selection_label") is not None else "face_select n/a", f"yaw gt={record['gt_yaw_deg']:.1f} direct={record['pred_yaw_deg']:.1f} err={record['yaw_abs_deg']:.1f} deg" if record["yaw_abs_deg"] is not None and record["gt_yaw_deg"] is not None and record["pred_yaw_deg"] is not None else "yaw unavailable", ( f"yaw gt={record['gt_yaw_deg']:.1f} edge={record['pred_edge_yaw_deg']:.1f}" ) if record["gt_yaw_deg"] is not None and record["pred_edge_yaw_deg"] is not None else "edge yaw unavailable", ( f"gt-yaw err direct={record['direct_visible_yaw_abs_deg']:.1f} edge={record['edge_visible_yaw_abs_deg']:.1f} " f"gain={record['direct_minus_edge_visible_yaw_abs_deg']:.1f}" ) if record["direct_visible_yaw_abs_deg"] is not None and record["edge_visible_yaw_abs_deg"] is not None and record["direct_minus_edge_visible_yaw_abs_deg"] is not None else "paired yaw comparison unavailable", ( f"|gt_x|={record['gt_lateral_abs_m']:.2f}m yaw_compare={record['yaw_compare_eligible']} " f"cut={record['is_cut_object']} pos_eligible={record['position_eligible']} basis={record.get('position_error_basis', 'n/a')}" ) if record["gt_lateral_abs_m"] is not None else ( f"yaw_compare={record['yaw_compare_eligible']} cut={record['is_cut_object']} " f"pos_eligible={record['position_eligible']} basis={record.get('position_error_basis', 'n/a')}" ), f"x gt={record['position_gt_x_m']:.2f} pred={record['position_pred_x_m']:.2f} abs={record['x_abs_m']:.2f} m" if record["x_abs_m"] is not None and record["position_gt_x_m"] is not None and record["position_pred_x_m"] is not None else "x unavailable", f"y gt={record['position_gt_y_m']:.2f} pred={record['position_pred_y_m']:.2f} abs={record['y_abs_m']:.2f} m" if record["y_abs_m"] is not None and record["position_gt_y_m"] is not None and record["position_pred_y_m"] is not None else "y unavailable", f"z gt={record['position_gt_z_m']:.2f} pred={record['position_pred_z_m']:.2f} abs={record['z_abs_m']:.2f} m" if record["z_abs_m"] is not None and record["position_gt_z_m"] is not None and record["position_pred_z_m"] is not None else "z unavailable", f"faces={record['visible_face_count']} [{record['visible_faces']}]", f"dist={record['distance_bin']} box={record['bbox_diag_bin']} frame={record['frame_name']}", ] y = 26 for line in lines: cv2.putText(panel, line, (10, y), cv2.FONT_HERSHEY_SIMPLEX, 0.54, (220, 220, 220), 1, cv2.LINE_AA) y += 34 return panel def metric_value_for_category(record: dict[str, Any], category: str) -> Optional[float]: if category == "yaw": return to_float(record.get("yaw_abs_deg")) if category == "horizontal": return to_float(record.get("x_abs_m")) if category == "vertical": return to_float(record.get("z_abs_m")) return None def metric_unit_for_category(category: str) -> str: return "deg" if category == "yaw" else "m" def make_interval_bin_text_panel( shape: tuple[int, int, int], title: str, category: str, threshold_start: float, threshold_label: str, records: list[dict[str, Any]], ) -> np.ndarray: panel = np.full(shape, 28, dtype=np.uint8) unit = metric_unit_for_category(category) metric_values = [metric_value_for_category(record, category) for record in records] metric_values = [float(value) for value in metric_values if value is not None] cls_names = sorted({str(record.get("cls_name", "unknown")) for record in records}) lines = [ title, f"objects with {category} error >= {format_float(threshold_start, 1 if unit == 'm' else 0)} {unit}: {len(records)}", f"bin={threshold_label} max={format_float(max(metric_values) if metric_values else None, 2)}{unit} mean={format_float(mean_or_none(metric_values), 2)}{unit}", f"classes={','.join(cls_names[:4])}" + (f" +{max(0, len(cls_names) - 4)} more" if len(cls_names) > 4 else ""), f"frame={records[0]['frame_name']}" if records else "frame=n/a", ] for index, record in enumerate(records[:8], start=1): metric_value = metric_value_for_category(record, category) lines.append( f"#{index} {record['cls_name']} err={format_float(metric_value, 2)}{unit} conf={format_float(record.get('confidence'), 2)} iou={format_float(record.get('match_iou'), 2)}" ) if len(records) > 8: lines.append(f"... and {len(records) - 8} more objects") y = 26 for line in lines[:11]: cv2.putText(panel, line, (10, y), cv2.FONT_HERSHEY_SIMPLEX, 0.54, (220, 220, 220), 1, cv2.LINE_AA) y += 34 return panel def make_2d_badcase_record( sample_index: int, roi_name: str, image_path: Path, label_path: Path, kind: str, cls_id: int, cls_name: str, bbox_xyxy: np.ndarray, gt_index: Optional[int] = None, pred_index: Optional[int] = None, confidence: Optional[float] = None, max_iou_any: Optional[float] = None, max_iou_same_class: Optional[float] = None, ) -> dict[str, Any]: return { "sample_index": int(sample_index), "roi": roi_name, "frame_name": image_path.name, "image_path": str(image_path), "label_path": str(label_path), "kind": str(kind), "cls_id": int(cls_id), "cls_name": str(cls_name), "gt_index": None if gt_index is None else int(gt_index), "pred_index": None if pred_index is None else int(pred_index), "confidence": to_float(confidence), "max_iou_any": to_float(max_iou_any), "max_iou_same_class": to_float(max_iou_same_class), "bbox_xyxy": [float(v) for v in np.asarray(bbox_xyxy, dtype=np.float32).tolist()], } def make_face_selection_badcase_record( sample_index: int, roi_name: str, image_path: Path, label_path: Path, gt_index: int, pred_index: int, gt_cls_id: int, gt_cls_name: str, pred_cls_id: int, pred_cls_name: str, gt_face_selection_label: str, pred_face_selection_label: str, gt_bbox_xyxy: np.ndarray, pred_bbox_xyxy: np.ndarray, match_iou: float, confidence: float, ) -> dict[str, Any]: return { "sample_index": int(sample_index), "roi": roi_name, "frame_name": image_path.name, "image_path": str(image_path), "label_path": str(label_path), "gt_index": int(gt_index), "pred_index": int(pred_index), "gt_cls_id": int(gt_cls_id), "gt_cls_name": str(gt_cls_name), "pred_cls_id": int(pred_cls_id), "pred_cls_name": str(pred_cls_name), "gt_face_selection_label": str(gt_face_selection_label), "pred_face_selection_label": str(pred_face_selection_label), "match_iou": float(match_iou), "confidence": to_float(confidence), "gt_bbox_xyxy": [float(v) for v in np.asarray(gt_bbox_xyxy, dtype=np.float32).tolist()], "pred_bbox_xyxy": [float(v) for v in np.asarray(pred_bbox_xyxy, dtype=np.float32).tolist()], } def make_fake_class_badcase_record( sample_index: int, roi_name: str, image_path: Path, label_path: Path, gt_index: int, pred_index: int, gt_cls_id: int, gt_cls_name: str, pred_cls_id: int, pred_cls_name: str, gt_bbox_xyxy: np.ndarray, pred_bbox_xyxy: np.ndarray, match_iou: float, confidence: float, ) -> dict[str, Any]: gt_fake_label = fake_class_label(gt_cls_name) pred_fake_label = fake_class_label(pred_cls_name) gt_occ = occlusion_binary_label(gt_cls_name) pred_occ = occlusion_binary_label(pred_cls_name) if gt_fake_label == pred_fake_label: kind = "fake_class_correct" elif gt_fake_label == "non_fake" or pred_fake_label == "non_fake": kind = "fake_vs_non_fake" elif gt_occ == "occluded" and pred_occ == "occluded": kind = "fake_internal_confusion" else: kind = "visible_internal_confusion" return { "sample_index": int(sample_index), "roi": roi_name, "frame_name": image_path.name, "image_path": str(image_path), "label_path": str(label_path), "kind": kind, "gt_index": int(gt_index), "pred_index": int(pred_index), "gt_cls_id": int(gt_cls_id), "gt_cls_name": str(gt_cls_name), "pred_cls_id": int(pred_cls_id), "pred_cls_name": str(pred_cls_name), "gt_fake_class_label": gt_fake_label, "pred_fake_class_label": pred_fake_label, "gt_occlusion_label": gt_occ, "pred_occlusion_label": pred_occ, "match_iou": float(match_iou), "confidence": to_float(confidence), "gt_bbox_xyxy": [float(v) for v in np.asarray(gt_bbox_xyxy, dtype=np.float32).tolist()], "pred_bbox_xyxy": [float(v) for v in np.asarray(pred_bbox_xyxy, dtype=np.float32).tolist()], } def make_fake_class_text_panel(shape: tuple[int, int, int], title: str, record: dict[str, Any]) -> np.ndarray: panel = np.full(shape, 28, dtype=np.uint8) lines = [ title, f"kind={record['kind']}", f"GT {record['gt_cls_name']} fake={record.get('gt_fake_class_label')} occ={record['gt_occlusion_label']}", f"Pred {record['pred_cls_name']} fake={record.get('pred_fake_class_label')} occ={record['pred_occlusion_label']} conf={format_float(record.get('confidence'), 2)}", f"iou={format_float(record.get('match_iou'), 3)} gt_index={record.get('gt_index')} pred_index={record.get('pred_index')}", f"frame={record['frame_name']}", ] y = 28 for line in lines: cv2.putText(panel, line, (10, y), cv2.FONT_HERSHEY_SIMPLEX, 0.56, (220, 220, 220), 1, cv2.LINE_AA) y += 34 return panel def make_2d_badcase_text_panel(shape: tuple[int, int, int], title: str, record: dict[str, Any]) -> np.ndarray: panel = np.full(shape, 28, dtype=np.uint8) lines = [ title, f"{record['cls_name']} kind={record['kind']}", f"conf={format_float(record.get('confidence'), 2)}" if record.get("confidence") is not None else "conf=n/a", f"max_iou_any={format_float(record.get('max_iou_any'), 3)}", f"max_iou_same_class={format_float(record.get('max_iou_same_class'), 3)}", f"gt_index={record.get('gt_index')} pred_index={record.get('pred_index')}", f"frame={record['frame_name']}", ] y = 28 for line in lines: cv2.putText(panel, line, (10, y), cv2.FONT_HERSHEY_SIMPLEX, 0.56, (220, 220, 220), 1, cv2.LINE_AA) y += 34 return panel def make_face_selection_text_panel(shape: tuple[int, int, int], title: str, record: dict[str, Any]) -> np.ndarray: panel = np.full(shape, 28, dtype=np.uint8) lines = [ title, f"GT select={record.get('gt_face_selection_label')} cls={record.get('gt_cls_name')}", f"Pred select={record.get('pred_face_selection_label')} cls={record.get('pred_cls_name')} conf={format_float(record.get('confidence'), 2)}", f"iou={format_float(record.get('match_iou'), 3)} gt_index={record.get('gt_index')} pred_index={record.get('pred_index')}", f"frame={record['frame_name']}", ] y = 28 for line in lines: cv2.putText(panel, line, (10, y), cv2.FONT_HERSHEY_SIMPLEX, 0.56, (220, 220, 220), 1, cv2.LINE_AA) y += 34 return panel def resize_panel(panel: np.ndarray, target_size: tuple[int, int]) -> np.ndarray: target_w, target_h = target_size if panel.shape[1] == target_w and panel.shape[0] == target_h: return panel return cv2.resize(panel, (target_w, target_h), interpolation=cv2.INTER_LINEAR) def assemble_visual_grid(panels: list[np.ndarray], panel_size: tuple[int, int]) -> np.ndarray: resized = [resize_panel(panel, panel_size) for panel in panels] top = np.concatenate(resized[:3], axis=1) bottom = np.concatenate(resized[3:], axis=1) return np.concatenate([top, bottom], axis=0) def save_badcase_visuals( records: list[dict[str, Any]], output_dir: Path, category: str, bundle, image_root: Path, ) -> list[dict[str, Any]]: output_dir.mkdir(parents=True, exist_ok=True) manifest = [] for rank, record in enumerate(records, start=1): image_path = Path(record["image_path"]) image_bgr = cv2.imread(str(image_path), cv2.IMREAD_COLOR) if image_bgr is None: continue raw_calib = read_raw_calib_from_label(image_path, Path(str(record["label_path"]))) prepared = prepare_roi_image(image_bgr, raw_calib, bundle.spec, bundle.imgsz) roi_image = prepared.image.copy() panel_size = (roi_image.shape[1], roi_image.shape[0]) panel_2d = draw_box_with_label(roi_image, record["gt_bbox_xyxy"], (0, 200, 0), "GT") panel_2d = draw_box_with_label(panel_2d, record["pred_bbox_xyxy"], (0, 0, 255), "Pred") panel_2d = annotate_panel_title(panel_2d, f"{bundle.spec.name} 2D {category}") panel_gt = make_3d_panel( roi_image, prepared.calib, record["gt_corners"], record["gt_dims"], record["gt_yaw_rad"], "GT 3D", visible_face_type=record.get("gt_visible_face_type"), face_center=record.get("gt_face_center"), visible_face_types=record.get("gt_visible_face_types") or parse_visible_face_names(record.get("visible_faces")), face_center_2d=record.get("gt_face_center_2d"), face_color=record.get("gt_face_color"), edge_points_2d=record.get("gt_edge_points_2d"), ) panel_pred = make_3d_panel( roi_image, prepared.calib, record["pred_corners"], record["pred_dims"], record["pred_yaw_rad"], "Pred 3D", visible_face_type=record.get("pred_visible_face_type"), face_center=record.get("pred_face_center"), visible_face_types=record.get("pred_visible_face_types") or ([int(record["pred_visible_face_type"])] if record.get("pred_visible_face_type") is not None else []), face_center_2d=record.get("pred_face_center_2d"), face_color=record.get("pred_face_color"), edge_points_2d=record.get("pred_edge_points_2d"), ) edge_panel_corners = record.get("pred_edge_corners") edge_panel_dims = record.get("pred_edge_dims") edge_panel_visible_face_type = record.get("pred_edge_visible_face_type") edge_panel_visible_face_types = record.get("pred_edge_visible_face_types") edge_panel_points_2d = record.get("pred_edge_edge_points_2d") panel_edge = make_3d_panel( roi_image, prepared.calib, edge_panel_corners if edge_panel_corners is not None else record["pred_corners"], edge_panel_dims if edge_panel_dims is not None else record["pred_dims"], record["pred_edge_yaw_rad"], "Pred 3D EdgeYaw", visible_face_type=( edge_panel_visible_face_type if edge_panel_visible_face_type is not None else record.get("pred_visible_face_type") ), face_center=None if edge_panel_corners is not None else record.get("pred_face_center"), visible_face_types=( edge_panel_visible_face_types or record.get("pred_visible_face_types") or ([int(record["pred_visible_face_type"])] if record.get("pred_visible_face_type") is not None else []) ), face_center_2d=None if edge_panel_corners is not None else record.get("pred_face_center_2d"), face_color=None if edge_panel_corners is not None else record.get("pred_face_color"), edge_points_2d=edge_panel_points_2d if edge_panel_points_2d is not None else record.get("pred_edge_points_2d"), rebuild=edge_panel_corners is None, ) gt_bev = [] pred_bev = [] if record["gt_yaw_rad"] is not None: gt_center_for_bev = np.asarray(record["gt_center"], dtype=np.float32) if record.get("gt_visible_face_type") is not None and record.get("gt_face_center") is not None: gt_corners = rebuild_box_corners_for_visualization( np.asarray(record["gt_corners"], dtype=np.float32), np.asarray(record["gt_dims"], dtype=np.float32), float(record["gt_yaw_rad"]), visible_face_type=int(record["gt_visible_face_type"]), face_center_3d=np.asarray(record["gt_face_center"], dtype=np.float32), ) if gt_corners is not None: gt_center_for_bev = np.asarray(gt_corners, dtype=np.float32).mean(axis=0) gt_bev.append( { "center": gt_center_for_bev, "dims": np.asarray(record["gt_dims"], dtype=np.float32), "yaw": record["gt_yaw_rad"], } ) if record["pred_yaw_rad"] is not None: pred_center_for_bev = np.asarray(record["pred_center"], dtype=np.float32) if record.get("pred_visible_face_type") is not None and record.get("pred_face_center") is not None: pred_corners = rebuild_box_corners_for_visualization( np.asarray(record["pred_corners"], dtype=np.float32), np.asarray(record["pred_dims"], dtype=np.float32), float(record["pred_yaw_rad"]), visible_face_type=int(record["pred_visible_face_type"]), face_center_3d=np.asarray(record["pred_face_center"], dtype=np.float32), ) if pred_corners is not None: pred_center_for_bev = np.asarray(pred_corners, dtype=np.float32).mean(axis=0) pred_bev.append( { "center": pred_center_for_bev, "dims": np.asarray(record["pred_dims"], dtype=np.float32), "yaw": record["pred_yaw_rad"], } ) bev_rgb = create_bev_image( gt_bev, pred_bev, max_range=max(80, int(max(record.get("gt_depth_m") or 0.0, record.get("pred_depth_m") or 0.0) + 20)), lateral_range=40, ) panel_bev = annotate_panel_title(cv2.cvtColor(bev_rgb, cv2.COLOR_RGB2BGR), "BEV GT vs Pred") panel_text = make_text_panel(roi_image.shape, f"{bundle.spec.name} {category} #{rank}", record) grid = assemble_visual_grid([panel_2d, panel_gt, panel_pred, panel_edge, panel_bev, panel_text], panel_size) filename = ( f"{rank:03d}_{sanitize_name(Path(record['frame_name']).stem)}_{sanitize_name(record['cls_name'])}" f"_g{record['gt_index']}_p{record['pred_index']}.jpg" ) image_out = output_dir / filename cv2.imwrite(str(image_out), grid) manifest.append({**record, "visualization": str(image_out)}) with (output_dir / "manifest.json").open("w", encoding="utf-8") as file: json.dump(manifest, file, indent=2, ensure_ascii=False) return manifest def save_class_badcase_visuals( class_records: dict[tuple[int, str], list[dict[str, Any]]], output_root: Path, category: str, bundle, image_root: Path, ) -> list[dict[str, Any]]: output_root.mkdir(parents=True, exist_ok=True) manifests: list[dict[str, Any]] = [] for (cls_id, cls_name), records in sorted(class_records.items(), key=lambda item: item[0][0]): if not records: continue class_dir = output_root / f"{int(cls_id):02d}_{sanitize_name(cls_name)}" manifest = save_badcase_visuals(records, class_dir, category, bundle, image_root) manifests.append( { "cls_id": int(cls_id), "cls_name": str(cls_name), "count": len(manifest), "manifest_path": str(class_dir / "manifest.json"), } ) return manifests def save_interval_badcase_visuals( interval_records: dict[float, list[dict[str, Any]]], frame_records: dict[str, list[dict[str, Any]]], output_root: Path, category: str, bin_width: float, unit: str, samples_per_bin: int, bundle, image_root: Path, ) -> list[dict[str, Any]]: output_root.mkdir(parents=True, exist_ok=True) manifests: list[dict[str, Any]] = [] for start, records in sorted(interval_records.items(), key=lambda item: item[0], reverse=True): if not records: continue bin_dir = output_root / interval_slug(float(start), float(bin_width), unit) bin_dir.mkdir(parents=True, exist_ok=True) label = interval_label(float(start), float(bin_width), unit=unit) frame_candidates: dict[str, list[dict[str, Any]]] = defaultdict(list) for record in records: frame_candidates[str(record["image_path"])].append(record) ranked_frames = [] for frame_key in frame_candidates: selected_records = [ record for record in frame_records.get(frame_key, []) if (metric_value_for_category(record, category) is not None) and float(metric_value_for_category(record, category)) >= float(start) ] if not selected_records: continue selected_records = sorted( selected_records, key=lambda record: ( float(metric_value_for_category(record, category) or -float("inf")), float(to_float(record.get("confidence")) or -float("inf")), ), reverse=True, ) ranked_frames.append( ( float(metric_value_for_category(selected_records[0], category) or -float("inf")), len(selected_records), frame_key, selected_records, ) ) ranked_frames.sort(key=lambda item: (item[0], item[1], item[2]), reverse=True) manifest = [] for rank, (_, _, frame_key, selected_records) in enumerate(ranked_frames[: max(1, int(samples_per_bin))], start=1): image_path = Path(frame_key) image_bgr = cv2.imread(str(image_path), cv2.IMREAD_COLOR) if image_bgr is None: continue raw_calib = read_raw_calib_from_label(image_path, Path(str(selected_records[0]["label_path"]))) prepared = prepare_roi_image(image_bgr, raw_calib, bundle.spec, bundle.imgsz) roi_image = prepared.image.copy() panel_size = (roi_image.shape[1], roi_image.shape[0]) panel_2d = roi_image.copy() panel_gt = roi_image.copy() panel_pred = roi_image.copy() panel_edge = roi_image.copy() gt_bev = [] pred_bev = [] for obj_index, record in enumerate(selected_records, start=1): panel_2d = draw_box_with_label(panel_2d, record["gt_bbox_xyxy"], (0, 200, 0), f"G{obj_index}") panel_2d = draw_box_with_label(panel_2d, record["pred_bbox_xyxy"], (0, 0, 255), f"P{obj_index}") draw_3d_panel_object( panel_gt, prepared.calib, record["gt_corners"], record["gt_dims"], record["gt_yaw_rad"], visible_face_type=record.get("gt_visible_face_type"), face_center=record.get("gt_face_center"), visible_face_types=record.get("gt_visible_face_types") or parse_visible_face_names(record.get("visible_faces")), face_center_2d=record.get("gt_face_center_2d"), face_color=record.get("gt_face_color"), edge_points_2d=record.get("gt_edge_points_2d"), thickness=2, ) draw_3d_panel_object( panel_pred, prepared.calib, record["pred_corners"], record["pred_dims"], record["pred_yaw_rad"], visible_face_type=record.get("pred_visible_face_type"), face_center=record.get("pred_face_center"), visible_face_types=record.get("pred_visible_face_types") or ([int(record["pred_visible_face_type"])] if record.get("pred_visible_face_type") is not None else []), face_center_2d=record.get("pred_face_center_2d"), face_color=record.get("pred_face_color"), edge_points_2d=record.get("pred_edge_points_2d"), thickness=2, ) draw_3d_panel_object( panel_edge, prepared.calib, record["pred_corners"], record["pred_dims"], record["pred_edge_yaw_rad"], visible_face_type=record.get("pred_visible_face_type"), face_center=record.get("pred_face_center"), visible_face_types=record.get("pred_visible_face_types") or ([int(record["pred_visible_face_type"])] if record.get("pred_visible_face_type") is not None else []), face_center_2d=record.get("pred_face_center_2d"), face_color=record.get("pred_face_color"), edge_points_2d=record.get("pred_edge_points_2d"), rebuild=True, thickness=2, ) if record["gt_yaw_rad"] is not None: gt_center_for_bev = np.asarray(record["gt_center"], dtype=np.float32) if record.get("gt_visible_face_type") is not None and record.get("gt_face_center") is not None: gt_corners = rebuild_box_corners_for_visualization( np.asarray(record["gt_corners"], dtype=np.float32), np.asarray(record["gt_dims"], dtype=np.float32), float(record["gt_yaw_rad"]), visible_face_type=int(record["gt_visible_face_type"]), face_center_3d=np.asarray(record["gt_face_center"], dtype=np.float32), ) if gt_corners is not None: gt_center_for_bev = np.asarray(gt_corners, dtype=np.float32).mean(axis=0) gt_bev.append( { "center": gt_center_for_bev, "dims": np.asarray(record["gt_dims"], dtype=np.float32), "yaw": record["gt_yaw_rad"], } ) if record["pred_yaw_rad"] is not None: pred_center_for_bev = np.asarray(record["pred_center"], dtype=np.float32) if record.get("pred_visible_face_type") is not None and record.get("pred_face_center") is not None: pred_corners = rebuild_box_corners_for_visualization( np.asarray(record["pred_corners"], dtype=np.float32), np.asarray(record["pred_dims"], dtype=np.float32), float(record["pred_yaw_rad"]), visible_face_type=int(record["pred_visible_face_type"]), face_center_3d=np.asarray(record["pred_face_center"], dtype=np.float32), ) if pred_corners is not None: pred_center_for_bev = np.asarray(pred_corners, dtype=np.float32).mean(axis=0) pred_bev.append( { "center": pred_center_for_bev, "dims": np.asarray(record["pred_dims"], dtype=np.float32), "yaw": record["pred_yaw_rad"], } ) panel_2d = annotate_panel_title(panel_2d, f"{bundle.spec.name} 2D {category}_{label}") panel_gt = annotate_panel_title(panel_gt, "GT 3D") panel_pred = annotate_panel_title(panel_pred, "Pred 3D") panel_edge = annotate_panel_title(panel_edge, "Pred 3D EdgeYaw") bev_rgb = create_bev_image( gt_bev, pred_bev, max_range=max( 80, int( max( [to_float(record.get("gt_depth_m")) or 0.0 for record in selected_records] + [to_float(record.get("pred_depth_m")) or 0.0 for record in selected_records] ) + 20 ), ), lateral_range=40, ) panel_bev = annotate_panel_title(cv2.cvtColor(bev_rgb, cv2.COLOR_RGB2BGR), "BEV GT vs Pred") panel_text = make_interval_bin_text_panel( roi_image.shape, f"{bundle.spec.name} {category} {label} #{rank}", category, float(start), label, selected_records, ) grid = assemble_visual_grid([panel_2d, panel_gt, panel_pred, panel_edge, panel_bev, panel_text], panel_size) metric_values = [metric_value_for_category(record, category) for record in selected_records] metric_values = [float(value) for value in metric_values if value is not None] image_out = bin_dir / ( f"{rank:03d}_{sanitize_name(image_path.stem)}_{sanitize_name(category)}_{sanitize_name(label)}_n{len(selected_records)}.jpg" ) cv2.imwrite(str(image_out), grid) manifest.append( { "roi": bundle.spec.name, "frame_name": image_path.name, "image_path": str(image_path), "bin_start": float(start), "bin_end": float(start + bin_width), "bin_label": label, "badcase_count": int(len(selected_records)), "max_metric_value": max(metric_values) if metric_values else None, "mean_metric_value": mean_or_none(metric_values), "metric_unit": unit, "cls_summary": ", ".join(sorted({str(record.get("cls_name", "unknown")) for record in selected_records})[:6]), "visualization": str(image_out), } ) with (bin_dir / "manifest.json").open("w", encoding="utf-8") as file: json.dump(manifest, file, indent=2, ensure_ascii=False) manifests.append( { "bin_start": float(start), "bin_end": float(start + bin_width), "bin_label": label, "count": len(manifest), "manifest_path": str(bin_dir / "manifest.json"), } ) return manifests def save_2d_badcase_visuals( records: list[dict[str, Any]], output_dir: Path, category: str, bundle, image_root: Path, ) -> list[dict[str, Any]]: output_dir.mkdir(parents=True, exist_ok=True) manifest = [] for rank, record in enumerate(records, start=1): image_path = Path(record["image_path"]) image_bgr = cv2.imread(str(image_path), cv2.IMREAD_COLOR) if image_bgr is None: continue raw_calib = read_raw_calib_from_label(image_path, Path(str(record["label_path"]))) prepared = prepare_roi_image(image_bgr, raw_calib, bundle.spec, bundle.imgsz) roi_image = prepared.image.copy() box_color = (0, 200, 0) if record["kind"] == "false_negative" else (0, 0, 255) box_label = "GT miss" if record["kind"] == "false_negative" else "Pred fp" panel_2d = draw_box_with_label(roi_image, record["bbox_xyxy"], box_color, box_label) panel_2d = annotate_panel_title(panel_2d, f"{bundle.spec.name} 2D {category}") text_panel = make_2d_badcase_text_panel(roi_image.shape, f"{bundle.spec.name} {category} #{rank}", record) grid = np.concatenate([panel_2d, text_panel], axis=1) filename = ( f"{rank:03d}_{sanitize_name(Path(record['frame_name']).stem)}_{sanitize_name(record['cls_name'])}" f"_{sanitize_name(record['kind'])}_g{record.get('gt_index')}_p{record.get('pred_index')}.jpg" ) image_out = output_dir / filename cv2.imwrite(str(image_out), grid) manifest.append({**record, "visualization": str(image_out)}) with (output_dir / "manifest.json").open("w", encoding="utf-8") as file: json.dump(manifest, file, indent=2, ensure_ascii=False) return manifest def save_face_selection_badcase_visuals( records: list[dict[str, Any]], output_dir: Path, category: str, bundle, image_root: Path, ) -> list[dict[str, Any]]: output_dir.mkdir(parents=True, exist_ok=True) manifest = [] for rank, record in enumerate(records, start=1): image_path = Path(record["image_path"]) image_bgr = cv2.imread(str(image_path), cv2.IMREAD_COLOR) if image_bgr is None: continue raw_calib = read_raw_calib_from_label(image_path, Path(str(record["label_path"]))) prepared = prepare_roi_image(image_bgr, raw_calib, bundle.spec, bundle.imgsz) roi_image = prepared.image.copy() panel_2d = draw_box_with_label( roi_image, record["gt_bbox_xyxy"], (0, 200, 0), f"GT {record.get('gt_face_selection_label', '')}", ) panel_2d = draw_box_with_label( panel_2d, record["pred_bbox_xyxy"], (0, 0, 255), f"Pred {record.get('pred_face_selection_label', '')}", ) panel_2d = annotate_panel_title(panel_2d, f"{bundle.spec.name} {category}") text_panel = make_face_selection_text_panel(roi_image.shape, f"{bundle.spec.name} {category} #{rank}", record) grid = np.concatenate([panel_2d, text_panel], axis=1) filename = ( f"{rank:03d}_{sanitize_name(Path(record['frame_name']).stem)}" f"_g{record.get('gt_index')}_p{record.get('pred_index')}.jpg" ) image_out = output_dir / filename cv2.imwrite(str(image_out), grid) manifest.append({**record, "visualization": str(image_out)}) with (output_dir / "manifest.json").open("w", encoding="utf-8") as file: json.dump(manifest, file, indent=2, ensure_ascii=False) return manifest def save_fake_class_badcase_visuals( records: list[dict[str, Any]], output_dir: Path, category: str, bundle, image_root: Path, ) -> list[dict[str, Any]]: output_dir.mkdir(parents=True, exist_ok=True) manifest = [] for rank, record in enumerate(records, start=1): image_path = Path(record["image_path"]) image_bgr = cv2.imread(str(image_path), cv2.IMREAD_COLOR) if image_bgr is None: continue raw_calib = read_raw_calib_from_label(image_path, Path(str(record["label_path"]))) prepared = prepare_roi_image(image_bgr, raw_calib, bundle.spec, bundle.imgsz) roi_image = prepared.image.copy() panel_2d = draw_box_with_label( roi_image, record["gt_bbox_xyxy"], (0, 200, 0), f"GT {record.get('gt_fake_class_label', record.get('gt_cls_name', ''))}", dashed=record.get("gt_occlusion_label") == "occluded", ) panel_2d = draw_box_with_label( panel_2d, record["pred_bbox_xyxy"], (0, 0, 255), f"Pred {record.get('pred_fake_class_label', record.get('pred_cls_name', ''))}", dashed=record.get("pred_occlusion_label") == "occluded", ) panel_2d = annotate_panel_title(panel_2d, f"{bundle.spec.name} fake_class {category}") text_panel = make_fake_class_text_panel(roi_image.shape, f"{bundle.spec.name} {category} #{rank}", record) grid = np.concatenate([panel_2d, text_panel], axis=1) filename = ( f"{rank:03d}_{sanitize_name(Path(record['frame_name']).stem)}" f"_{sanitize_name(record['kind'])}" f"_g{record.get('gt_index')}_p{record.get('pred_index')}.jpg" ) image_out = output_dir / filename cv2.imwrite(str(image_out), grid) manifest.append({**record, "visualization": str(image_out)}) with (output_dir / "manifest.json").open("w", encoding="utf-8") as file: json.dump(manifest, file, indent=2, ensure_ascii=False) return manifest def build_thresholded_2d_artifacts( eval_packets: list[dict[str, Any]], roi_name: str, names_dict: dict[int, str], confidence_threshold: float, topk_badcases: int, badcase_random_seed: int, roi_output: Path, bundle, image_root: Path, max_abs_lateral_m: float = FOCUSED_CONFUSION_MAX_ABS_LATERAL_M, max_abs_longitudinal_m: float = FOCUSED_CONFUSION_MAX_ABS_LONGITUDINAL_M, required_difficulty: int = FOCUSED_CONFUSION_REQUIRED_DIFFICULTY, ) -> dict[str, Any]: bad_2d_fn_handle, bad_2d_fn_writer = make_writer(roi_output / "bad_cases_2d_false_negative.csv", fieldnames=BADCASE_2D_FIELDS) bad_2d_fp_handle, bad_2d_fp_writer = make_writer(roi_output / "bad_cases_2d_false_positive.csv", fieldnames=BADCASE_2D_FIELDS) fake_class_handle, fake_class_writer = make_writer(roi_output / "bad_cases_fake_class.csv", fieldnames=FAKE_CLASS_BADCASE_FIELDS) class_counts = defaultdict(lambda: {"gt_total": 0, "pred_total": 0, "matched_2d": 0}) overall_counts = {"gt_total": 0, "pred_total": 0, "matched_2d": 0} fake_class_store_metrics = make_label_accuracy_store(FAKE_CLASS_LABEL_ORDER) fake_class_counts = {"fake_vs_non_fake": 0, "fake_internal_confusion": 0, "visible_internal_confusion": 0} confusion_matrix_2d = ConfusionMatrix(names=names_dict, task="detect") focused_confusion_matrix_2d = ConfusionMatrix(names=names_dict, task="detect") focused_confusion_gt_total = 0 focused_confusion_pred_spatial_total = 0 rng = random.Random(int(badcase_random_seed) + (2000 if str(roi_name).lower() == "roi1" else 1500)) bad_2d_fn_store = make_reservoir_store() bad_2d_fp_store = make_reservoir_store() fake_class_store = make_reservoir_store() for packet in eval_packets: gt_cls = np.asarray(packet["gt_cls"], dtype=np.int32).reshape(-1) gt_boxes = np.asarray(packet["gt_boxes"], dtype=np.float32).reshape(-1, 4) pred_cls_all = np.asarray(packet["pred_cls"], dtype=np.int32).reshape(-1) pred_boxes_all = np.asarray(packet["pred_boxes"], dtype=np.float32).reshape(-1, 4) pred_conf_all = np.asarray(packet["pred_conf"], dtype=np.float32).reshape(-1) gt_focus_mask = np.asarray(packet["gt_focus_mask"], dtype=bool).reshape(-1) pred_focus_mask_all = np.asarray(packet["pred_focus_mask"], dtype=bool).reshape(-1) keep = pred_conf_all > float(confidence_threshold) if pred_conf_all.size else np.zeros((0,), dtype=bool) pred_cls = pred_cls_all[keep] pred_boxes = pred_boxes_all[keep] pred_conf = pred_conf_all[keep] pred_focus_mask = pred_focus_mask_all[keep] overall_counts["gt_total"] += int(len(gt_cls)) overall_counts["pred_total"] += int(len(pred_cls)) focused_confusion_gt_total += int(np.sum(gt_focus_mask)) focused_confusion_pred_spatial_total += int(np.sum(pred_focus_mask)) for cls_id in gt_cls.tolist(): class_counts[int(cls_id)]["gt_total"] += 1 for cls_id in pred_cls.tolist(): class_counts[int(cls_id)]["pred_total"] += 1 confusion_matrix_2d.process_batch( detections={ "cls": torch.from_numpy(pred_cls_all.astype(np.float32)), "conf": torch.from_numpy(pred_conf_all.astype(np.float32)), "bboxes": torch.from_numpy(pred_boxes_all.astype(np.float32)), }, batch={ "cls": torch.from_numpy(gt_cls.astype(np.float32)), "bboxes": torch.from_numpy(gt_boxes.astype(np.float32)), }, conf=float(confidence_threshold), iou_thres=0.5, ) update_subset_confusion_matrix( focused_confusion_matrix_2d, gt_cls=gt_cls, gt_boxes=gt_boxes, pred_cls=pred_cls_all, pred_boxes=pred_boxes_all, pred_conf=pred_conf_all, gt_mask=gt_focus_mask, pred_mask=pred_focus_mask_all, conf=float(confidence_threshold), iou_thres=0.5, ) matches, iou_matrix = greedy_match_indices(gt_cls, gt_boxes, pred_cls, pred_boxes, iou_thr=0.5) fake_matches, fake_iou_matrix = greedy_match_indices_any_class(gt_boxes, pred_boxes, iou_thr=0.5) overall_counts["matched_2d"] += int(len(matches)) matched_gt = set(matches[:, 0].tolist()) if len(matches) else set() matched_pred = set(matches[:, 1].tolist()) if len(matches) else set() image_path = Path(str(packet["image_path"])) label_path = Path(str(packet["label_path"])) sample_index = int(packet["sample_index"]) for gt_index, pred_index in matches.tolist() if len(matches) else []: class_counts[int(gt_cls[int(gt_index)])]["matched_2d"] += 1 for gt_index, pred_index in fake_matches.tolist() if len(fake_matches) else []: gt_ci = int(gt_cls[int(gt_index)]) pred_ci = int(pred_cls[int(pred_index)]) gt_cn = get_cls_name(names_dict, gt_ci) pred_cn = get_cls_name(names_dict, pred_ci) gt_fake_label = fake_class_label(gt_cn) pred_fake_label = fake_class_label(pred_cn) add_label_accuracy_sample(fake_class_store_metrics, gt_fake_label, pred_fake_label) if gt_fake_label == pred_fake_label: continue match_iou = float(fake_iou_matrix[int(gt_index), int(pred_index)]) if fake_iou_matrix.size else 0.0 fake_record = make_fake_class_badcase_record( sample_index=sample_index, roi_name=roi_name, image_path=image_path, label_path=label_path, gt_index=int(gt_index), pred_index=int(pred_index), gt_cls_id=gt_ci, gt_cls_name=gt_cn, pred_cls_id=pred_ci, pred_cls_name=pred_cn, gt_bbox_xyxy=gt_boxes[int(gt_index)], pred_bbox_xyxy=pred_boxes[int(pred_index)], match_iou=match_iou, confidence=float(pred_conf[int(pred_index)]), ) fake_class_writer.writerow(fake_class_record_to_csv_row(fake_record)) reservoir_add(fake_class_store, fake_record, topk_badcases, rng) fake_class_counts[fake_record["kind"]] += 1 for gt_index, gt_box in enumerate(gt_boxes): if gt_index in matched_gt: continue cls_id = int(gt_cls[gt_index]) cls_name = get_cls_name(names_dict, cls_id) iou_row = iou_matrix[gt_index] if iou_matrix.shape[1] > 0 else np.zeros((0,), dtype=np.float32) same_class_mask = pred_cls == cls_id if len(pred_cls) else np.zeros((0,), dtype=bool) record_2d = make_2d_badcase_record( sample_index=sample_index, roi_name=roi_name, image_path=image_path, label_path=label_path, kind="false_negative", cls_id=cls_id, cls_name=cls_name, bbox_xyxy=gt_box, gt_index=gt_index, pred_index=None, confidence=None, max_iou_any=float(np.max(iou_row)) if iou_row.size else 0.0, max_iou_same_class=float(np.max(iou_row[same_class_mask])) if same_class_mask.any() else 0.0, ) bad_2d_fn_writer.writerow(record_2d_to_csv_row(record_2d)) reservoir_add(bad_2d_fn_store, record_2d, topk_badcases, rng) for pred_index, pred_box in enumerate(pred_boxes): if pred_index in matched_pred: continue cls_id = int(pred_cls[pred_index]) cls_name = get_cls_name(names_dict, cls_id) iou_col = iou_matrix[:, pred_index] if iou_matrix.shape[0] > 0 else np.zeros((0,), dtype=np.float32) same_class_mask = gt_cls == cls_id if len(gt_cls) else np.zeros((0,), dtype=bool) record_2d = make_2d_badcase_record( sample_index=sample_index, roi_name=roi_name, image_path=image_path, label_path=label_path, kind="false_positive", cls_id=cls_id, cls_name=cls_name, bbox_xyxy=pred_box, gt_index=None, pred_index=pred_index, confidence=float(pred_conf[pred_index]), max_iou_any=float(np.max(iou_col)) if iou_col.size else 0.0, max_iou_same_class=float(np.max(iou_col[same_class_mask])) if same_class_mask.any() else 0.0, ) bad_2d_fp_writer.writerow(record_2d_to_csv_row(record_2d)) reservoir_add(bad_2d_fp_store, record_2d, topk_badcases, rng) bad_2d_fn_handle.close() bad_2d_fp_handle.close() class_rows = {} for cls_id, counts in class_counts.items(): precision = rate_or_none(int(counts["matched_2d"]), int(counts["pred_total"])) recall = rate_or_none(int(counts["matched_2d"]), int(counts["gt_total"])) class_rows[int(cls_id)] = { "gt_total": int(counts["gt_total"]), "pred_total": int(counts["pred_total"]), "matched_2d": int(counts["matched_2d"]), "precision_2d": precision, "recall_2d": recall, "f1_2d": f1_or_none(precision, recall), "false_negatives_2d": max(0, int(counts["gt_total"]) - int(counts["matched_2d"])), "false_positives_2d": max(0, int(counts["pred_total"]) - int(counts["matched_2d"])), } overall_precision = rate_or_none(int(overall_counts["matched_2d"]), int(overall_counts["pred_total"])) overall_recall = rate_or_none(int(overall_counts["matched_2d"]), int(overall_counts["gt_total"])) overall_row = { "gt_total": int(overall_counts["gt_total"]), "pred_total": int(overall_counts["pred_total"]), "matched_2d": int(overall_counts["matched_2d"]), "precision_2d": overall_precision, "recall_2d": overall_recall, "f1_2d": f1_or_none(overall_precision, overall_recall), "false_negatives_2d": max(0, int(overall_counts["gt_total"]) - int(overall_counts["matched_2d"])), "false_positives_2d": max(0, int(overall_counts["pred_total"]) - int(overall_counts["matched_2d"])), } classification_summary_2d = summarize_2d_classification_from_confusion(confusion_matrix_2d) focused_classification_summary_2d = summarize_2d_classification_from_confusion(focused_confusion_matrix_2d) fake_class_summary = summarize_label_accuracy_store(fake_class_store_metrics) confusion_matrix_plot_path = roi_output / "confusion_matrix_normalized.png" confusion_matrix_2d.plot(normalize=True, save_dir=str(roi_output)) confusion_matrix_plot = str(confusion_matrix_plot_path) if confusion_matrix_plot_path.is_file() else None focused_confusion_output = roi_output / "confusion_matrix_easy_near_no_occlusion" focused_confusion_output.mkdir(parents=True, exist_ok=True) focused_confusion_matrix_plot_path = focused_confusion_output / "confusion_matrix_normalized.png" focused_confusion_matrix_2d.plot(normalize=True, save_dir=str(focused_confusion_output)) focused_confusion_plot = str(focused_confusion_matrix_plot_path) if focused_confusion_matrix_plot_path.is_file() else None bad_2d_fn_records = reservoir_records(bad_2d_fn_store, rng) bad_2d_fp_records = reservoir_records(bad_2d_fp_store, rng) fake_class_records = reservoir_records(fake_class_store, rng) bad_2d_fn_manifest = save_2d_badcase_visuals( bad_2d_fn_records, roi_output / "visuals" / "2d_false_negative", "2d_false_negative", bundle, image_root, ) bad_2d_fp_manifest = save_2d_badcase_visuals( bad_2d_fp_records, roi_output / "visuals" / "2d_false_positive", "2d_false_positive", bundle, image_root, ) fake_class_manifest = save_fake_class_badcase_visuals( fake_class_records, roi_output / "visuals" / "fake_class", "fake_class", bundle, image_root, ) fake_class_handle.close() return { "overall": overall_row, "per_class": class_rows, "classification_summary_2d": classification_summary_2d, "focused_classification_summary_2d": focused_classification_summary_2d, "confusion_matrix_plot_path_2d": confusion_matrix_plot, "focused_confusion_matrix_plot_path_2d": focused_confusion_plot, "focused_confusion_filter": { "max_abs_lateral_m": float(max_abs_lateral_m), "max_abs_longitudinal_m": float(max_abs_longitudinal_m), "required_difficulty": int(required_difficulty), "confidence_threshold": float(confidence_threshold), "gt_count": int(focused_confusion_gt_total), "pred_spatial_count": int(focused_confusion_pred_spatial_total), }, "fake_class_summary": { **fake_class_summary, "fake_vs_non_fake": int(fake_class_counts["fake_vs_non_fake"]), "fake_internal_confusion": int(fake_class_counts["fake_internal_confusion"]), "visible_internal_confusion": int(fake_class_counts["visible_internal_confusion"]), "saved_topk": len(fake_class_records), }, "badcase_counts": { "2d_false_negative_saved_topk": len(bad_2d_fn_records), "2d_false_positive_saved_topk": len(bad_2d_fp_records), "fake_class_saved_topk": len(fake_class_records), }, "badcase_manifest_paths": { "2d_false_negative": str(roi_output / "visuals" / "2d_false_negative" / "manifest.json"), "2d_false_positive": str(roi_output / "visuals" / "2d_false_positive" / "manifest.json"), "fake_class": str(roi_output / "visuals" / "fake_class" / "manifest.json"), }, "badcase_manifest_sizes": { "2d_false_negative": len(bad_2d_fn_manifest), "2d_false_positive": len(bad_2d_fp_manifest), "fake_class": len(fake_class_manifest), }, } def write_json(path: Path, payload: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8") as file: json.dump(payload, file, indent=2, ensure_ascii=False) def write_class_csv(path: Path, rows: list[dict[str, Any]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) if not rows: with path.open("w", encoding="utf-8", newline="") as file: writer = csv.writer(file) writer.writerow(["cls_id", "cls_name"]) return fieldnames = list(rows[0].keys()) with path.open("w", encoding="utf-8", newline="") as file: writer = csv.DictWriter(file, fieldnames=fieldnames) writer.writeheader() writer.writerows(rows) def write_markdown_summary(path: Path, roi_name: str, payload: dict[str, Any], model_path: str, split_file: str) -> None: overall = payload["overall"] threshold_advice_2d = payload.get("threshold_advice_2d") or {} confidence_curve_paths_2d = payload.get("confidence_curve_paths_2d") or {} focused_confusion_filter = payload.get("focused_confusion_filter") or {} focused_classification_summary_2d = payload.get("focused_classification_summary_2d") or {} face_selection_summary = payload.get("face_selection_summary") or {} occlusion_binary_summary = payload.get("occlusion_binary_summary") or {} large_vehicle_compare = payload.get("large_vehicle_compare") or {} large_vehicle_summary = large_vehicle_compare.get("summary") or {} yaw_contributors = payload["top_yaw_contributors"] horizontal = payload["top_horizontal_classes"] vertical = payload["top_vertical_classes"] worst_depth = payload["worst_depth_bins"] worst_diag = payload["worst_bbox_bins"] worst_faces = payload["worst_face_buckets"] lines = [ f"# {roi_name.upper()} Validation Error Analysis", "", f"- model: `{model_path}`", f"- split_file: `{split_file}`", f"- matched_2d: {overall['matched_2d']}", f"- matched_3d: {overall['matched_3d']}", f"- yaw_mae_deg: {overall['yaw_mae_deg'] if overall['yaw_mae_deg'] is not None else 'n/a'}", f"- x_abs_mae_m: {overall['x_abs_mae_m'] if overall['x_abs_mae_m'] is not None else 'n/a'}", f"- z_abs_mae_m: {overall['z_abs_mae_m'] if overall['z_abs_mae_m'] is not None else 'n/a'}", f"- configured_2d_confidence: {format_float(to_float(payload.get('configured_confidence_2d')), 3)}", f"- recommended_2d_confidence: {format_float(to_float(threshold_advice_2d.get('recommended_confidence')), 3)}", f"- position_error_basis: face center for face_3d classes; box center otherwise", f"- 2d_precision@recommended_conf: {format_percent(overall.get('precision_2d'))}", f"- 2d_recall@recommended_conf: {format_percent(overall.get('recall_2d'))}", f"- 2d_f1@recommended_conf: {format_percent(overall.get('f1_2d'))}", f"- mean_class_f1_for_threshold_advice: {format_percent(threshold_advice_2d.get('mean_f1'))}", f"- focused_2d_gt_count: {int(focused_confusion_filter.get('gt_count', 0) or 0)}", f"- focused_2d_cls_acc: {focused_classification_summary_2d.get('overall_accuracy') if focused_classification_summary_2d.get('overall_accuracy') is not None else 'n/a'}", f"- focused_2d_longitudinal_limit_m: {format_float(to_float(focused_confusion_filter.get('max_abs_longitudinal_m')), 1)}", f"- face_selection_acc: {format_percent(face_selection_summary.get('overall_accuracy'))}", f"- face_selection_mean_acc: {format_percent(face_selection_summary.get('mean_accuracy'))}", f"- occlusion_binary_acc: {format_percent(occlusion_binary_summary.get('overall_accuracy'))}", f"- large_vehicle_scope: {large_vehicle_compare.get('class_scope', LARGE_VEHICLE_CLASS_SCOPE_TEXT)}", f"- large_vehicle_two_face_pairs: {int(large_vehicle_summary.get('yaw_compare_count', 0) or 0)}", f"- large_vehicle_edge_better_rate: {large_vehicle_summary.get('yaw_compare_edge_better_rate') if large_vehicle_summary.get('yaw_compare_edge_better_rate') is not None else 'n/a'}", "", "## 2D Confidence Advice", "", f"- Detection-dependent report results use `conf > {format_float(to_float(payload.get('report_confidence_2d')), 3)}`.", f"- Threshold advice source: `{threshold_advice_2d.get('source', 'n/a')}` from the ROI F1-confidence sweep instead of the configured default `{format_float(to_float(payload.get('configured_confidence_2d')), 3)}`.", f"- Mean class precision / recall / F1 at the advised threshold: {format_percent(threshold_advice_2d.get('mean_precision'))} / {format_percent(threshold_advice_2d.get('mean_recall'))} / {format_percent(threshold_advice_2d.get('mean_f1'))}.", "", "## Why Yaw Error Is Large", "", "Top class contributors by total yaw error:", ] for row in yaw_contributors: lines.append( f"- {row['cls_name']} (id={row['cls_id']}): matched_3d={row['matched_3d']}, " f"yaw_mae_deg={row['yaw_mae_deg']}, yaw_contribution={row['yaw_contribution_rate']}" ) lines.extend(["", "Worst depth bins by yaw MAE:"]) for key, row in worst_depth: lines.append(f"- {key}: matched_3d={row['matched_3d']}, yaw_mae_deg={row['yaw_mae_deg']}") lines.extend(["", "Worst box-size bins by yaw MAE:"]) for key, row in worst_diag: lines.append(f"- {key}: matched_3d={row['matched_3d']}, yaw_mae_deg={row['yaw_mae_deg']}") lines.extend(["", "Worst face-visibility buckets by yaw MAE:"]) for key, row in worst_faces: lines.append( f"- {key}: matched_3d={row['matched_3d']}, yaw_mae_deg={row['yaw_mae_deg']}, " f"edge_based_yaw_mae_deg={row['edge_based_yaw_mae_deg']}" ) lines.extend(["", "## Bad Horizontal Errors (> threshold)", ""]) for row in horizontal: lines.append(f"- {row['cls_name']} (id={row['cls_id']}): x_bad_count={row['x_bad_count']}, x_bad_rate={row['x_bad_rate']}") lines.extend(["", "## Bad Vertical Errors (> threshold)", ""]) for row in vertical: lines.append(f"- {row['cls_name']} (id={row['cls_id']}): z_bad_count={row['z_bad_count']}, z_bad_rate={row['z_bad_rate']}") lines.extend( [ "", "## Extra Artifacts", "", f"- 2d_f1_curve: `{confidence_curve_paths_2d.get('f1_curve', 'n/a')}`", f"- focused_confusion_matrix: `{payload.get('focused_confusion_matrix_plot_path_2d', 'n/a')}`", f"- html_report: `{path.parent.parent / 'report.html'}`", ] ) path.parent.mkdir(parents=True, exist_ok=True) path.write_text("\n".join(lines) + "\n", encoding="utf-8") def write_markdown_summary_zh( path: Path, roi_name: str, payload: dict[str, Any], model_path: str, split_file: str, horizontal_csv: Path, vertical_csv: Path, yaw_depth_csv: Path, yaw_heading_csv: Path, ) -> None: overall = payload["overall"] threshold_advice_2d = payload.get("threshold_advice_2d") or {} confidence_curve_paths_2d = payload.get("confidence_curve_paths_2d") or {} focused_confusion_filter = payload.get("focused_confusion_filter") or {} focused_classification_summary_2d = payload.get("focused_classification_summary_2d") or {} face_selection_summary = payload.get("face_selection_summary") or {} occlusion_binary_summary = payload.get("occlusion_binary_summary") or {} large_vehicle_compare = payload.get("large_vehicle_compare") or {} large_vehicle_summary = large_vehicle_compare.get("summary") or {} per_class_insights = payload["per_class_interval_insights"] lines = [ f"# {roi_name.upper()} 验证集误差统计分析", "", "## 统计口径", "", f"- 模型: `{model_path}`", f"- 验证列表: `{split_file}`", "- 匹配规则: 同类别 + IoU>=0.5,与训练期 validator 保持一致。", "- 区间解释: 这里把“每 1m / 5m / 10m interval”理解为按 GT 前向距离 `z_gt` 分桶。", "- 水平误差: `|pred_x - gt_x|`,按 GT 横向位置 `x_gt` 在 [-30m, 30m) 内做 5m 分桶统计。", "- 垂直误差: `|pred_y - gt_y|`,按 5m 深度分桶统计。", "- 偏航误差: `|yaw_pred - yaw_gt|`,按 10m 深度分桶统计均值。", "- GT 偏航分桶: 额外提供按 GT yaw 每 10deg 分桶的偏航绝对误差与相对误差,其中相对误差按 `|yaw_err| / |gt_yaw|` 统计。", "- 位置误差口径: `face_3d_classes` 用 face center,其他类别用 whole-box center。", "", "## 整体结果", "", f"- matched_3d: {overall['matched_3d']}", f"- yaw_mae_deg: {format_float(overall['yaw_mae_deg'], 3)}", f"- x_abs_mae_m: {format_float(overall['x_abs_mae_m'], 3)}", f"- z_abs_mae_m: {format_float(overall['z_abs_mae_m'], 3)}", f"- 当前配置 2D conf: {format_float(to_float(payload.get('configured_confidence_2d')), 3)}", f"- 建议 2D conf: {format_float(to_float(threshold_advice_2d.get('recommended_confidence')), 3)}", f"- 建议阈值下 2D precision / recall / F1: {format_percent(overall.get('precision_2d'))} / {format_percent(overall.get('recall_2d'))} / {format_percent(overall.get('f1_2d'))}", f"- F1 阈值建议对应的 mean class F1: {format_percent(threshold_advice_2d.get('mean_f1'))}", f"- 近距离无遮挡 GT 数: {int(focused_confusion_filter.get('gt_count', 0) or 0)}", f"- 近距离无遮挡 2D 分类准确率: {format_percent(focused_classification_summary_2d.get('overall_accuracy'))}", f"- 近距离无遮挡纵向范围: |z|<{format_float(to_float(focused_confusion_filter.get('max_abs_longitudinal_m')), 1)}m", f"- face/cut 选择准确率: {format_percent(face_selection_summary.get('overall_accuracy'))},mean acc: {format_percent(face_selection_summary.get('mean_accuracy'))}", f"- 遮挡二分类准确率: {format_percent(occlusion_binary_summary.get('overall_accuracy'))}", f"- 大车对比范围: {large_vehicle_compare.get('class_scope', LARGE_VEHICLE_CLASS_SCOPE_TEXT)}", f"- 大车 two-face 对比样本: {int(large_vehicle_summary.get('yaw_compare_count', 0) or 0)}", f"- 大车 edge 更优占比: {format_percent(large_vehicle_summary.get('yaw_compare_edge_better_rate'))}", f"- yaw>5deg 占比: {format_percent(overall['yaw_bad_rate'])}", f"- horizontal>0.5m 占比: {format_percent(overall['x_bad_rate'])}", f"- vertical(z)>0.5m 占比: {format_percent(overall['z_bad_rate'])}", "", "## 2D 阈值建议", "", f"- 本报告里所有依赖检测结果的统计统一使用 `conf > {format_float(to_float(payload.get('report_confidence_2d')), 3)}`。", f"- 建议来源: ROI 的 F1-confidence 曲线最大 mean class F1,而不是默认 `{format_float(to_float(payload.get('configured_confidence_2d')), 3)}`。", f"- 阈值建议曲线: `{confidence_curve_paths_2d.get('f1_curve', 'n/a')}`", "", "## 分类别结论", "", ] for item in per_class_insights: lines.extend( [ f"### {item['cls_name']} (id={item['cls_id']})", "", f"- 匹配样本: 3D={item['matched_3d']}, 位置有效={item['matched_pos']}", f"- 全局误差: yaw={format_float(item['yaw_mae_deg'], 2)}deg, " f"x={format_float(item['x_abs_mae_m'], 3)}m, z={format_float(item['z_abs_mae_m'], 3)}m", f"- 超阈值占比: yaw>5deg={format_percent(item['yaw_bad_rate'])}, " f"x>0.5m={format_percent(item['x_bad_rate'])}, z>0.5m={format_percent(item['z_bad_rate'])}", f"- 水平误差最差横向 5m 区间: {item['horizontal_bins_text']}", f"- 垂直误差最差 5m 深度区间: {item['vertical_bins_text']}", f"- 偏航误差最差 10m 深度区间: {item['yaw_bins_text']}", "", ] ) lines.extend( [ "## 明细文件", "", f"- 水平误差横向 5m 分桶: `{horizontal_csv}`", f"- 垂直误差 5m 深度分桶: `{vertical_csv}`", f"- 偏航误差 10m 深度分桶: `{yaw_depth_csv}`", f"- 偏航误差 GT yaw 10deg 分桶: `{yaw_heading_csv}`", f"- 偏航对比横向 5m 分桶: `{path.parent / 'yaw_compare_signed_lateral_5m.csv'}`", f"- 偏航对比分 face-visibility 横向 5m 分桶: `{path.parent / 'yaw_compare_signed_lateral_5m_by_face_visibility.csv'}`", f"- face/cut 选择准确率: `{path.parent / 'face_selection_accuracy.csv'}`", f"- 遮挡二分类准确率: `{path.parent / 'occlusion_binary_accuracy.csv'}`", f"- 分类别整体指标: `{path.parent / 'class_metrics.csv'}`", f"- 2D F1 曲线: `{confidence_curve_paths_2d.get('f1_curve', 'n/a')}`", f"- 近距离无遮挡 confusion matrix: `{payload.get('focused_confusion_matrix_plot_path_2d', 'n/a')}`", f"- 坏例汇总: `{path.parent / 'bad_cases_yaw.csv'}`, `{path.parent / 'bad_cases_horizontal.csv'}`, `{path.parent / 'bad_cases_vertical.csv'}`", f"- 可视化目录: `{path.parent / 'visuals'}`", ] ) path.parent.mkdir(parents=True, exist_ok=True) path.write_text("\n".join(lines) + "\n", encoding="utf-8") def sort_nullable_desc(rows: Iterable[tuple[str, dict[str, Any]]], key_name: str) -> list[tuple[str, dict[str, Any]]]: def key_fn(item): value = item[1].get(key_name) if value is None: return float("-inf") return float(value) return sorted(rows, key=key_fn, reverse=True) def write_combined_summary_zh(path: Path, data_yaml: Path, split_path: Path, image_root: Path, summary_by_roi: dict[str, Any]) -> None: lines = [ "# 双 ROI 验证集误差统计总结", "", "## 数据与口径", "", f"- 数据配置: `{data_yaml}`", f"- 验证列表: `{split_path}`", f"- 图像根目录: `{image_root}`", "- 匹配规则: 同类别 + IoU>=0.5。", "- 区间解释: 1m / 5m / 10m interval 均按 GT 前向距离 `z_gt` 分桶。", "- 坏例导出规则: yaw 使用 `>5deg`,位置使用 `>0.5m`;图像导出采用分桶抽样,CSV 保留全部超阈值样本。", "", "## ROI 总览", "", ] for roi_name, payload in summary_by_roi.items(): overall = payload["overall"] threshold_advice_2d = payload.get("threshold_advice_2d") or {} focused_confusion_filter = payload.get("focused_confusion_filter") or {} lines.extend( [ f"### {roi_name.upper()}", "", f"- matched_3d={overall['matched_3d']}, matched_pos={overall['matched_pos']}", f"- 2D 建议阈值={format_float(to_float(threshold_advice_2d.get('recommended_confidence')), 3)}, " f"precision={format_percent(overall.get('precision_2d'))}, recall={format_percent(overall.get('recall_2d'))}, F1={format_percent(overall.get('f1_2d'))}", f"- 近距离无遮挡纵向范围: |z|<{format_float(to_float(focused_confusion_filter.get('max_abs_longitudinal_m')), 1)}m", f"- yaw_mae={format_float(overall['yaw_mae_deg'], 3)}deg, yaw>5deg={format_percent(overall['yaw_bad_rate'])}", f"- x_mae={format_float(overall['x_abs_mae_m'], 3)}m, x>0.5m={format_percent(overall['x_bad_rate'])}", f"- z_mae={format_float(overall['z_abs_mae_m'], 3)}m, z>0.5m={format_percent(overall['z_bad_rate'])}", f"- yaw 主要贡献类别: " + " / ".join( f"{row['cls_name']}({format_float(row.get('yaw_mae_deg'), 1)}deg, 占比{format_percent(row.get('yaw_contribution_rate'))})" for row in payload["top_yaw_contributors"][:4] if row.get("matched_3d") ), "", ] ) lines.extend(["## 分 ROI 细结论", ""]) for roi_name, payload in summary_by_roi.items(): lines.extend( [ f"### {roi_name.upper()}", "", f"- 中文详版: `{path.parent / roi_name / 'summary_zh.md'}`", f"- 英文简版: `{path.parent / roi_name / 'summary.md'}`", ] ) for item in payload["per_class_interval_insights"]: lines.extend( [ f"- {item['cls_name']}: yaw={format_float(item['yaw_mae_deg'], 2)}deg, " f"x={format_float(item['x_abs_mae_m'], 3)}m, z={format_float(item['z_abs_mae_m'], 3)}m; " f"最差水平区间={item['horizontal_bins_text']}; " f"最差垂直区间={item['vertical_bins_text']}; " f"最差偏航区间={item['yaw_bins_text']}", ] ) lines.append("") path.parent.mkdir(parents=True, exist_ok=True) path.write_text("\n".join(lines) + "\n", encoding="utf-8") def interval_row_key(row: dict[str, Any]) -> tuple[int, str, float]: return ( int(row["cls_id"]), str(row["cls_name"]), float(row.get("interval_start", row.get("depth_bin_start_m", 0.0))), ) def compute_interval_top_keys(rows: list[dict[str, Any]], metric_key: str, topn: int = 3, min_count: int = 30) -> set[tuple[int, str, float]]: top_keys: set[tuple[int, str, float]] = set() for _, class_rows in group_rows_by_class(rows).items(): for row in top_interval_rows(class_rows, metric_key=metric_key, topn=topn, min_count=min_count): top_keys.add(interval_row_key(row)) return top_keys def build_yaw_compare_diagnostics( overall_summary: dict[str, Any], breakdown_summaries: dict[str, dict[str, dict[str, Any]]], ) -> dict[str, Any]: focus_bucket = "two-face" face_visibility_summary = breakdown_summaries.get("face_visibility", {}) face_rows = [] for bucket_name in FACE_VISIBILITY_BUCKET_ORDER: row = dict(face_visibility_summary.get(bucket_name) or summarize_metric_bucket(make_metric_bucket())) row["key"] = bucket_name face_rows.append(row) face_row_map = {str(row["key"]): row for row in face_rows} dominant_face = next((row for row in face_rows if int(row.get("yaw_compare_count") or 0) > 0), None) focus_face = face_row_map.get(focus_bucket, {}) compare_count = int(focus_face.get("yaw_compare_count") or 0) matched_3d = int(overall_summary.get("matched_3d") or 0) subset_rate = rate_or_none(compare_count, matched_3d) explanation_parts = [ "All yaw errors in this report use pi-periodic orientation error, so 180-degree front-rear flips are treated as near-zero orientation error.", "Both direct-regression and edge-based yaw metrics in this report are compared against gt_yaw.", "The face-visibility diagnostic splits paired samples into front_rear_only, side only, and two-face buckets.", "For front_rear_only and side only buckets, the diagnostic uses the best available single-face edge yaw so those rows can still compare direct versus edge-based yaw.", f"Signed lateral yaw-compare bins use {format_float(HORIZONTAL_LATERAL_BIN_M, 1)}m bins inside the configured lateral range.", ( f"In this ROI, the paired subset is {compare_count}/{matched_3d} " f"({format_percent(subset_rate)}) of matched_3d." ), ] if focus_face: explanation_parts.append( f"The focused two-face bucket has direct={format_float(focus_face.get('direct_regression_yaw_mae_deg'), 2)} deg and " f"edge={format_float(focus_face.get('edge_based_yaw_mae_deg'), 2)} deg over {compare_count} pairs." ) if int(focus_face.get("length_compare_count") or 0) > 0: explanation_parts.append( f"Its side-edge length MAE is {format_float(focus_face.get('side_edge_length_mae_m'), 3)} m versus " f"{format_float(focus_face.get('direct_regression_length_mae_m'), 3)} m from direct regression." ) if dominant_face is not None: explanation_parts.append( f"The largest face-visibility bucket is {dominant_face['key']} with {dominant_face['yaw_compare_count']} pairs " f"(direct={format_float(dominant_face.get('direct_regression_yaw_mae_deg'), 2)} deg, " f"edge={format_float(dominant_face.get('edge_based_yaw_mae_deg'), 2)} deg)." ) return { "focus_bucket": focus_bucket, "focus_face_row": focus_face, "compare_count": compare_count, "matched_3d": matched_3d, "paired_subset_rate": subset_rate, "overall_yaw_mae_deg": overall_summary.get("yaw_mae_deg"), "face_visibility_rows": face_rows, "explanation": " ".join(part for part in explanation_parts if part), } def html_escape(value: Any) -> str: return html_lib.escape("" if value is None else str(value), quote=True) def manual_summary_key(value: Any, fallback: str = "manual-summary") -> str: slug = "".join(char.lower() if char.isalnum() else "-" for char in ("" if value is None else str(value)).strip()) while "--" in slug: slug = slug.replace("--", "-") slug = slug.strip("-") return slug or fallback def render_manual_summary_slot(section_title: Optional[str] = None) -> str: section_label = str(section_title).strip() if section_title else "Manual Summary" section_key = manual_summary_key(section_label) return ( f"
" f"" "
" ) def format_bool(value: Any) -> str: return "yes" if bool(value) else "no" def relative_href(report_path: Path, target_path: str | Path) -> str: return html_escape(os.path.relpath(str(Path(target_path)), start=str(report_path.parent))) def load_manifest_records(path: str | Path) -> list[dict[str, Any]]: manifest_path = Path(path) if not manifest_path.is_file(): return [] with manifest_path.open("r", encoding="utf-8") as file: payload = json.load(file) return payload if isinstance(payload, list) else [] def render_html_table(title: str, headers: list[str], rows: list[list[str]], table_class: str = "report-table") -> str: if not rows: return ( f"

{html_escape(title)}

" f"{render_manual_summary_slot(title)}

No data.

" ) head_html = "".join(f"{html_escape(header)}" for header in headers) body_html = [] for row in rows: body_html.append("" + "".join(f"{cell}" for cell in row) + "") return ( f"

{html_escape(title)}

" f"{render_manual_summary_slot(title)}" f"
{head_html}" f"{''.join(body_html)}
" ) def render_metric_value_table(title: str, rows: list[tuple[str, str]]) -> str: return render_html_table(title, ["Metric", "Value"], [[html_escape(metric), value] for metric, value in rows], table_class="metric-table") def _pie_palette() -> list[str]: return ["#0e7490", "#1f6f50", "#c26d1a", "#8b5e3c", "#8e3b46", "#5b6cfa", "#0f766e", "#7c3aed", "#64748b"] def build_pie_segments( rows: list[dict[str, Any]], label_key: str, value_key: str, max_segments: int = 7, ) -> tuple[list[dict[str, Any]], int]: nonzero = [ {"label": str(row.get(label_key, "")), "count": max(0, int(row.get(value_key, 0) or 0))} for row in rows if max(0, int(row.get(value_key, 0) or 0)) > 0 ] total = int(sum(int(item["count"]) for item in nonzero)) if not nonzero or total <= 0: return [], total ranked = sorted(nonzero, key=lambda item: (int(item["count"]), item["label"]), reverse=True) if len(ranked) > max_segments: kept = ranked[: max_segments - 1] other_count = int(sum(int(item["count"]) for item in ranked[max_segments - 1 :])) if other_count > 0: kept.append({"label": "Other", "count": other_count}) ranked = kept return ranked, total def polar_to_cartesian(cx: float, cy: float, radius: float, angle_deg: float) -> tuple[float, float]: angle_rad = math.radians(float(angle_deg)) return float(cx + radius * math.cos(angle_rad)), float(cy + radius * math.sin(angle_rad)) def pie_slice_path(cx: float, cy: float, radius: float, start_angle_deg: float, end_angle_deg: float) -> str: start_x, start_y = polar_to_cartesian(cx, cy, radius, start_angle_deg) end_x, end_y = polar_to_cartesian(cx, cy, radius, end_angle_deg) large_arc = 1 if (end_angle_deg - start_angle_deg) > 180.0 else 0 return ( f"M {cx:.1f} {cy:.1f} " f"L {start_x:.1f} {start_y:.1f} " f"A {radius:.1f} {radius:.1f} 0 {large_arc} 1 {end_x:.1f} {end_y:.1f} Z" ) def render_count_bar_svg( rows: list[dict[str, Any]], label_key: str, value_key: str, x_axis_title: str, y_axis_title: str, aria_label: str, fill_color: str = "#0e7490", ) -> str: if not rows: return "

No distribution data.

" ordered = list(rows) pie_segments, total_count = build_pie_segments(ordered, label_key, value_key) max_value = max((max(0, int(row.get(value_key, 0) or 0)) for row in ordered), default=0) max_value = max(1, max_value) left = 74 right = 16 top = 18 bottom = 96 plot_height = 190 bar_pitch = 30 if len(ordered) <= 24 else 26 plot_width = max(760, len(ordered) * bar_pitch) chart_right = left + plot_width pie_panel_x = chart_right + 26 pie_panel_width = 320 bar_width = max(10.0, min(22.0, bar_pitch * 0.72)) major_tick_count = 4 pie_radius = 74.0 pie_center_x = pie_panel_x + 104.0 pie_center_y = top + 92.0 legend_y = pie_center_y + pie_radius + 18.0 legend_step = 18.0 height = int(max(top + plot_height + bottom, legend_y + max(1, len(pie_segments)) * legend_step + 20.0)) width = int(chart_right + pie_panel_width + right) palette = _pie_palette() parts = [ f"", f"", f"", f"", ] for tick_idx in range(major_tick_count + 1): tick_value = max_value * tick_idx / float(major_tick_count) y = top + plot_height - (tick_value / max_value) * plot_height parts.append(f"") parts.append( f"{html_escape(str(int(round(tick_value))))}" ) label_stride = max(1, math.ceil(len(ordered) / 24)) for idx, row in enumerate(ordered): count = max(0, int(row.get(value_key, 0) or 0)) label = str(row.get(label_key, "")) bar_height = (count / max_value) * plot_height if max_value > 0 else 0.0 x = left + idx * bar_pitch + (bar_pitch - bar_width) / 2.0 y = top + plot_height - bar_height title = f"{label}\ncount={count}" parts.append( f"" f"{html_escape(title)}" ) if count > 0: text_y = max(top + 11.0, y - 6.0) parts.append( f"{html_escape(str(count))}" ) if len(ordered) <= 12 or idx % label_stride == 0 or count == max_value: label_y = top + plot_height + 14 parts.append( f"{html_escape(label)}" ) if pie_segments and total_count > 0: angle_start = -90.0 for index, segment in enumerate(pie_segments): ratio = float(segment["count"]) / float(total_count) angle_end = angle_start + ratio * 360.0 color = "#94a3b8" if segment["label"] == "Other" else palette[index % len(palette)] if math.isclose(ratio, 1.0, rel_tol=0.0, abs_tol=1e-9): parts.append( f"" f"{html_escape(segment['label'])}: {segment['count']} ({format_percent(ratio)})" ) else: parts.append( f"" f"{html_escape(segment['label'])}: {segment['count']} ({format_percent(ratio)})" ) angle_start = angle_end parts.append(f"") parts.append( f"Total" ) parts.append( f"{html_escape(str(total_count))}" ) for index, segment in enumerate(pie_segments): color = "#94a3b8" if segment["label"] == "Other" else palette[index % len(palette)] ratio = float(segment["count"]) / float(total_count) legend_row_y = legend_y + index * legend_step parts.append( f"" ) parts.append( f"" f"{html_escape(segment['label'])}: {html_escape(str(segment['count']))} ({html_escape(format_percent(ratio))})" "" ) else: parts.append( f"No nonzero slices" ) parts.append( f"{html_escape(x_axis_title)}" ) parts.append( f"{html_escape(y_axis_title)}" ) parts.append( f"Percentage distribution" ) parts.append("") return "".join(parts) def render_distribution_cards_section( title: str, cards: list[dict[str, Any]], label_key: str, value_key: str, x_axis_title: str, y_axis_title: str, fill_color: str, note: str = "", ) -> str: if not cards: return ( f"" ) note_html = f"

{html_escape(note)}

" if note else "" card_html = [] for card in cards: subtitle = str(card.get("subtitle", "") or "") subtitle_html = f"{html_escape(subtitle)}" if subtitle else "" card_html.append( "
" f"

{html_escape(str(card.get('title', 'Untitled')))}{subtitle_html}

" f"{render_count_bar_svg(card.get('rows', []), label_key, value_key, x_axis_title, y_axis_title, title, fill_color=fill_color)}" "
" ) return ( f"" ) def render_collapsible_distribution_groups( title: str, groups: list[dict[str, Any]], label_key: str, value_key: str, x_axis_title: str, y_axis_title: str, fill_color: str, note: str = "", ) -> str: if not groups: return ( f"" ) note_html = f"

{html_escape(note)}

" if note else "" group_html = [] for group in groups: rows = group.get("rows", []) if not rows: continue subtitle = str(group.get("subtitle", "") or "") subtitle_html = f"{html_escape(subtitle)}" if subtitle else "" group_html.append( f"
{html_escape(str(group.get('summary', 'group')))}" "
" "
" f"

{html_escape(str(group.get('title', group.get('summary', 'group'))))}{subtitle_html}

" f"{render_count_bar_svg(rows, label_key, value_key, x_axis_title, y_axis_title, str(group.get('title', title)), fill_color=fill_color)}" "
" "
" ) if not group_html: return ( f"" ) return ( f"" ) def render_data_portrait_section(portrait_payload: dict[str, Any]) -> str: split_name = str(portrait_payload.get("split", "train")) split_label = split_name.upper() summary = portrait_payload.get("summary") or {} vehicle_rows = portrait_payload.get("vehicle_rows", []) class_total_rows = portrait_payload.get("class_total_rows", []) class_vehicle_groups = portrait_payload.get("class_vehicle_groups", []) vehicle_cards = [ { "title": "All Vehicles", "subtitle": ( f"total_frames={int(summary.get('num_entries', 0) or 0)} " f"vehicles={int(summary.get('vehicles', 0) or 0)}" ), "rows": [{"vehicle_alias": str(row.get("vehicle_alias", "")), "frame_count": int(row.get("frame_count", 0) or 0)} for row in vehicle_rows], } ] class_total_cards = [ { "title": "All Vehicles", "subtitle": ( f"class_frame_hits={sum(int(row.get('frame_count', 0) or 0) for row in class_total_rows)} " f"objects={sum(int(row.get('object_count', 0) or 0) for row in class_total_rows)}" ), "rows": [ {"cls_name": str(row.get("cls_name", "")), "frame_count": int(row.get("frame_count", 0) or 0)} for row in class_total_rows if int(row.get("frame_count", 0) or 0) > 0 ], } ] class_group_charts = [ { "summary": str(group.get("summary", "group")), "title": str(group.get("vehicle_alias", group.get("summary", "group"))), "subtitle": ( f"class_frame_hits={sum(int(row.get('frame_count', 0) or 0) for row in group.get('rows', []))} " f"objects={sum(int(row.get('object_count', 0) or 0) for row in group.get('rows', []))}" ), "rows": [ {"cls_name": str(row.get("cls_name", "")), "frame_count": int(row.get("frame_count", 0) or 0)} for row in group.get("rows", []) if int(row.get("frame_count", 0) or 0) > 0 ], } for group in class_vehicle_groups ] portrait_overview_html = ( f"" ) return ( "
" f"

{html_escape(split_label)} Portrait

" "

" "Frame-level statistics decode vehicle alias and capture time from GT filenames. " "Object-level statistics reuse the dataset class map and GT 3D decoding logic from this report pipeline." "

" f"{portrait_overview_html}" f"{render_distribution_cards_section('Frames Per Vehicle', vehicle_cards, 'vehicle_alias', 'frame_count', 'Vehicle alias', 'Frames', '#0e7490', note='Each vehicle bar shows frame count, while the pie chart summarizes the frame-share percentage by vehicle alias.')}" f"{render_distribution_cards_section('Frames Per Day', portrait_payload.get('daily_cards', []), 'day', 'frame_count', 'Capture day', 'Frames', '#0e7490', note='The first card is the total distribution across all vehicles; following cards keep the same day axis per vehicle.')}" f"{render_distribution_cards_section('Frames Per Hour', portrait_payload.get('hourly_cards', []), 'hour', 'frame_count', 'Hour of day', 'Frames', '#1f6f50', note='Hours are aggregated by local capture hour decoded from the GT filename timestamp.')}" f"{render_distribution_cards_section('Class Frame Distribution Across All Vehicles', class_total_cards, 'cls_name', 'frame_count', 'Object class', 'Frames', '#8e3b46', note='This distribution uses frame_count, meaning each class counts how many frames contain at least one instance of that class.')}" f"{render_collapsible_distribution_groups('Class Frame Distribution By Vehicle', class_group_charts, 'cls_name', 'frame_count', 'Object class', 'Frames', '#8e3b46', note='Each collapsed vehicle card uses frame_count for the histogram and pie chart; subtitles also show the corresponding object totals.')}" f"{render_distribution_cards_section('Heading Distribution (10deg bins)', portrait_payload.get('heading_cards', []), 'heading_bin_label', 'count', 'Heading bin', 'Objects', '#c26d1a', note='Heading uses GT yaw wrapped into [-180, 180) and counted with 10-degree bins. The first card is the total without vehicle alias.')}" f"{render_distribution_cards_section('Lateral Distance By Class (|x|, 5m bins)', portrait_payload.get('lateral_cards', []), 'lateral_bin_label', 'count', 'Absolute lateral distance bin', 'Objects', '#0f766e', note='Each card aggregates all vehicles for one class; lateral distance uses the absolute GT x position in meters.')}" f"{render_distribution_cards_section('Longitudinal Distance By Class (z, 10m bins)', portrait_payload.get('longitudinal_cards', []), 'longitudinal_bin_label', 'count', 'Longitudinal distance bin', 'Objects', '#8b5e3c', note='Each card aggregates all vehicles for one class; longitudinal distance uses GT depth z in meters.')}" "
" ) def highlight_if_needed(value: str, highlight: bool) -> str: if not highlight: return value return f"{value}" def semantic_emphasis(value: str, css_class: Optional[str]) -> str: if not css_class: return value return f"{value}" def format_signed_float(value: Optional[float], digits: int = 2) -> str: if value is None or not math.isfinite(value): return "n/a" return f"{value:+.{digits}f}" def render_edge_gain_deg(value: Optional[float], digits: int = 2) -> str: text = html_escape(format_signed_float(value, digits)) if value is None or not math.isfinite(value): return text if value < 0: return semantic_emphasis(text, "emphasis-green") if value > 0: return semantic_emphasis(text, "emphasis-red") return text def render_rate_vs_half(value: Optional[float], digits: int = 1) -> str: text = html_escape(format_percent(value, digits)) if value is None or not math.isfinite(value): return text if value > 0.5: return semantic_emphasis(text, "emphasis-green") if value < 0.5: return semantic_emphasis(text, "emphasis-red") return text def top_bad_class_ids( rows: list[dict[str, Any]], metric_key: str, *, higher_is_worse: bool, support_key: Optional[str] = None, topn: int = 3, ) -> set[int]: ranked: list[tuple[float, int, int]] = [] for row in rows: cls_id = int(row.get("cls_id", -1)) value = row.get(metric_key) if value is None: continue value_f = to_float(value) if value_f is None: continue support = int(row.get(support_key) or 0) if support_key else 0 if support_key and support <= 0: continue ranked.append((float(value_f), support, cls_id)) if higher_is_worse: ranked.sort(key=lambda item: (item[0], item[1], -item[2]), reverse=True) else: ranked.sort(key=lambda item: (item[0], -item[1], item[2])) return {cls_id for _, _, cls_id in ranked[:topn]} def render_interval_trend_svg( rows: list[dict[str, Any]], metric_key: str, top_keys: set[tuple[int, str, float]], unit_suffix: str, x_axis_title: str = "Depth bin", relative_metric_key: Optional[str] = None, relative_y_axis_title: str = "Relative error (%)", relative_reference_label: str = "gt_bin", relative_reference_unit: str = "m", ) -> str: valid_rows = [row for row in rows if row.get(metric_key) is not None] if not valid_rows: return "

No interval data.

" ordered = sorted(valid_rows, key=lambda row: float(row.get("interval_start", row.get("depth_bin_start_m", 0.0)))) max_value = max(float(row[metric_key]) for row in ordered) if not math.isfinite(max_value) or max_value <= 0: max_value = 1.0 show_relative = bool(relative_metric_key) and any(to_float(row.get(relative_metric_key)) is not None for row in ordered) max_relative_value = 1.0 if show_relative: max_relative_value = max(float(to_float(row.get(relative_metric_key)) or 0.0) for row in ordered) if not math.isfinite(max_relative_value) or max_relative_value <= 0: max_relative_value = 1.0 left = 74 right = 24 top = 18 plot_height = 166 if show_relative else 190 relative_gap = 34 if show_relative else 0 relative_height = 104 if show_relative else 0 bottom = 98 if show_relative else 88 point_pitch = 26 width = max(860, left + right + len(ordered) * point_pitch) height = top + plot_height + relative_gap + relative_height + bottom plot_width = width - left - right point_radius = 3.5 major_tick_count = 4 minor_subdivisions = 4 relative_tick_count = 3 abs_bottom = top + plot_height relative_top = abs_bottom + relative_gap relative_bottom = relative_top + relative_height x_label_y = (relative_bottom if show_relative else abs_bottom) + 12 bar_width = min(18.0, point_pitch * 0.72) def scaled_y(value: float, max_plot_value: float, panel_top: float, panel_height: float) -> float: return panel_top + panel_height - (0.0 if max_plot_value <= 0 else (value / max_plot_value) * panel_height) parts = [ f"", f"", f"", ] if show_relative: parts.extend( [ f"", f"", ] ) total_minor_steps = major_tick_count * minor_subdivisions for minor_idx in range(1, total_minor_steps): if minor_idx % minor_subdivisions == 0: continue tick_value = max_value * minor_idx / float(total_minor_steps) y = scaled_y(tick_value, max_value, top, plot_height) parts.append(f"") parts.append( f"{html_escape(format_float(tick_value, 2))}" ) for tick_idx in range(major_tick_count + 1): tick_value = max_value * tick_idx / float(major_tick_count) y = scaled_y(tick_value, max_value, top, plot_height) parts.append(f"") parts.append( f"{html_escape(format_float(tick_value, 2))}" ) if show_relative: for tick_idx in range(relative_tick_count + 1): tick_value = max_relative_value * tick_idx / float(relative_tick_count) y = scaled_y(tick_value, max_relative_value, relative_top, relative_height) parts.append(f"") parts.append( f"{html_escape(format_float(tick_value, 1))}%" ) line_points = [] relative_line_points = [] relative_bar_parts = [] abs_point_parts = [] relative_point_parts = [] label_parts = [] label_stride = max(1, math.ceil(len(ordered) / 22)) for idx, row in enumerate(ordered): value = float(row[metric_key]) y = scaled_y(value, max_value, top, plot_height) x = left + idx * point_pitch + point_pitch / 2.0 is_top = interval_row_key(row) in top_keys line_points.append(f"{x:.1f},{y:.1f}") tooltip_lines = [ f"{row['cls_name']} {row.get('interval_label', row.get('depth_bin_label', 'n/a'))}\n" f"count={row['count']}\n" f"mean={format_float(row.get(metric_key), 3)}{unit_suffix}\n" f"p90={format_float(row.get(metric_key.replace('mean_', 'p90_')), 3)}{unit_suffix}\n" f"max={format_float(row.get(metric_key.replace('mean_', 'max_')), 3)}{unit_suffix}" ] if show_relative and relative_metric_key: tooltip_lines.append( f"\n{relative_reference_label}={format_float(to_float(row.get('relative_reference_value')), 2)}{relative_reference_unit}\n" f"relative_mean={format_float(row.get(relative_metric_key), 2)}%\n" f"relative_p90={format_float(row.get(relative_metric_key.replace('mean_', 'p90_')), 2)}%\n" f"relative_max={format_float(row.get(relative_metric_key.replace('mean_', 'max_')), 2)}%" ) tooltip = "".join(tooltip_lines) point_class = "point-top" if is_top else "point-default" count_class = "count-top" if is_top else "count-default" count_y = max(top + 10.0, y - 6.0) abs_point_parts.append( f"" f"{html_escape(tooltip)}" ) abs_point_parts.append( f"n={html_escape(str(row['count']))}" ) if show_relative and relative_metric_key: relative_value = to_float(row.get(relative_metric_key)) if relative_value is not None: relative_y = scaled_y(float(relative_value), max_relative_value, relative_top, relative_height) relative_bar_class = "relative-bar-top" if is_top else "relative-bar" relative_point_class = "relative-point-top" if is_top else "relative-point-default" relative_bar_parts.append( f"" f"{html_escape(tooltip)}" ) relative_line_points.append(f"{x:.1f},{relative_y:.1f}") relative_point_parts.append( f"" f"{html_escape(tooltip)}" ) if idx % label_stride == 0 or is_top: label_parts.append( f"{html_escape(str(row.get('interval_label', row.get('depth_bin_label', 'n/a'))))}" ) if len(line_points) >= 2: parts.append(f"") parts.extend(abs_point_parts) if show_relative: parts.extend(relative_bar_parts) if len(relative_line_points) >= 2: parts.append(f"") parts.extend(relative_point_parts) parts.extend(label_parts) parts.append( f"{html_escape(x_axis_title)}" ) parts.append( f"Mean error ({html_escape(unit_suffix.strip())})" ) if show_relative: parts.append( f"{html_escape(relative_y_axis_title)}" ) parts.append("") return "".join(parts) def render_interval_trend_sections( grouped_rows: dict[tuple[int, str], list[dict[str, Any]]], metric_key: str, top_keys: set[tuple[int, str, float]], unit_suffix: str, section_title: str, x_axis_title: str = "Depth bin", relative_metric_key: Optional[str] = None, relative_y_axis_title: str = "Relative error (%)", relative_reference_label: str = "gt_bin", relative_reference_unit: str = "m", ) -> str: if not grouped_rows: return ( f"

{html_escape(section_title)}

" f"{render_manual_summary_slot(section_title)}

No interval data.

" ) cards = [] for (cls_id, cls_name), rows in sorted(grouped_rows.items(), key=lambda item: item[0][0]): cards.append( "
" f"

{html_escape(str(cls_name))} id={html_escape(str(cls_id))}

" f"{render_interval_trend_svg(rows, metric_key=metric_key, top_keys=top_keys, unit_suffix=unit_suffix, x_axis_title=x_axis_title, relative_metric_key=relative_metric_key, relative_y_axis_title=relative_y_axis_title, relative_reference_label=relative_reference_label, relative_reference_unit=relative_reference_unit)}" "
" ) return ( f"" ) def render_badcase_cards(title: str, records: list[dict[str, Any]], report_path: Path) -> str: if not records: return ( f"
{html_escape(title)} (0)" "

No saved bad cases in this category.

" ) cards = [] for index, record in enumerate(records, start=1): image_path = record.get("visualization") if not image_path: continue image_href = relative_href(report_path, image_path) caption = f"{record.get('roi', '')} #{index} {record.get('frame_name', '')} {record.get('cls_name', '')}" metrics_rows = [ ("conf / IoU", f"{format_float(to_float(record.get('confidence')), 2)} / {format_float(to_float(record.get('match_iou')), 2)}"), ("yaw err", f"{format_float(to_float(record.get('yaw_abs_deg')), 1)} deg"), ( "direct / edge yaw err", f"{format_float(to_float(record.get('direct_visible_yaw_abs_deg')), 1)} / " f"{format_float(to_float(record.get('edge_visible_yaw_abs_deg')), 1)} deg", ), ("direct-edge gain", f"{format_float(to_float(record.get('direct_minus_edge_visible_yaw_abs_deg')), 1)} deg"), ("|gt_x|", f"{format_float(to_float(record.get('gt_lateral_abs_m')), 2)} m"), ("yaw compare", html_escape(format_bool(record.get("yaw_compare_eligible")))), ("frame", html_escape(str(record.get("frame_name", "")))), ] metric_html = "".join( "{}{}".format(html_escape(label), value if label == "yaw compare" else html_escape(value)) for label, value in metrics_rows ) cards.append( "
" f"" "
" f"

#{index} {html_escape(str(record.get('cls_name', 'unknown')))}" f"{html_escape(str(record.get('frame_name', '')))}

" "" f"{metric_html}" "
" f"Open image" "
" ) if not cards: return ( f"
{html_escape(title)} (0)" "

No saved bad cases in this category.

" ) return ( f"
{html_escape(title)} ({len(cards)})" f"
{''.join(cards)}
" ) def class_badcase_title(metric_name: str, threshold: Optional[float], unit: str, cls_name: str) -> str: if threshold is None or not math.isfinite(float(threshold)): return f"{metric_name}({cls_name})" threshold_text = format_float(float(threshold), 1 if unit == "m" else 0 if float(threshold).is_integer() else 1) return f"{metric_name}(err>{threshold_text}{unit}, {cls_name})" def render_image_section(title: str, image_path: Optional[str], report_path: Path, note: str = "") -> str: if not image_path: return ( f"" ) resolved = Path(image_path) if not resolved.is_file(): return ( f"" ) image_href = relative_href(report_path, resolved) note_html = f"

{html_escape(note)}

" if note else "" return ( f"" ) def render_class_badcase_sections( category_name: str, threshold: Optional[float], unit: str, manifest_entries: list[dict[str, Any]], report_path: Path, ) -> str: if not manifest_entries: return ( f"" ) sections = [] for entry in manifest_entries: title = class_badcase_title(category_name, threshold, unit, str(entry.get("cls_name", "unknown"))) records = load_manifest_records(entry.get("manifest_path", "")) sections.append(render_badcase_cards(title, records, report_path)) return ( f"" ) def render_interval_badcase_sections( section_title: str, metric_name: str, manifest_entries: list[dict[str, Any]], report_path: Path, note: str = "", ) -> str: if not manifest_entries: return ( f"" ) sections = [] for entry in manifest_entries: records = load_manifest_records(entry.get("manifest_path", "")) title = f"{metric_name} {entry.get('bin_label', 'unknown')}" if not records: sections.append( f"
{html_escape(title)} (0)" "

No saved bad cases in this category.

" ) continue cards = [] for index, record in enumerate(records, start=1): image_path = record.get("visualization") if not image_path: continue image_href = relative_href(report_path, image_path) caption = f"{record.get('roi', '')} #{index} {record.get('frame_name', '')} {title}" metrics_rows = [ ("threshold", html_escape(str(record.get("bin_label", "n/a")))), ("objects", html_escape(str(record.get("badcase_count", 0)))), ( "max / mean", html_escape( f"{format_float(to_float(record.get('max_metric_value')), 2)} / {format_float(to_float(record.get('mean_metric_value')), 2)} {record.get('metric_unit', '')}" ), ), ("classes", html_escape(str(record.get("cls_summary", "n/a")))), ("frame", html_escape(str(record.get("frame_name", "")))), ] metric_html = "".join(f"{html_escape(label)}{value}" for label, value in metrics_rows) cards.append( "
" f"" "
" f"

#{index} {html_escape(str(record.get('frame_name', '')))}" f"{html_escape(str(record.get('cls_summary', '')))}

" "" f"{metric_html}" "
" f"Open image" "
" ) sections.append( f"
{html_escape(title)} ({len(cards)})" f"
{''.join(cards)}
" ) note_html = f"

{html_escape(note)}

" if note else "" return ( f"" ) def render_2d_badcase_cards(title: str, records: list[dict[str, Any]], report_path: Path) -> str: if not records: return ( f"
{html_escape(title)} (0)" "

No saved 2D bad cases in this category.

" ) cards = [] for index, record in enumerate(records, start=1): image_path = record.get("visualization") if not image_path: continue image_href = relative_href(report_path, image_path) caption = f"{record.get('roi', '')} #{index} {record.get('frame_name', '')} {record.get('cls_name', '')} {record.get('kind', '')}" metrics_rows = [ ("kind", html_escape(str(record.get("kind", "n/a")))), ("conf", html_escape(format_float(to_float(record.get("confidence")), 2))), ("max_iou_any", html_escape(format_float(to_float(record.get("max_iou_any")), 3))), ("max_iou_same_class", html_escape(format_float(to_float(record.get("max_iou_same_class")), 3))), ("frame", html_escape(str(record.get("frame_name", "")))), ] metric_html = "".join(f"{html_escape(label)}{value}" for label, value in metrics_rows) cards.append( "
" f"" "
" f"

#{index} {html_escape(str(record.get('cls_name', 'unknown')))}" f"{html_escape(str(record.get('frame_name', '')))}

" "" f"{metric_html}" "
" f"Open image" "
" ) return ( f"
{html_escape(title)} ({len(cards)})" f"
{''.join(cards)}
" ) def write_combined_html_report( path: Path, data_yaml: Path, split_path: Path, image_root: Path, combined_payload: dict[str, Any], summary_by_roi: dict[str, Any], portrait_payload: Optional[dict[str, Any]] = None, portrait_pending: bool = False, ) -> None: path.parent.mkdir(parents=True, exist_ok=True) overview_info_rows = [ ("Dataset YAML", html_escape(str(data_yaml))), ("Split File", html_escape(str(split_path))), ("Image Root", html_escape(str(image_root))), ("Entries", html_escape(str(combined_payload.get("num_entries", 0)))), ("Elapsed Minutes", html_escape(format_float(to_float(combined_payload.get("elapsed_minutes")), 2))), ] portrait_button_html = "" portrait_section_html = "" report_title = "Two-ROI Validation Analysis Report" report_subtitle = "Interactive HTML summary for localization errors and signed-lateral-bin yaw comparison." if portrait_payload: portrait_summary = portrait_payload.get("summary") or {} portrait_split_label = str(portrait_payload.get("split", "train")).upper() overview_info_rows.extend( [ ("Portrait Split", html_escape(str(portrait_payload.get("split", "n/a")))), ("Portrait Vehicles", html_escape(str(int(portrait_summary.get("vehicles", 0) or 0)))), ("Portrait Objects", html_escape(str(int(portrait_summary.get("mapped_objects", 0) or 0)))), ("Portrait Summary JSON", html_escape(str(portrait_payload.get("summary_path", "n/a")))), ] ) portrait_button_html = f"" portrait_section_html = render_data_portrait_section(portrait_payload) report_title = "Two-ROI Validation Analysis Report with Dataset Portrait" report_subtitle = "Interactive HTML summary for validation errors plus dataset portrait statistics." elif portrait_pending: overview_info_rows.append(("Portrait Status", html_escape("building"))) report_subtitle = "Interactive HTML summary for validation errors. Dataset portrait is still building and this report will refresh when it is ready." overview_rows = [] for roi_name, payload in summary_by_roi.items(): overall = payload["overall"] diagnostics = payload.get("yaw_compare_diagnostics") or {} focus_face = diagnostics.get("focus_face_row") or {} overview_rows.append( [ html_escape(roi_name.upper()), html_escape(str(overall.get("matched_3d", 0))), html_escape(str(overall.get("matched_pos", 0))), html_escape(str(diagnostics.get("compare_count", 0))), html_escape(format_float(overall.get("yaw_mae_deg"), 2)), html_escape(format_float(overall.get("x_abs_mae_m"), 3)), html_escape(format_float(overall.get("z_abs_mae_m"), 3)), html_escape(format_float(focus_face.get("direct_regression_yaw_mae_deg"), 2)), html_escape(format_float(focus_face.get("edge_based_yaw_mae_deg"), 2)), html_escape(format_float(focus_face.get("mean_direct_minus_edge_yaw_deg"), 2)), html_escape(format_percent(focus_face.get("yaw_compare_edge_better_rate"))), ] ) sections = [] for roi_name, payload in summary_by_roi.items(): overall = payload["overall"] compare_threshold = float(payload.get("yaw_compare_max_lateral_dist_m", DEFAULT_YAW_COMPARE_MAX_LATERAL_DIST_M)) diagnostics = payload.get("yaw_compare_diagnostics") or {} focus_face = diagnostics.get("focus_face_row") or {} threshold_advice_2d = payload.get("threshold_advice_2d") or {} confidence_curve_paths_2d = payload.get("confidence_curve_paths_2d") or {} confusion_matrix_plot_path_2d = payload.get("confusion_matrix_plot_path_2d") focused_confusion_matrix_plot_path_2d = payload.get("focused_confusion_matrix_plot_path_2d") focused_confusion_filter = payload.get("focused_confusion_filter") or {} focused_classification_summary_2d = payload.get("focused_classification_summary_2d") or {} face_selection_summary = payload.get("face_selection_summary") or {} fake_class_summary = payload.get("fake_class_summary") or {} occlusion_binary_summary = payload.get("occlusion_binary_summary") or {} large_vehicle_compare = payload.get("large_vehicle_compare") or {} large_vehicle_compare_summary = large_vehicle_compare.get("summary") or {} large_vehicle_compare_rows = large_vehicle_compare.get("class_rows") or [] large_vehicle_yaw_compare_lateral_rows = large_vehicle_compare.get("yaw_compare_lateral_rows") or [] horizontal_rows = payload.get("horizontal_interval_rows", []) vertical_rows = payload.get("vertical_interval_rows", []) yaw_rows = payload.get("yaw_interval_rows", []) yaw_heading_rows = payload.get("yaw_heading_interval_rows", []) yaw_horizontal_rows = payload.get("yaw_horizontal_interval_rows", []) yaw_compare_lateral_rows = payload.get("yaw_compare_lateral_rows", []) yaw_compare_lateral_rows_by_face_visibility = payload.get("yaw_compare_lateral_rows_by_face_visibility", {}) or {} horizontal_grouped = group_rows_by_class(horizontal_rows) vertical_grouped = group_rows_by_class(vertical_rows) yaw_grouped = group_rows_by_class(yaw_rows) yaw_heading_grouped = group_rows_by_class(yaw_heading_rows) yaw_horizontal_grouped = group_rows_by_class(yaw_horizontal_rows) horizontal_top_keys = compute_interval_top_keys(horizontal_rows, metric_key="mean_x_abs_m", min_count=0) vertical_top_keys = compute_interval_top_keys(vertical_rows, metric_key="mean_z_abs_m", min_count=0) yaw_top_keys = compute_interval_top_keys(yaw_rows, metric_key="mean_yaw_abs_deg", min_count=0) yaw_heading_top_keys = compute_interval_top_keys(yaw_heading_rows, metric_key="mean_yaw_abs_deg", min_count=0) yaw_horizontal_top_keys = compute_interval_top_keys(yaw_horizontal_rows, metric_key="mean_yaw_abs_deg", min_count=0) yaw_bad_threshold_deg = to_float(payload.get("yaw_bad_threshold_deg")) horizontal_bad_threshold_m = to_float(payload.get("horizontal_bad_threshold_m")) vertical_bad_threshold_m = to_float(payload.get("vertical_bad_threshold_m")) error_bin_badcases = int(payload.get("error_bin_badcases") or 0) error_bin_samples_per_bin = int(payload.get("error_bin_samples_per_bin") or 0) metric_rows = [ ("Model", html_escape(payload.get("model_path", "n/a"))), ("Matched 3D", html_escape(str(overall.get("matched_3d", 0)))), ("Matched Pos", html_escape(str(overall.get("matched_pos", 0)))), ("Yaw MAE", html_escape(f"{format_float(overall.get('yaw_mae_deg'), 2)} deg")), ("Horizontal X MAE", html_escape(f"{format_float(overall.get('x_abs_mae_m'), 3)} m")), ("Vertical Z MAE", html_escape(f"{format_float(overall.get('z_abs_mae_m'), 3)} m")), ("2D GT min_wh filter", html_escape(f"{format_float(payload.get('min_wh_px'), 1)} px")), ("Yaw Compare Range", html_escape(f"gt_x in [{-compare_threshold:.1f}, {compare_threshold:.1f}) m")), ( "Yaw Compare Longitudinal Limit", html_escape(f"gt_z < {format_float(payload.get('yaw_compare_max_longitudinal_dist_m', DEFAULT_YAW_COMPARE_MAX_LONGITUDINAL_DIST_M), 1)} m"), ), ("Yaw Compare Binning", html_escape(f"{HORIZONTAL_LATERAL_BIN_M:.1f} m signed lateral bins")), ("Yaw Compare Focus Bucket", html_escape(str(diagnostics.get("focus_bucket", "two-face")))), ("Yaw Compare Pairs", html_escape(str(diagnostics.get("compare_count", 0)))), ("Yaw Error Periodicity", html_escape("pi-periodic orientation error")), ] large_vehicle_metric_rows = [ ("Classes", html_escape(str(large_vehicle_compare.get("class_scope", LARGE_VEHICLE_CLASS_SCOPE_TEXT)))), ("Two-Face Matched 3D", html_escape(str(large_vehicle_compare_summary.get("matched_3d", 0)))), ("Yaw Compare Pairs", html_escape(str(large_vehicle_compare_summary.get("yaw_compare_count", 0)))), ("Direct regression MAE", html_escape(f"{format_float(large_vehicle_compare_summary.get('direct_regression_yaw_mae_deg'), 2)} deg")), ("Edge-based MAE", html_escape(f"{format_float(large_vehicle_compare_summary.get('edge_based_yaw_mae_deg'), 2)} deg")), ( "Mean edge gain", render_edge_gain_deg( None if large_vehicle_compare_summary.get("mean_direct_minus_edge_yaw_deg") is None else -float(large_vehicle_compare_summary.get("mean_direct_minus_edge_yaw_deg")), digits=2, ), ), ("Edge better rate", render_rate_vs_half(large_vehicle_compare_summary.get("yaw_compare_edge_better_rate"))), ("Length compare pairs", html_escape(str(large_vehicle_compare_summary.get("length_compare_count", 0)))), ("Direct length MAE", html_escape(f"{format_float(large_vehicle_compare_summary.get('direct_regression_length_mae_m'), 3)} m")), ("Side-edge length MAE", html_escape(f"{format_float(large_vehicle_compare_summary.get('side_edge_length_mae_m'), 3)} m")), ( "Mean side-edge gain", render_edge_gain_deg( None if large_vehicle_compare_summary.get("mean_direct_minus_side_edge_length_m") is None else -float(large_vehicle_compare_summary.get("mean_direct_minus_side_edge_length_m")), digits=3, ), ), ("Side-edge better rate", render_rate_vs_half(large_vehicle_compare_summary.get("length_compare_edge_better_rate"))), ] compare_rows = [ ["Direct regression MAE", html_escape(f"{format_float(focus_face.get('direct_regression_yaw_mae_deg'), 2)} deg")], ["Direct regression P90", html_escape(f"{format_float(focus_face.get('direct_regression_yaw_p90_deg'), 2)} deg")], ["Edge-based MAE", html_escape(f"{format_float(focus_face.get('edge_based_yaw_mae_deg'), 2)} deg")], ["Edge-based P90", html_escape(f"{format_float(focus_face.get('edge_based_yaw_p90_deg'), 2)} deg")], [ "Mean direct-edge gain", html_escape(f"{format_float(focus_face.get('mean_direct_minus_edge_yaw_deg'), 2)} deg"), ], [ "Median direct-edge gain", html_escape(f"{format_float(focus_face.get('median_direct_minus_edge_yaw_deg'), 2)} deg"), ], ["Edge better rate", html_escape(format_percent(focus_face.get("yaw_compare_edge_better_rate")))], ["Direct better rate", html_escape(format_percent(focus_face.get("yaw_compare_direct_better_rate")))], ["Tie rate", html_escape(format_percent(focus_face.get("yaw_compare_tie_rate")))], ] face_selection_table_rows = [ [ html_escape(str(row.get("selection"))), html_escape(str(row.get("total", 0))), html_escape(str(row.get("correct", 0))), html_escape(format_percent(row.get("accuracy"))), ] for row in label_accuracy_rows(face_selection_summary, label_key="selection") ] face_selection_confusion_rows = [ [html_escape(str(row.get("gt_selection")))] + [html_escape(str(row.get(label, 0))) for label in face_selection_summary.get("label_order", [])] for row in label_confusion_matrix_rows(face_selection_summary, label_key="gt_selection") ] fake_class_table_rows = [ [ html_escape(str(row.get("fake_class"))), html_escape(str(row.get("total", 0))), html_escape(str(row.get("correct", 0))), html_escape(format_percent(row.get("accuracy"))), ] for row in label_accuracy_rows(fake_class_summary, label_key="fake_class") ] occlusion_binary_table_rows = [ [ html_escape(str(row.get("occlusion"))), html_escape(str(row.get("total", 0))), html_escape(str(row.get("correct", 0))), html_escape(format_percent(row.get("accuracy"))), ] for row in label_accuracy_rows(occlusion_binary_summary, label_key="occlusion") ] class_rows = sorted( payload.get("class_rows", []), key=lambda row: (int(row.get("two_face_compare_count") or 0), int(row.get("matched_3d") or 0), float(row.get("yaw_abs_sum") or 0.0)), reverse=True, ) class_2d_highlights = { "cls_acc_2d": top_bad_class_ids(class_rows, "cls_acc_2d", higher_is_worse=False, support_key="cls_eval_pairs_2d"), "precision_2d": top_bad_class_ids(class_rows, "precision_2d", higher_is_worse=False, support_key="pred_total"), "recall_2d": top_bad_class_ids(class_rows, "recall_2d", higher_is_worse=False, support_key="gt_total"), "f1_2d": top_bad_class_ids(class_rows, "f1_2d", higher_is_worse=False, support_key="gt_total"), "map50_2d": top_bad_class_ids(class_rows, "map50_2d", higher_is_worse=False, support_key="gt_total"), "map50_95_2d": top_bad_class_ids(class_rows, "map50_95_2d", higher_is_worse=False, support_key="gt_total"), "false_negatives_2d": top_bad_class_ids(class_rows, "false_negatives_2d", higher_is_worse=True, support_key="gt_total"), "false_positives_2d": top_bad_class_ids(class_rows, "false_positives_2d", higher_is_worse=True, support_key="pred_total"), } class_3d_highlights = { "yaw_mae_deg": top_bad_class_ids(class_rows, "yaw_mae_deg", higher_is_worse=True, support_key="matched_3d"), "x_abs_mae_m": top_bad_class_ids(class_rows, "x_abs_mae_m", higher_is_worse=True, support_key="matched_pos"), "z_abs_mae_m": top_bad_class_ids(class_rows, "z_abs_mae_m", higher_is_worse=True, support_key="matched_pos"), } class_table_rows = [ [ html_escape(str(row.get("cls_id"))), html_escape(str(row.get("cls_name"))), html_escape(str(row.get("matched_3d", 0))), html_escape(str(row.get("two_face_compare_count", 0))), highlight_if_needed( html_escape(format_float(row.get("yaw_mae_deg"), 2)), int(row.get("cls_id", -1)) in class_3d_highlights["yaw_mae_deg"], ), html_escape(format_float(row.get("direct_regression_yaw_mae_deg"), 2)), html_escape(format_float(row.get("edge_based_yaw_mae_deg"), 2)), render_edge_gain_deg( None if row.get("mean_direct_minus_edge_yaw_deg") is None else -float(row.get("mean_direct_minus_edge_yaw_deg")), digits=2, ), render_rate_vs_half(row.get("yaw_compare_edge_better_rate")), highlight_if_needed( html_escape(format_float(row.get("x_abs_mae_m"), 3)), int(row.get("cls_id", -1)) in class_3d_highlights["x_abs_mae_m"], ), highlight_if_needed( html_escape(format_float(row.get("z_abs_mae_m"), 3)), int(row.get("cls_id", -1)) in class_3d_highlights["z_abs_mae_m"], ), ] for row in class_rows ] large_vehicle_class_table_rows = [ [ html_escape(str(row.get("cls_id"))), html_escape(str(row.get("cls_name"))), html_escape(str(row.get("two_face_matched_3d", 0))), html_escape(str(row.get("two_face_compare_count", 0))), html_escape(format_float(row.get("direct_regression_yaw_mae_deg"), 2)), html_escape(format_float(row.get("edge_based_yaw_mae_deg"), 2)), render_edge_gain_deg( None if row.get("mean_direct_minus_edge_yaw_deg") is None else -float(row.get("mean_direct_minus_edge_yaw_deg")), digits=2, ), render_rate_vs_half(row.get("yaw_compare_edge_better_rate")), ] for row in large_vehicle_compare_rows ] class_2d_table_rows = [ [ html_escape(str(row.get("cls_id"))), html_escape(str(row.get("cls_name"))), html_escape(str(row.get("gt_total", 0))), html_escape(str(row.get("pred_total", 0))), html_escape(str(row.get("matched_2d", 0))), html_escape(str(row.get("cls_eval_pairs_2d", 0))), highlight_if_needed( html_escape(format_percent(row.get("cls_acc_2d"))), int(row.get("cls_id", -1)) in class_2d_highlights["cls_acc_2d"], ), highlight_if_needed( html_escape(format_percent(row.get("precision_2d"))), int(row.get("cls_id", -1)) in class_2d_highlights["precision_2d"], ), highlight_if_needed( html_escape(format_percent(row.get("recall_2d"))), int(row.get("cls_id", -1)) in class_2d_highlights["recall_2d"], ), highlight_if_needed( html_escape(format_percent(row.get("f1_2d"))), int(row.get("cls_id", -1)) in class_2d_highlights["f1_2d"], ), highlight_if_needed( html_escape(format_percent(row.get("map50_2d"))), int(row.get("cls_id", -1)) in class_2d_highlights["map50_2d"], ), highlight_if_needed( html_escape(format_percent(row.get("map50_95_2d"))), int(row.get("cls_id", -1)) in class_2d_highlights["map50_95_2d"], ), highlight_if_needed( html_escape(str(int(row.get("false_negatives_2d", 0)))), int(row.get("cls_id", -1)) in class_2d_highlights["false_negatives_2d"], ), highlight_if_needed( html_escape(str(int(row.get("false_positives_2d", 0)))), int(row.get("cls_id", -1)) in class_2d_highlights["false_positives_2d"], ), ] for row in class_rows ] interval_overview_rows = [ [ html_escape(str(item.get("cls_id"))), html_escape(str(item.get("cls_name"))), html_escape(str(item.get("matched_3d", 0))), html_escape(str(item.get("yaw_compare_count", item.get("matched_3d", 0)))), html_escape(str(item.get("horizontal_bins_text", "n/a"))), html_escape(str(item.get("vertical_bins_text", "n/a"))), html_escape(str(item.get("yaw_bins_text", "n/a"))), ] for item in payload.get("per_class_interval_insights", []) ] face_visibility_rows = [ [ html_escape(str(row.get("key"))), html_escape(str(row.get("matched_3d", 0))), html_escape(str(row.get("yaw_compare_count", 0))), html_escape(format_float(row.get("direct_regression_yaw_mae_deg"), 2)), html_escape(format_float(row.get("edge_based_yaw_mae_deg"), 2)), html_escape(format_percent(row.get("yaw_compare_edge_better_rate"))), html_escape(str(row.get("length_compare_count", 0))), html_escape(format_float(row.get("direct_regression_length_mae_m"), 3)), html_escape(format_float(row.get("side_edge_length_mae_m"), 3)), html_escape(format_percent(row.get("length_compare_edge_better_rate"))), ] for row in diagnostics.get("face_visibility_rows", []) ] yaw_compare_lateral_headers = [ "lateral_bin", "yaw_compare_count", "direct_regression_mae_deg", "edge_based_mae_deg", "edge_gain_deg", "edge_better_rate", ] yaw_compare_lateral_table_rows = [ [ html_escape(str(row.get("lateral_bin_label"))), html_escape(str(row.get("yaw_compare_count", 0))), html_escape(format_float(row.get("direct_regression_yaw_mae_deg"), 2)), html_escape(format_float(row.get("edge_based_yaw_mae_deg"), 2)), render_edge_gain_deg( None if row.get("mean_direct_minus_edge_yaw_deg") is None else -float(row.get("mean_direct_minus_edge_yaw_deg")), digits=2, ), render_rate_vs_half(row.get("yaw_compare_edge_better_rate")), ] for row in yaw_compare_lateral_rows ] large_vehicle_yaw_compare_lateral_table_rows = [ [ html_escape(str(row.get("lateral_bin_label"))), html_escape(str(row.get("yaw_compare_count", 0))), html_escape(format_float(row.get("direct_regression_yaw_mae_deg"), 2)), html_escape(format_float(row.get("edge_based_yaw_mae_deg"), 2)), render_edge_gain_deg( None if row.get("mean_direct_minus_edge_yaw_deg") is None else -float(row.get("mean_direct_minus_edge_yaw_deg")), digits=2, ), render_rate_vs_half(row.get("yaw_compare_edge_better_rate")), ] for row in large_vehicle_yaw_compare_lateral_rows ] yaw_compare_lateral_bucket_sections = [ render_html_table( f"Yaw Comparison by Signed Lateral Bin ({bucket_name})", yaw_compare_lateral_headers, [ [ html_escape(str(row.get("lateral_bin_label"))), html_escape(str(row.get("yaw_compare_count", 0))), html_escape(format_float(row.get("direct_regression_yaw_mae_deg"), 2)), html_escape(format_float(row.get("edge_based_yaw_mae_deg"), 2)), render_edge_gain_deg( None if row.get("mean_direct_minus_edge_yaw_deg") is None else -float(row.get("mean_direct_minus_edge_yaw_deg")), digits=2, ), render_rate_vs_half(row.get("yaw_compare_edge_better_rate")), ] for row in yaw_compare_lateral_rows_by_face_visibility.get(bucket_name, []) ], ) for bucket_name in FACE_VISIBILITY_BUCKET_ORDER ] manifests = { category: load_manifest_records(path_str) for category, path_str in (payload.get("badcase_manifest_paths") or {}).items() } error_bin_badcase_entries = payload.get("error_bin_badcase_manifest_entries") or {} class_badcase_manifest_entries = payload.get("class_badcase_manifest_paths") or {} section_html = [ f"
", f"

{html_escape(roi_name.upper())}

", "

" f"Yaw comparison in this section uses paired samples with gt_x binned over " f"[{-compare_threshold:.1f}, {compare_threshold:.1f})m and gt_z < " f"{format_float(payload.get('yaw_compare_max_longitudinal_dist_m', DEFAULT_YAW_COMPARE_MAX_LONGITUDINAL_DIST_M), 1)}m in " f"{HORIZONTAL_LATERAL_BIN_M:.1f}m signed lateral bins. " "Direct-regression and edge-based yaw metrics below are both compared against gt_yaw, with pi-periodic orientation error. " "The signed-lateral-bin tables are shown both for all paired samples and split into front_rear_only, side only, and two-face buckets. " "The Per-Class 3D Metrics table uses the two-face subset for its direct-vs-edge comparison columns. " "Position errors use face-center deltas for face_3d classes and whole-box-center deltas for the remaining classes. " f"Detection-dependent report results below use conf > {format_float(payload.get('report_confidence_2d'), 3)} from the ROI F1-confidence sweep " f"instead of the configured default {format_float(payload.get('configured_confidence_2d'), 3)}. " f"An additional focused 2D confusion matrix below keeps GT objects with |x| < {format_float(focused_confusion_filter.get('max_abs_lateral_m'), 1)}m, " f"|z| < {format_float(focused_confusion_filter.get('max_abs_longitudinal_m'), 1)}m, and difficulty={int(focused_confusion_filter.get('required_difficulty', 0) or 0)}. " f"The large-vehicle comparison section aggregates {html_escape(str(large_vehicle_compare.get('class_scope', LARGE_VEHICLE_CLASS_SCOPE_TEXT)))} on the same two-face comparison subset." "

", render_metric_value_table("ROI Summary", metric_rows), render_metric_value_table( "Face/Cut Selection Summary", [ ("Eval pairs", html_escape(str(int(face_selection_summary.get("total", 0) or 0)))), ("Overall accuracy", html_escape(format_percent(face_selection_summary.get("overall_accuracy")))), ("Mean per-selection accuracy", html_escape(format_percent(face_selection_summary.get("mean_accuracy")))), ("CSV", html_escape(str(payload.get("face_selection_csv", "n/a")))), ("Confusion CSV", html_escape(str(payload.get("face_selection_confusion_csv", "n/a")))), ], ), render_html_table( "Face/Cut Selection Accuracy", ["selection", "total", "correct", "accuracy"], face_selection_table_rows, ), render_html_table( "Face/Cut Selection Confusion Matrix", ["gt \\ pred", *[html_escape(str(label)) for label in face_selection_summary.get("label_order", [])]], face_selection_confusion_rows, ), render_metric_value_table( "Fake Class Classification Summary", [ ("Eval pairs", html_escape(str(int(fake_class_summary.get("total", 0) or 0)))), ("Overall accuracy", html_escape(format_percent(fake_class_summary.get("overall_accuracy")))), ("Mean class accuracy", html_escape(format_percent(fake_class_summary.get("mean_accuracy")))), ("Fake vs non-fake bad cases", html_escape(str(int(fake_class_summary.get("fake_vs_non_fake", 0) or 0)))), ("Fake internal bad cases", html_escape(str(int(fake_class_summary.get("fake_internal_confusion", 0) or 0)))), ("Visible internal bad cases", html_escape(str(int(fake_class_summary.get("visible_internal_confusion", 0) or 0)))), ("CSV", html_escape(str(payload.get("fake_class_csv", "n/a")))), ], ), render_html_table( "Fake Class Classification Accuracy", ["fake_class", "total", "correct", "accuracy"], fake_class_table_rows, ), render_metric_value_table( "Occlusion Binary Classification Summary", [ ("Eval pairs", html_escape(str(int(occlusion_binary_summary.get("total", 0) or 0)))), ("Overall accuracy", html_escape(format_percent(occlusion_binary_summary.get("overall_accuracy")))), ("Mean class accuracy", html_escape(format_percent(occlusion_binary_summary.get("mean_accuracy")))), ("CSV", html_escape(str(payload.get("occlusion_binary_csv", "n/a")))), ], ), render_html_table( "Occlusion Binary Classification Accuracy", ["occlusion", "total", "correct", "accuracy"], occlusion_binary_table_rows, ), render_metric_value_table( "2D Threshold Advice", [ ("Configured 2D conf", html_escape(format_float(payload.get("configured_confidence_2d"), 3))), ("Recommended 2D conf", html_escape(format_float(threshold_advice_2d.get("recommended_confidence"), 3))), ("Advice source", html_escape(str(threshold_advice_2d.get("source", "n/a")))), ("Mean class precision", html_escape(format_percent(threshold_advice_2d.get("mean_precision")))), ("Mean class recall", html_escape(format_percent(threshold_advice_2d.get("mean_recall")))), ("Mean class F1", html_escape(format_percent(threshold_advice_2d.get("mean_f1")))), ], ), render_metric_value_table( "2D Summary", [ ("GT total", html_escape(str(overall.get("gt_total", 0)))), ("Pred total", html_escape(str(overall.get("pred_total", 0)))), ("Matched 2D", html_escape(str(overall.get("matched_2d", 0)))), ("2D confidence threshold", html_escape(format_float(threshold_advice_2d.get("recommended_confidence"), 3))), ("2D cls eval pairs", html_escape(str(int(overall.get("cls_eval_pairs_2d", 0) or 0)))), ("2D cls acc", html_escape(format_percent(overall.get("cls_acc_2d")))), ("2D precision", html_escape(format_percent(overall.get("precision_2d")))), ("2D recall", html_escape(format_percent(overall.get("recall_2d")))), ("2D F1", html_escape(format_percent(overall.get("f1_2d")))), ("2D mAP50", html_escape(format_percent(overall.get("map50_2d")))), ("2D mAP50-95", html_escape(format_percent(overall.get("map50_95_2d")))), ("Focused GT count", html_escape(str(int(focused_confusion_filter.get("gt_count", 0) or 0)))), ("Focused 2D cls eval pairs", html_escape(str(int(focused_classification_summary_2d.get("overall_pairs", 0) or 0)))), ("Focused 2D cls acc", html_escape(format_percent(focused_classification_summary_2d.get("overall_accuracy")))), ("2D false negatives", html_escape(str(max(0, int(overall.get("gt_total", 0)) - int(overall.get("matched_2d", 0)))))), ("2D false positives", html_escape(str(max(0, int(overall.get("pred_total", 0)) - int(overall.get("matched_2d", 0)))))), ], ), render_html_table( "Per-Class 2D Metrics", [ "cls_id", "cls_name", "gt_total", "pred_total", "matched_2d", "cls_eval_pairs_2d", "cls_acc_2d", "precision_2d", "recall_2d", "f1_2d", "mAP50", "mAP50-95", "false_negatives", "false_positives", ], class_2d_table_rows, ), render_image_section( "2D F1-Confidence Curve", confidence_curve_paths_2d.get("f1_curve"), path, note=( f"Report threshold is {format_float(payload.get('report_confidence_2d'), 3)}, " f"where mean class F1 reaches {format_percent(threshold_advice_2d.get('mean_f1'))}." ), ), render_image_section( "2D Confusion Matrix", confusion_matrix_plot_path_2d, path, note=( f"Stats in this matrix use conf > {format_float(payload.get('report_confidence_2d'), 3)}. " "Class-agnostic IoU>=0.5 matching, normalized by true class. Background row/column indicates false positives and false negatives." ), ), render_image_section( "2D Confusion Matrix (Easy Near-Range, No Occlusion)", focused_confusion_matrix_plot_path_2d, path, note=( f"Focused subset keeps GT objects with |lateral|<{format_float(focused_confusion_filter.get('max_abs_lateral_m'), 1)}m, " f"|longitudinal|<{format_float(focused_confusion_filter.get('max_abs_longitudinal_m'), 1)}m, and difficulty=" f"{int(focused_confusion_filter.get('required_difficulty', 0) or 0)}. " "Unmatched FP counts only include predictions whose predicted centers fall inside the same spatial window." ), ), "", render_html_table("Focused Two-Face Yaw Comparison", ["Metric", "Value"], compare_rows), render_metric_value_table("Large-Vehicle Two-Face Comparison", large_vehicle_metric_rows), render_html_table( "Large-Vehicle Per-Class Comparison", [ "cls_id", "cls_name", "two_face_matched_3d", "two_face_compare_count", "direct_regression_mae_deg", "edge_based_mae_deg", "edge_gain_deg", "edge_better_rate", ], large_vehicle_class_table_rows, ), render_html_table( "Large-Vehicle Yaw Comparison by Signed Lateral Bin", yaw_compare_lateral_headers, large_vehicle_yaw_compare_lateral_table_rows, ), render_html_table( "Yaw Comparison by Signed Lateral Bin (All Paired Samples)", yaw_compare_lateral_headers, yaw_compare_lateral_table_rows, ), *yaw_compare_lateral_bucket_sections, render_html_table( "Face-Visibility Bucket Diagnostics", [ "bucket", "matched_3d", "yaw_compare_count", "direct_regression_mae_deg", "edge_based_mae_deg", "edge_better_rate", "length_compare_count", "direct_length_mae_m", "side_edge_length_mae_m", "side_edge_better_rate", ], face_visibility_rows, ), render_html_table( "Per-Class 3D Metrics", [ "cls_id", "cls_name", "matched_3d", "two_face_compare_count", "yaw_mae_deg", "direct_regression_mae_deg", "edge_based_mae_deg", "edge_gain_deg", "edge_better_rate", "x_mae_m", "z_mae_m", ], class_table_rows, ), render_html_table( "Depth Interval Overview", ["cls_id", "cls_name", "matched_3d", "yaw_compare_count", "worst horizontal bins", "worst vertical bins", "worst yaw bins"], interval_overview_rows, ), "", render_interval_trend_sections( horizontal_grouped, metric_key="mean_x_abs_m", top_keys=horizontal_top_keys, unit_suffix="m", section_title="Horizontal X Error by 5m Lateral Bin", x_axis_title="Lateral bin", relative_metric_key="mean_x_abs_pct", ), render_interval_trend_sections( vertical_grouped, metric_key="mean_z_abs_m", top_keys=vertical_top_keys, unit_suffix="m", section_title="Vertical Z Error by 5m Depth Bin", x_axis_title="Depth bin", relative_metric_key="mean_z_abs_pct", ), render_interval_trend_sections( yaw_grouped, metric_key="mean_yaw_abs_deg", top_keys=yaw_top_keys, unit_suffix="deg", section_title="Yaw Orientation Error by 10m Depth Bin", x_axis_title="Depth bin", ), render_interval_trend_sections( yaw_heading_grouped, metric_key="mean_yaw_abs_deg", top_keys=yaw_heading_top_keys, unit_suffix="deg", section_title="Yaw Orientation Error by 10 degree Bin", x_axis_title="GT yaw bin", relative_metric_key="mean_yaw_abs_pct", relative_y_axis_title="Relative yaw error / |gt_yaw| (%)", relative_reference_label="mean |gt_yaw|", relative_reference_unit="deg", ), render_interval_trend_sections( yaw_horizontal_grouped, metric_key="mean_yaw_abs_deg", top_keys=yaw_horizontal_top_keys, unit_suffix="deg", section_title="Yaw Orientation Error by 5m Horizontal Bin", x_axis_title="Horizontal bin", ), render_interval_badcase_sections( "Yaw By Error", "Yaw Error", error_bin_badcase_entries.get("yaw", []), path, note=( f"Each bin keeps up to {error_bin_badcases} object candidates, then exports up to {error_bin_samples_per_bin} frame-level samples. " f"Each image overlays all objects in that frame whose yaw error is at least the bin lower bound ({format_float(ERROR_YAW_BIN_DEG, 1)}deg bins)." ), ), render_interval_badcase_sections( "Horizontal By Error", "Horizontal Error", error_bin_badcase_entries.get("horizontal", []), path, note=( f"Each bin keeps up to {error_bin_badcases} object candidates, then exports up to {error_bin_samples_per_bin} frame-level samples. " f"Each image overlays all objects in that frame whose horizontal error is at least the bin lower bound ({format_float(ERROR_DISTANCE_BIN_M, 1)}m bins)." ), ), render_interval_badcase_sections( "Vertical By Error", "Vertical Error", error_bin_badcase_entries.get("vertical", []), path, note=( f"Each bin keeps up to {error_bin_badcases} object candidates, then exports up to {error_bin_samples_per_bin} frame-level samples. " f"Each image overlays all objects in that frame whose vertical error is at least the bin lower bound ({format_float(ERROR_DISTANCE_BIN_M, 1)}m bins)." ), ), render_class_badcase_sections("Yaw", yaw_bad_threshold_deg, "deg", class_badcase_manifest_entries.get("yaw", []), path), render_class_badcase_sections( "Horizontal", horizontal_bad_threshold_m, "m", class_badcase_manifest_entries.get("horizontal", []), path ), render_class_badcase_sections("Vertical", vertical_bad_threshold_m, "m", class_badcase_manifest_entries.get("vertical", []), path), "
", ] sections.append("".join(section_html)) html_text = f""" Two-ROI Validation Analysis Report

{html_escape(report_title)}

{html_escape(report_subtitle)}

{portrait_button_html} {''.join(f"" for roi_name in summary_by_roi)}
{render_metric_value_table("Run Info", overview_info_rows)} {render_html_table( "ROI Overview", ["ROI", "matched_3d", "matched_pos", "yaw_compare_count", "yaw_mae_deg", "x_mae_m", "z_mae_m", "direct_regression_mae_deg", "edge_based_mae_deg", "direct-edge_gain_deg", "edge_better_rate"], overview_rows, )}
{portrait_section_html} {''.join(sections)}
""" path.write_text(html_text, encoding="utf-8") def run_roi_analysis( bundle, entries: list[tuple[str, str]], image_root: Path, class_map: dict[str, int], difficulty_weights: list[float], face_3d_classes: set[int], complete_3d_classes: set[int], classes: Optional[set[int]], min_wh_px: float, face_visibility_score_thresh: float, yaw_bad_threshold_deg: float, edge_yaw_max_lateral_dist_m: float, yaw_compare_max_lateral_dist_m: float, horizontal_bad_threshold_m: float, vertical_bad_threshold_m: float, topk_badcases: int, per_class_badcases: int, error_bin_badcases: int, error_bin_samples_per_bin: int, badcase_random_seed: int, output_root: Path, log_every: int, split_file: str, inference_batch_size: int, ) -> dict[str, Any]: roi_name = bundle.spec.name.lower() roi_output = output_root / roi_name roi_output.mkdir(parents=True, exist_ok=True) edge_yaw_compare_score_thresh = float(EDGE_YAW_VALID_VISIBILITY_SCORE_THRESH) focused_confusion_max_abs_longitudinal_m = focused_confusion_max_abs_longitudinal_m_for_roi(roi_name) configured_confidence_2d = float(bundle.spec.conf) threshold_search_conf = min(float(configured_confidence_2d), float(MIN_CONFIDENCE_FOR_2D_THRESHOLD_SEARCH)) bad_yaw_handle, bad_yaw_writer = make_writer(roi_output / "bad_cases_yaw.csv") bad_x_handle, bad_x_writer = make_writer(roi_output / "bad_cases_horizontal.csv") bad_y_handle, bad_y_writer = make_writer(roi_output / "bad_cases_vertical.csv") bad_face_selection_handle, bad_face_selection_writer = make_writer( roi_output / "bad_cases_face_selection.csv", fieldnames=FACE_SELECTION_BADCASE_FIELDS, ) names_dict = names_to_dict(bundle.names) ap_summary_2d, threshold_advice_2d, report_confidence_2d = collect_2d_confidence_advice( bundle=bundle, entries=entries, image_root=image_root, class_map=class_map, difficulty_weights=difficulty_weights, face_3d_classes=face_3d_classes, complete_3d_classes=complete_3d_classes, classes=classes, min_wh_px=min_wh_px, names_dict=names_dict, roi_name=roi_name, roi_output=roi_output, topk_badcases=topk_badcases, badcase_random_seed=badcase_random_seed, focused_confusion_max_abs_longitudinal_m=focused_confusion_max_abs_longitudinal_m, threshold_search_conf=threshold_search_conf, configured_confidence_2d=configured_confidence_2d, inference_batch_size=inference_batch_size, log_every=log_every, ) overall_bucket = make_metric_bucket() class_buckets = defaultdict(make_metric_bucket) detailed_class_buckets = defaultdict(make_metric_bucket) class_face_visibility_buckets = defaultdict(lambda: defaultdict(make_metric_bucket)) detailed_class_face_visibility_buckets = defaultdict(lambda: defaultdict(make_metric_bucket)) two_d_eval_packets: list[dict[str, Any]] = [] face_selection_store = make_label_accuracy_store(FACE_SELECTION_LABEL_ORDER) occlusion_binary_store = make_label_accuracy_store(OCCLUSION_BINARY_LABEL_ORDER) breakdowns = make_breakdown_buckets() horizontal_interval_store = make_interval_store() vertical_interval_store = make_interval_store() yaw_interval_store = make_interval_store() yaw_heading_interval_store = make_interval_store() yaw_horizontal_interval_store = make_interval_store() yaw_compare_lateral_store = defaultdict(make_metric_bucket) large_vehicle_compare_bucket = make_metric_bucket() large_vehicle_yaw_compare_lateral_store = defaultdict(make_metric_bucket) yaw_compare_lateral_store_by_face_visibility = { bucket_name: defaultdict(make_metric_bucket) for bucket_name in FACE_VISIBILITY_BUCKET_ORDER } rng = random.Random(int(badcase_random_seed) + (0 if roi_name == "roi0" else 1000)) bad_yaw_store = make_reservoir_store() bad_x_store = make_reservoir_store() bad_y_store = make_reservoir_store() bad_face_selection_store = make_reservoir_store() yaw_visual_stores = make_interval_visual_reservoirs() x_visual_stores = make_interval_visual_reservoirs() z_visual_stores = make_interval_visual_reservoirs() class_badcase_stores = { "yaw": defaultdict(make_reservoir_store), "horizontal": defaultdict(make_reservoir_store), "vertical": defaultdict(make_reservoir_store), } error_frame_records = { "yaw": defaultdict(list), "horizontal": defaultdict(list), "vertical": defaultdict(list), } error_bin_per_bucket_capacity = max(1, int(error_bin_badcases)) if error_bin_badcases > 0 else 0 total_samples = len(entries) start_time = time.time() for batch_item, raw_outputs in iter_roi_analysis_samples( bundle=bundle, entries=entries, image_root=image_root, class_map=class_map, difficulty_weights=difficulty_weights, face_3d_classes=face_3d_classes, complete_3d_classes=complete_3d_classes, classes=classes, min_wh_px=min_wh_px, inference_batch_size=inference_batch_size, ): sample_index = int(batch_item["sample_index"]) image_path = batch_item["image_path"] label_path = batch_item["label_path"] prepared = batch_item["prepared"] gt = batch_item["gt"] analysis_outputs = filter_prediction_outputs( raw_outputs=raw_outputs, conf_thres=float(report_confidence_2d), max_det=bundle.spec.max_det, classes=classes, ) analysis_detections = analysis_outputs.detections analysis_preds_3d = analysis_outputs.preds_3d analysis_anchors = analysis_outputs.anchors analysis_strides = analysis_outputs.strides predictions = decode_prepared_roi_predictions( bundle=bundle, prepared=prepared, filtered_outputs=analysis_outputs, edge_yaw_max_lateral_dist_m=edge_yaw_max_lateral_dist_m, ) gt_class_names = list(gt.get("class_names") or []) for gt_count_index, cls_id in enumerate(gt["classes"].tolist()): detailed_count_name = ( str(gt_class_names[int(gt_count_index)]) if int(gt_count_index) < len(gt_class_names) else get_cls_name(bundle.names, int(cls_id)) ) add_gt_count(overall_bucket) add_gt_count(class_buckets[int(cls_id)]) add_gt_count(detailed_class_buckets[(int(cls_id), detailed_count_name)]) for prediction in predictions: add_prediction_count(overall_bucket) add_prediction_count(class_buckets[int(prediction["cls_id"])]) img_h, img_w = prepared.image.shape[:2] gt_labels_3d = gt["lb_3d"] pred_boxes = np.asarray([prediction["bbox_xyxy"] for prediction in predictions], dtype=np.float32) if predictions else np.zeros((0, 4), dtype=np.float32) pred_cls = np.asarray([prediction["cls_id"] for prediction in predictions], dtype=np.int32) if predictions else np.zeros((0,), dtype=np.int32) pred_conf = np.asarray([prediction["confidence"] for prediction in predictions], dtype=np.float32) if predictions else np.zeros((0,), dtype=np.float32) analysis_pred_boxes = ( np.asarray(analysis_detections[:, :4], dtype=np.float32) if len(analysis_detections) else np.zeros((0, 4), dtype=np.float32) ) analysis_pred_cls = ( np.asarray(analysis_detections[:, 5], dtype=np.int32).reshape(-1) if len(analysis_detections) else np.zeros((0,), dtype=np.int32) ) analysis_pred_conf = ( np.asarray(analysis_detections[:, 4], dtype=np.float32).reshape(-1) if len(analysis_detections) else np.zeros((0,), dtype=np.float32) ) gt_focus_mask = np.zeros((len(gt["classes"]),), dtype=bool) gt_difficulties = np.asarray(gt["lb_2d"].get("difficulties", []), dtype=np.float32).reshape(-1) if gt_labels_3d is not None and len(gt_labels_3d): for gt_index, cls_id in enumerate(gt["classes"].tolist()): if gt_index >= len(gt_labels_3d): break gt_attrs_focus = extract_3d_attrs_from_gt( gt_labels_3d[gt_index], int(cls_id), prepared.calib, img_w, img_h, face_3d_classes, complete_3d_classes, score_thr=face_visibility_score_thresh, ) if gt_attrs_focus is None: continue center = np.asarray(gt_attrs_focus.get("center"), dtype=np.float32).reshape(-1) lateral_m = to_float(center[0]) if center.size > 0 else None longitudinal_m = to_float(center[2]) if center.size > 2 else None difficulty = gt_difficulties[gt_index] if gt_index < len(gt_difficulties) else None gt_focus_mask[gt_index] = is_focused_confusion_gt_object( lateral_m=lateral_m, longitudinal_m=longitudinal_m, difficulty=difficulty, max_abs_longitudinal_m=focused_confusion_max_abs_longitudinal_m, ) analysis_pred_focus_mask = build_prediction_focus_mask( analysis_detections, analysis_preds_3d, analysis_anchors, analysis_strides, prepared.calib, max_abs_longitudinal_m=focused_confusion_max_abs_longitudinal_m, ) two_d_eval_packets.append( { "sample_index": int(sample_index), "image_path": str(image_path), "label_path": str(label_path), "gt_cls": gt["classes"].copy(), "gt_boxes": gt["boxes_xyxy"].copy(), "gt_focus_mask": gt_focus_mask.copy(), "pred_cls": analysis_pred_cls.copy(), "pred_boxes": analysis_pred_boxes.copy(), "pred_conf": analysis_pred_conf.copy(), "pred_focus_mask": analysis_pred_focus_mask.copy(), } ) matches, iou_matrix = greedy_match_indices(gt["classes"], gt["boxes_xyxy"], pred_cls, pred_boxes, iou_thr=0.5) face_selection_matches, face_selection_iou_matrix = greedy_match_indices_any_class(gt["boxes_xyxy"], pred_boxes, iou_thr=0.5) occlusion_matches, _ = greedy_match_indices_any_class(gt["boxes_xyxy"], pred_boxes, iou_thr=0.5) for gt_occ_index, pred_occ_index in occlusion_matches: gt_occ_cls_id = int(gt["classes"][gt_occ_index]) pred_occ_cls_id = int(pred_cls[pred_occ_index]) add_label_accuracy_sample( occlusion_binary_store, occlusion_binary_label(get_cls_name(bundle.names, gt_occ_cls_id)), occlusion_binary_label(get_cls_name(bundle.names, pred_occ_cls_id)), ) for gt_face_index, pred_face_index in face_selection_matches: if gt_labels_3d is None or gt_face_index >= len(gt_labels_3d): continue gt_face_cls_id = int(gt["classes"][gt_face_index]) pred_face_cls_id = int(pred_cls[pred_face_index]) if gt_face_cls_id not in face_3d_classes: continue gt_face_decoded = decode_3d_target( gt_labels_3d[gt_face_index], gt_face_cls_id, prepared.calib, img_w, img_h, face_3d_classes, complete_3d_classes, score_thr=face_visibility_score_thresh, bbox_xyxy=gt["boxes_xyxy"][gt_face_index], ) pred_face_decoded = predictions[pred_face_index].get("decoded") if gt_face_decoded is None or pred_face_decoded is None: continue gt_selection_label = gt_face_selection_label(gt_labels_3d[gt_face_index], gt_face_decoded) pred_selection_label = pred_face_selection_label(predictions[pred_face_index]["pred_41"], pred_face_decoded) add_label_accuracy_sample(face_selection_store, gt_selection_label, pred_selection_label) if gt_selection_label is not None and pred_selection_label is not None and gt_selection_label != pred_selection_label: gt_face_cls_name = ( str(gt_class_names[int(gt_face_index)]) if int(gt_face_index) < len(gt_class_names) else get_cls_name(bundle.names, gt_face_cls_id) ) face_selection_record = make_face_selection_badcase_record( sample_index=sample_index, roi_name=roi_name, image_path=image_path, label_path=label_path, gt_index=int(gt_face_index), pred_index=int(pred_face_index), gt_cls_id=gt_face_cls_id, gt_cls_name=gt_face_cls_name, pred_cls_id=pred_face_cls_id, pred_cls_name=get_cls_name(bundle.names, pred_face_cls_id), gt_face_selection_label=gt_selection_label, pred_face_selection_label=pred_selection_label, gt_bbox_xyxy=gt["boxes_xyxy"][gt_face_index], pred_bbox_xyxy=pred_boxes[pred_face_index], match_iou=float(face_selection_iou_matrix[int(gt_face_index), int(pred_face_index)]) if face_selection_iou_matrix.size else 0.0, confidence=float(pred_conf[pred_face_index]), ) bad_face_selection_writer.writerow(face_selection_record_to_csv_row(face_selection_record)) reservoir_add(bad_face_selection_store, face_selection_record, topk_badcases, rng) for gt_index, pred_index in matches: cls_id = int(gt["classes"][gt_index]) detailed_cls_name = ( str(gt_class_names[int(gt_index)]) if int(gt_index) < len(gt_class_names) else get_cls_name(bundle.names, cls_id) ) detailed_cls_key = (cls_id, detailed_cls_name) cls_bucket = class_buckets[cls_id] detailed_cls_bucket = detailed_class_buckets[detailed_cls_key] add_2d_match(overall_bucket) add_2d_match(cls_bucket) if gt_labels_3d is None or gt_index >= len(gt_labels_3d): continue gt_visible_faces = ( select_gt_visible_faces(gt_labels_3d[gt_index], score_thr=face_visibility_score_thresh) if cls_id in face_3d_classes else [] ) gt_decoded = decode_3d_target( gt_labels_3d[gt_index], cls_id, prepared.calib, img_w, img_h, face_3d_classes, complete_3d_classes, score_thr=face_visibility_score_thresh, bbox_xyxy=gt["boxes_xyxy"][gt_index], ) pred_decoded = predictions[pred_index].get("decoded") pred_box_attrs = extract_3d_attrs_from_prediction( predictions[pred_index]["pred_41"], predictions[pred_index]["anchor_xy"], float(predictions[pred_index]["stride"]), prepared.calib, face_type=None, pred_edge_60=predictions[pred_index].get("pred_edge_60"), ) pred_position_attrs = predictions[pred_index]["attrs"] if gt_decoded is None or pred_box_attrs is None or pred_position_attrs is None or pred_decoded is None: continue gt_attrs = extract_3d_attrs_from_gt( gt_labels_3d[gt_index], cls_id, prepared.calib, img_w, img_h, face_3d_classes, complete_3d_classes, score_thr=face_visibility_score_thresh, ) if gt_attrs is None: continue gt_position_attrs = gt_attrs position_error_basis = "box_center" if cls_id in face_3d_classes: eval_face_type = resolve_face_center_eval_face_type(gt_visible_faces, gt_decoded, pred_decoded) if eval_face_type is not None: gt_face_position_attrs = extract_3d_attrs_from_gt( gt_labels_3d[gt_index], cls_id, prepared.calib, img_w, img_h, face_3d_classes, complete_3d_classes, face_type=int(eval_face_type), score_thr=face_visibility_score_thresh, ) pred_face_position_attrs = extract_3d_attrs_from_prediction( predictions[pred_index]["pred_41"], predictions[pred_index]["anchor_xy"], float(predictions[pred_index]["stride"]), prepared.calib, face_type=int(eval_face_type), pred_edge_60=predictions[pred_index].get("pred_edge_60"), ) if gt_face_position_attrs is not None and pred_face_position_attrs is not None: gt_position_attrs = gt_face_position_attrs pred_position_attrs = pred_face_position_attrs position_error_basis = "face_center" gt_edge_yaw_faces = ( select_gt_visible_faces(gt_labels_3d[gt_index], score_thr=edge_yaw_compare_score_thresh) if cls_id in face_3d_classes else [] ) fallback_face_type = max(gt_edge_yaw_faces, key=lambda item: float(item[1][6]))[0] if gt_edge_yaw_faces else None gt_visible_yaw = None pred_edge_visible_yaw = None pred_edge_bucket_visible_yaw = None edge_selection = predictions[pred_index].get("edge_selection") if fallback_face_type is not None: gt_visible_yaw = decode_multi_visible_face_yaw_from_gt( gt_labels_3d[gt_index], cls_id, prepared.calib, img_w, img_h, face_3d_classes, complete_3d_classes, fallback_face_type=fallback_face_type, score_thr=edge_yaw_compare_score_thresh, bbox_xyxy=gt["boxes_xyxy"][gt_index], ) edge_selection_yaw = to_float(predictions[pred_index].get("edge_yaw")) pred_edge_bucket_visible_yaw = edge_selection_yaw pred_edge_visible_yaw = edge_selection_yaw if bool(predictions[pred_index].get("edge_confident")) else None cut_object = bool(is_gt_cut_object(gt_labels_3d[gt_index])) if cls_id in face_3d_classes else False position_eligible = bool(cls_id in complete_3d_classes or not cut_object) cls_name = detailed_cls_name record = make_record( sample_index=sample_index, roi_name=roi_name, image_path=image_path, label_path=label_path, cls_name=cls_name, gt_index=int(gt_index), pred_index=int(pred_index), gt_box=gt["boxes_xyxy"][gt_index], pred_box=predictions[pred_index]["bbox_xyxy"], match_iou=float(iou_matrix[gt_index, pred_index]), prediction=predictions[pred_index], gt_attrs={**gt_attrs, "corners_3d": gt_decoded["corners_3d"]}, pred_attrs={**pred_box_attrs, "corners_3d": pred_decoded["corners_3d"]}, gt_position_attrs=None if gt_position_attrs is None else {**gt_position_attrs, "corners_3d": gt_decoded["corners_3d"]}, pred_position_attrs=None if pred_position_attrs is None else {**pred_position_attrs, "corners_3d": pred_decoded["corners_3d"]}, gt_decoded=gt_decoded, pred_decoded=pred_decoded, gt_visible_faces=gt_visible_faces, gt_visible_yaw=to_float(gt_visible_yaw), pred_edge_visible_yaw=to_float(pred_edge_visible_yaw), pred_edge_bucket_visible_yaw=to_float(pred_edge_bucket_visible_yaw), pred_edge_decoded=predictions[pred_index].get("edge_heading_decoded"), pred_edge_box=predictions[pred_index].get("edge_box"), is_cut_object_flag=cut_object, position_eligible=position_eligible, position_error_basis=position_error_basis, yaw_compare_max_lateral_dist_m=yaw_compare_max_lateral_dist_m, pred_yaw_compare_face_types=(() if edge_selection is None else edge_selection.get("face_types", ())), pred_yaw_compare_valid=bool(predictions[pred_index].get("edge_confident")), ) add_3d_record(overall_bucket, record, yaw_bad_threshold_deg, horizontal_bad_threshold_m, vertical_bad_threshold_m) add_3d_record(cls_bucket, record, yaw_bad_threshold_deg, horizontal_bad_threshold_m, vertical_bad_threshold_m) add_3d_record(detailed_cls_bucket, record, yaw_bad_threshold_deg, horizontal_bad_threshold_m, vertical_bad_threshold_m) add_lateral_interval_sample( horizontal_interval_store, cls_id=cls_id, cls_name=cls_name, lateral_m=record.get("position_gt_x_m"), value=record["x_abs_m"] if record["position_eligible"] else None, bin_width_m=HORIZONTAL_LATERAL_BIN_M, lateral_limit_m=HORIZONTAL_LATERAL_RANGE_M, ) add_interval_sample( vertical_interval_store, cls_id=cls_id, cls_name=cls_name, depth_m=record.get("position_gt_z_m"), value=record["z_abs_m"] if record["position_eligible"] else None, bin_width_m=VERTICAL_DEPTH_BIN_M, ) add_interval_sample( yaw_interval_store, cls_id=cls_id, cls_name=cls_name, depth_m=record["gt_depth_m"], value=record["yaw_abs_deg"], bin_width_m=YAW_DEPTH_BIN_M, ) add_heading_interval_sample( yaw_heading_interval_store, cls_id=cls_id, cls_name=cls_name, heading_deg=record["gt_yaw_deg"], value=record["yaw_abs_deg"], bin_width_deg=YAW_HEADING_BIN_DEG, relative_reference_value=( None if record["gt_yaw_deg"] is None else wrap_heading_deg(float(record["gt_yaw_deg"])) ), relative_reference_abs=True, ) add_lateral_interval_sample( yaw_horizontal_interval_store, cls_id=cls_id, cls_name=cls_name, lateral_m=record["gt_x_m"], value=record["yaw_abs_deg"], bin_width_m=HORIZONTAL_LATERAL_BIN_M, lateral_limit_m=HORIZONTAL_LATERAL_RANGE_M, ) yaw_compare_lateral_start = lateral_interval_start( record["gt_x_m"], HORIZONTAL_LATERAL_BIN_M, yaw_compare_max_lateral_dist_m, ) face_visibility_bucket = ( record.get("yaw_compare_face_bucket") if cls_id in face_3d_classes else None ) large_vehicle_focus_match = bool(cls_id in LARGE_VEHICLE_CLASS_IDS and face_visibility_bucket == "two-face") if large_vehicle_focus_match: add_3d_record( large_vehicle_compare_bucket, record, yaw_bad_threshold_deg, horizontal_bad_threshold_m, vertical_bad_threshold_m, edge_visible_key="edge_bucket_visible_yaw_abs_deg", direct_minus_edge_key="direct_minus_edge_bucket_visible_yaw_abs_deg", ) if yaw_compare_lateral_start is not None and is_signed_lateral_yaw_compare_longitudinal_eligible(record): add_3d_record( yaw_compare_lateral_store[float(yaw_compare_lateral_start)], record, yaw_bad_threshold_deg, horizontal_bad_threshold_m, vertical_bad_threshold_m, ) if face_visibility_bucket in yaw_compare_lateral_store_by_face_visibility: add_3d_record( yaw_compare_lateral_store_by_face_visibility[str(face_visibility_bucket)][float(yaw_compare_lateral_start)], record, yaw_bad_threshold_deg, horizontal_bad_threshold_m, vertical_bad_threshold_m, edge_visible_key="edge_bucket_visible_yaw_abs_deg", direct_minus_edge_key="direct_minus_edge_bucket_visible_yaw_abs_deg", ) if large_vehicle_focus_match: add_3d_record( large_vehicle_yaw_compare_lateral_store[float(yaw_compare_lateral_start)], record, yaw_bad_threshold_deg, horizontal_bad_threshold_m, vertical_bad_threshold_m, edge_visible_key="edge_bucket_visible_yaw_abs_deg", direct_minus_edge_key="direct_minus_edge_bucket_visible_yaw_abs_deg", ) breakdown_keys = { "cut_status": "cut" if cut_object else ("non_cut" if cls_id in face_3d_classes else "n/a"), "distance_bin": record["distance_bin"], "bbox_diag_bin": record["bbox_diag_bin"], "class_group": "face_3d" if cls_id in face_3d_classes else ("complete_3d" if cls_id in complete_3d_classes else "2d_only"), } if face_visibility_bucket is not None: breakdown_keys["face_visibility"] = face_visibility_bucket add_3d_record( class_face_visibility_buckets[cls_id][face_visibility_bucket], record, yaw_bad_threshold_deg, horizontal_bad_threshold_m, vertical_bad_threshold_m, edge_visible_key="edge_bucket_visible_yaw_abs_deg", direct_minus_edge_key="direct_minus_edge_bucket_visible_yaw_abs_deg", ) add_3d_record( detailed_class_face_visibility_buckets[detailed_cls_key][face_visibility_bucket], record, yaw_bad_threshold_deg, horizontal_bad_threshold_m, vertical_bad_threshold_m, edge_visible_key="edge_bucket_visible_yaw_abs_deg", direct_minus_edge_key="direct_minus_edge_bucket_visible_yaw_abs_deg", ) for breakdown_name, breakdown_key in breakdown_keys.items(): add_3d_record( breakdowns[breakdown_name][breakdown_key], record, yaw_bad_threshold_deg, horizontal_bad_threshold_m, vertical_bad_threshold_m, edge_visible_key=( "edge_bucket_visible_yaw_abs_deg" if breakdown_name == "face_visibility" else "edge_visible_yaw_abs_deg" ), direct_minus_edge_key=( "direct_minus_edge_bucket_visible_yaw_abs_deg" if breakdown_name == "face_visibility" else "direct_minus_edge_visible_yaw_abs_deg" ), ) if record["yaw_abs_deg"] is not None and record["yaw_abs_deg"] >= yaw_bad_threshold_deg: bad_yaw_writer.writerow(record_to_csv_row(record)) reservoir_add(bad_yaw_store, record, topk_badcases, rng) reservoir_add(class_badcase_stores["yaw"][(cls_id, cls_name)], record, per_class_badcases, rng) error_frame_records["yaw"][str(record["image_path"])].append(dict(record)) add_interval_visual_record( yaw_visual_stores, record["yaw_abs_deg"], record, bin_width=ERROR_YAW_BIN_DEG, per_bin_capacity=error_bin_per_bucket_capacity, rng=rng, ) if record["position_eligible"] and record["x_abs_m"] is not None and record["x_abs_m"] > horizontal_bad_threshold_m: bad_x_writer.writerow(record_to_csv_row(record)) reservoir_add(bad_x_store, record, topk_badcases, rng) reservoir_add(class_badcase_stores["horizontal"][(cls_id, cls_name)], record, per_class_badcases, rng) error_frame_records["horizontal"][str(record["image_path"])].append(dict(record)) add_interval_visual_record( x_visual_stores, record["x_abs_m"], record, bin_width=ERROR_DISTANCE_BIN_M, per_bin_capacity=error_bin_per_bucket_capacity, rng=rng, ) if record["position_eligible"] and record["z_abs_m"] is not None and record["z_abs_m"] > vertical_bad_threshold_m: bad_y_writer.writerow(record_to_csv_row(record)) reservoir_add(bad_y_store, record, topk_badcases, rng) reservoir_add(class_badcase_stores["vertical"][(cls_id, cls_name)], record, per_class_badcases, rng) error_frame_records["vertical"][str(record["image_path"])].append(dict(record)) add_interval_visual_record( z_visual_stores, record["z_abs_m"], record, bin_width=ERROR_DISTANCE_BIN_M, per_bin_capacity=error_bin_per_bucket_capacity, rng=rng, ) if (sample_index + 1) % max(1, log_every) == 0 or (sample_index + 1) == total_samples: elapsed = time.time() - start_time per_sample = elapsed / max(sample_index + 1, 1) remaining = total_samples - (sample_index + 1) eta = remaining * per_sample print( f"[{roi_name}] {sample_index + 1}/{total_samples} " f"elapsed={elapsed / 60:.1f}m eta={eta / 60:.1f}m matched_3d={overall_bucket['matched_3d']}" ) bad_yaw_handle.close() bad_x_handle.close() bad_y_handle.close() bad_face_selection_handle.close() thresholded_2d = build_thresholded_2d_artifacts( eval_packets=two_d_eval_packets, roi_name=roi_name, names_dict=names_dict, confidence_threshold=float(report_confidence_2d), topk_badcases=int(topk_badcases), badcase_random_seed=int(badcase_random_seed), roi_output=roi_output, bundle=bundle, image_root=image_root, max_abs_longitudinal_m=focused_confusion_max_abs_longitudinal_m, ) classification_summary_2d = thresholded_2d["classification_summary_2d"] focused_classification_summary_2d = thresholded_2d["focused_classification_summary_2d"] face_selection_summary = summarize_label_accuracy_store(face_selection_store) occlusion_binary_summary = summarize_label_accuracy_store(occlusion_binary_store) overall_summary = summarize_metric_bucket(overall_bucket) overall_summary.update(thresholded_2d["overall"]) overall_summary["map50_2d"] = ap_summary_2d.get("map50") overall_summary["map50_95_2d"] = ap_summary_2d.get("map50_95") overall_summary["precision_ap_2d"] = threshold_advice_2d.get("mean_precision") overall_summary["recall_ap_2d"] = threshold_advice_2d.get("mean_recall") overall_summary["f1_ap_2d"] = threshold_advice_2d.get("mean_f1") overall_summary["recommended_confidence_2d"] = float(report_confidence_2d) overall_summary["report_confidence_2d"] = float(report_confidence_2d) overall_summary["cls_acc_2d"] = classification_summary_2d.get("overall_accuracy") overall_summary["cls_eval_pairs_2d"] = int(classification_summary_2d.get("overall_pairs") or 0) total_yaw_sum = overall_summary["yaw_abs_sum"] or 0.0 class_rows = [] detailed_class_keys = sorted(detailed_class_buckets.keys(), key=lambda item: (int(item[0]), str(item[1]))) class_ids_with_2d_only = sorted(set((thresholded_2d.get("per_class") or {}).keys()) - {int(key[0]) for key in detailed_class_keys}) class_keys = [*detailed_class_keys, *[(int(cls_id), get_cls_name(bundle.names, int(cls_id))) for cls_id in class_ids_with_2d_only]] for cls_id, detailed_cls_name in class_keys: bucket = detailed_class_buckets.get((int(cls_id), str(detailed_cls_name)), make_metric_bucket()) row = summarize_metric_bucket(bucket) two_face_row = summarize_metric_bucket( (detailed_class_face_visibility_buckets.get((int(cls_id), str(detailed_cls_name)), {}) or {}).get("two-face", make_metric_bucket()) ) row["cls_id"] = int(cls_id) row["cls_name"] = str(detailed_cls_name) row["mapped_cls_name"] = get_cls_name(bundle.names, int(cls_id)) row["yaw_contribution_rate"] = (row["yaw_abs_sum"] / total_yaw_sum) if total_yaw_sum > 0 else None thresholded_2d_row = (thresholded_2d.get("per_class") or {}).get(int(cls_id), {}) row["gt_total"] = int(thresholded_2d_row.get("gt_total", row.get("gt_total", 0)) or 0) row["pred_total"] = int(thresholded_2d_row.get("pred_total", row.get("pred_total", 0)) or 0) row["matched_2d"] = int(thresholded_2d_row.get("matched_2d", row.get("matched_2d", 0)) or 0) row["precision_2d"] = thresholded_2d_row.get("precision_2d") row["recall_2d"] = thresholded_2d_row.get("recall_2d") row["f1_2d"] = thresholded_2d_row.get("f1_2d") ap_row = ap_summary_2d["per_class"].get(int(cls_id), {}) row["map50_2d"] = ap_row.get("map50") row["map50_95_2d"] = ap_row.get("map50_95") row["precision_ap_2d"] = ap_row.get("precision_ap") row["recall_ap_2d"] = ap_row.get("recall_ap") row["f1_ap_2d"] = ap_row.get("f1_at_recommended_confidence") row["optimal_confidence_2d"] = ap_row.get("optimal_confidence") row["optimal_f1_2d"] = ap_row.get("optimal_f1") row["false_negatives_2d"] = int( thresholded_2d_row.get("false_negatives_2d", max(0, int(row.get("gt_total", 0)) - int(row.get("matched_2d", 0)))) ) row["false_positives_2d"] = int( thresholded_2d_row.get("false_positives_2d", max(0, int(row.get("pred_total", 0)) - int(row.get("matched_2d", 0)))) ) cls_classification = classification_summary_2d["per_class"].get(int(cls_id), {}) row["cls_eval_pairs_2d"] = int(cls_classification.get("cls_eval_pairs_2d") or 0) row["cls_correct_2d"] = int(cls_classification.get("cls_correct_2d") or 0) row["cls_acc_2d"] = cls_classification.get("cls_acc_2d") row["overall_yaw_compare_count"] = int(row.get("yaw_compare_count") or 0) row["two_face_matched_3d"] = int(two_face_row.get("matched_3d") or 0) row["two_face_compare_count"] = int(two_face_row.get("yaw_compare_count") or 0) row["direct_regression_yaw_mae_deg"] = two_face_row.get("direct_regression_yaw_mae_deg") row["direct_regression_yaw_p90_deg"] = two_face_row.get("direct_regression_yaw_p90_deg") row["edge_based_yaw_mae_deg"] = two_face_row.get("edge_based_yaw_mae_deg") row["edge_based_yaw_p90_deg"] = two_face_row.get("edge_based_yaw_p90_deg") row["mean_direct_minus_edge_yaw_deg"] = two_face_row.get("mean_direct_minus_edge_yaw_deg") row["median_direct_minus_edge_yaw_deg"] = two_face_row.get("median_direct_minus_edge_yaw_deg") row["yaw_compare_edge_better_count"] = int(two_face_row.get("yaw_compare_edge_better_count") or 0) row["yaw_compare_direct_better_count"] = int(two_face_row.get("yaw_compare_direct_better_count") or 0) row["yaw_compare_tie_count"] = int(two_face_row.get("yaw_compare_tie_count") or 0) row["yaw_compare_edge_better_rate"] = two_face_row.get("yaw_compare_edge_better_rate") row["yaw_compare_direct_better_rate"] = two_face_row.get("yaw_compare_direct_better_rate") row["yaw_compare_tie_rate"] = two_face_row.get("yaw_compare_tie_rate") class_rows.append(row) class_rows_sorted = sorted( class_rows, key=lambda row: ((row.get("yaw_abs_sum") or 0.0), row.get("matched_3d") or 0), reverse=True, ) write_class_csv(roi_output / "class_metrics.csv", class_rows) horizontal_interval_rows = build_interval_rows( horizontal_interval_store, bin_width_m=HORIZONTAL_LATERAL_BIN_M, value_name="x_abs_m", threshold=horizontal_bad_threshold_m, relative_value_name="x_abs_pct", relative_reference_abs=True, ) vertical_interval_rows = build_interval_rows( vertical_interval_store, bin_width_m=VERTICAL_DEPTH_BIN_M, value_name="z_abs_m", threshold=vertical_bad_threshold_m, relative_value_name="z_abs_pct", ) if roi_name == "roi1": # ROI1 near-range longitudinal z error is not meaningful in the report. vertical_interval_rows = [ row for row in vertical_interval_rows if float(row.get("depth_bin_start_m", 0.0)) >= ROI1_MIN_Z_ERROR_DEPTH_M ] yaw_interval_rows = build_interval_rows( yaw_interval_store, bin_width_m=YAW_DEPTH_BIN_M, value_name="yaw_abs_deg", threshold=yaw_bad_threshold_deg, ) yaw_heading_interval_rows = build_interval_rows( yaw_heading_interval_store, bin_width_m=YAW_HEADING_BIN_DEG, value_name="yaw_abs_deg", threshold=yaw_bad_threshold_deg, relative_value_name="yaw_abs_pct", interval_prefix="yaw_bin", interval_unit="deg", relative_reference_label="mean |gt_yaw|", relative_reference_unit="deg", relative_reference_field="relative_gt_yaw_deg", ) yaw_horizontal_interval_rows = build_interval_rows( yaw_horizontal_interval_store, bin_width_m=HORIZONTAL_LATERAL_BIN_M, value_name="yaw_abs_deg", threshold=yaw_bad_threshold_deg, ) yaw_compare_lateral_rows = build_metric_bucket_interval_rows( yaw_compare_lateral_store, bin_width_m=HORIZONTAL_LATERAL_BIN_M, prefix="lateral_bin", ) yaw_compare_lateral_rows_by_face_visibility = build_grouped_metric_bucket_interval_rows( yaw_compare_lateral_store_by_face_visibility, bin_width_m=HORIZONTAL_LATERAL_BIN_M, prefix="lateral_bin", group_order=FACE_VISIBILITY_BUCKET_ORDER, ) large_vehicle_yaw_compare_lateral_rows = build_metric_bucket_interval_rows( large_vehicle_yaw_compare_lateral_store, bin_width_m=HORIZONTAL_LATERAL_BIN_M, prefix="lateral_bin", ) yaw_compare_lateral_rows_by_face_visibility_flat = flatten_grouped_metric_bucket_interval_rows( yaw_compare_lateral_rows_by_face_visibility, group_key="face_visibility_bucket", ) horizontal_interval_csv = roi_output / "class_horizontal_lateral_5m.csv" vertical_interval_csv = roi_output / "class_vertical_depth_5m.csv" yaw_interval_csv = roi_output / "class_yaw_depth_10m.csv" yaw_heading_interval_csv = roi_output / "class_yaw_heading_10deg.csv" yaw_horizontal_interval_csv = roi_output / "class_yaw_horizontal_5m.csv" yaw_compare_lateral_csv = roi_output / "yaw_compare_signed_lateral_5m.csv" yaw_compare_lateral_by_face_visibility_csv = roi_output / "yaw_compare_signed_lateral_5m_by_face_visibility.csv" face_selection_csv = roi_output / "face_selection_accuracy.csv" face_selection_confusion_csv = roi_output / "face_selection_confusion_matrix.csv" fake_class_csv = roi_output / "fake_class_accuracy.csv" occlusion_binary_csv = roi_output / "occlusion_binary_accuracy.csv" write_rows_csv( horizontal_interval_csv, horizontal_interval_rows, fallback_fields=["cls_id", "cls_name", "depth_bin_start_m", "depth_bin_end_m", "depth_bin_label", "count"], ) write_rows_csv( vertical_interval_csv, vertical_interval_rows, fallback_fields=["cls_id", "cls_name", "depth_bin_start_m", "depth_bin_end_m", "depth_bin_label", "count"], ) write_rows_csv( yaw_interval_csv, yaw_interval_rows, fallback_fields=["cls_id", "cls_name", "depth_bin_start_m", "depth_bin_end_m", "depth_bin_label", "count"], ) write_rows_csv( yaw_heading_interval_csv, yaw_heading_interval_rows, fallback_fields=["cls_id", "cls_name", "yaw_bin_start_deg", "yaw_bin_end_deg", "yaw_bin_label", "count"], ) write_rows_csv( yaw_horizontal_interval_csv, yaw_horizontal_interval_rows, fallback_fields=["cls_id", "cls_name", "depth_bin_start_m", "depth_bin_end_m", "depth_bin_label", "count"], ) write_rows_csv( yaw_compare_lateral_csv, yaw_compare_lateral_rows, fallback_fields=["lateral_bin_start_m", "lateral_bin_end_m", "lateral_bin_label", "yaw_compare_count"], ) write_rows_csv( yaw_compare_lateral_by_face_visibility_csv, yaw_compare_lateral_rows_by_face_visibility_flat, fallback_fields=["face_visibility_bucket", "lateral_bin_start_m", "lateral_bin_end_m", "lateral_bin_label", "yaw_compare_count"], ) write_rows_csv( face_selection_csv, label_accuracy_rows(face_selection_summary, label_key="selection"), fallback_fields=["selection", "total", "correct", "accuracy"], ) write_rows_csv( face_selection_confusion_csv, label_confusion_matrix_rows(face_selection_summary, label_key="gt_selection"), fallback_fields=["gt_selection", *list(face_selection_summary.get("label_order", FACE_SELECTION_LABEL_ORDER))], ) write_rows_csv( fake_class_csv, label_accuracy_rows(thresholded_2d.get("fake_class_summary", {}), label_key="fake_class"), fallback_fields=["fake_class", "total", "correct", "accuracy"], ) write_rows_csv( occlusion_binary_csv, label_accuracy_rows(occlusion_binary_summary, label_key="occlusion"), fallback_fields=["occlusion", "total", "correct", "accuracy"], ) breakdown_summaries = { name: {key: summarize_metric_bucket(bucket) for key, bucket in group.items()} for name, group in breakdowns.items() } write_json(roi_output / "breakdowns.json", breakdown_summaries) yaw_compare_diagnostics = build_yaw_compare_diagnostics(overall_summary, breakdown_summaries) large_vehicle_compare_summary = summarize_metric_bucket(large_vehicle_compare_bucket) large_vehicle_class_rows = sorted( [row for row in class_rows if int(row.get("cls_id", -1)) in LARGE_VEHICLE_CLASS_IDS], key=lambda row: ( int(row.get("two_face_compare_count") or 0), int(row.get("matched_3d") or 0), float(row.get("yaw_abs_sum") or 0.0), ), reverse=True, ) bad_yaw_records = reservoir_records(bad_yaw_store, rng) bad_x_records = reservoir_records(bad_x_store, rng) bad_y_records = reservoir_records(bad_y_store, rng) bad_face_selection_records = reservoir_records(bad_face_selection_store, rng) yaw_error_visual_records = { float(start): reservoir_records(store, rng) for start, store in sorted(yaw_visual_stores.items(), key=lambda item: item[0], reverse=True) if store.get("records") } horizontal_error_visual_records = { float(start): reservoir_records(store, rng) for start, store in sorted(x_visual_stores.items(), key=lambda item: item[0], reverse=True) if store.get("records") } vertical_error_visual_records = { float(start): reservoir_records(store, rng) for start, store in sorted(z_visual_stores.items(), key=lambda item: item[0], reverse=True) if store.get("records") } error_bin_badcase_manifests = { "yaw": save_interval_badcase_visuals( yaw_error_visual_records, error_frame_records["yaw"], roi_output / "visuals" / "yaw_bins", "yaw", ERROR_YAW_BIN_DEG, "deg", max(1, int(error_bin_samples_per_bin)), bundle, image_root, ), "horizontal": save_interval_badcase_visuals( horizontal_error_visual_records, error_frame_records["horizontal"], roi_output / "visuals" / "horizontal_bins", "horizontal", ERROR_DISTANCE_BIN_M, "m", max(1, int(error_bin_samples_per_bin)), bundle, image_root, ), "vertical": save_interval_badcase_visuals( vertical_error_visual_records, error_frame_records["vertical"], roi_output / "visuals" / "vertical_bins", "vertical", ERROR_DISTANCE_BIN_M, "m", max(1, int(error_bin_samples_per_bin)), bundle, image_root, ), } class_badcase_manifests = { category: save_class_badcase_visuals( { class_key: reservoir_records(store, rng) for class_key, store in stores.items() }, roi_output / "visuals" / f"{category}_by_class", category, bundle, image_root, ) for category, stores in class_badcase_stores.items() } face_selection_manifest = save_face_selection_badcase_visuals( bad_face_selection_records, roi_output / "visuals" / "face_selection", "face_selection", bundle, image_root, ) worst_depth_bins = sort_nullable_desc(breakdown_summaries["distance_bin"].items(), "yaw_mae_deg")[:5] worst_bbox_bins = sort_nullable_desc(breakdown_summaries["bbox_diag_bin"].items(), "yaw_mae_deg")[:5] worst_face_buckets = sort_nullable_desc(breakdown_summaries["face_visibility"].items(), "yaw_mae_deg")[:5] top_horizontal_classes = sorted(class_rows, key=lambda row: row.get("x_bad_count") or 0, reverse=True)[:10] top_vertical_classes = sorted(class_rows, key=lambda row: row.get("z_bad_count") or 0, reverse=True)[:10] top_yaw_contributors = class_rows_sorted[:10] horizontal_by_class = group_rows_by_class(horizontal_interval_rows) vertical_by_class = group_rows_by_class(vertical_interval_rows) yaw_by_class = group_rows_by_class(yaw_interval_rows) per_class_interval_insights = [] for row in class_rows_sorted: if int(row.get("matched_3d") or 0) <= 0: continue cls_key = (int(row["cls_id"]), str(row["cls_name"])) horizontal_bins = top_interval_rows(horizontal_by_class.get(cls_key, []), "mean_x_abs_m", topn=3, min_count=30) vertical_bins = top_interval_rows(vertical_by_class.get(cls_key, []), "mean_z_abs_m", topn=3, min_count=30) yaw_bins = top_interval_rows(yaw_by_class.get(cls_key, []), "mean_yaw_abs_deg", topn=3, min_count=30) per_class_insight = { "cls_id": int(row["cls_id"]), "cls_name": str(row["cls_name"]), "matched_3d": int(row.get("matched_3d") or 0), "matched_pos": int(row.get("matched_pos") or 0), "yaw_mae_deg": row.get("yaw_mae_deg"), "x_abs_mae_m": row.get("x_abs_mae_m"), "y_abs_mae_m": row.get("y_abs_mae_m"), "z_abs_mae_m": row.get("z_abs_mae_m"), "yaw_bad_rate": row.get("yaw_bad_rate"), "x_bad_rate": row.get("x_bad_rate"), "y_bad_rate": row.get("y_bad_rate"), "z_bad_rate": row.get("z_bad_rate"), "horizontal_bins": horizontal_bins, "vertical_bins": vertical_bins, "yaw_bins": yaw_bins, "horizontal_bins_text": ";".join( f"{bin_row['depth_bin_label']}: mean={format_float(bin_row.get('mean_x_abs_m'))}m, p90={format_float(bin_row.get('p90_x_abs_m'))}m, n={bin_row['count']}" for bin_row in horizontal_bins ) or "n/a", "vertical_bins_text": ";".join( f"{bin_row['depth_bin_label']}: mean={format_float(bin_row.get('mean_z_abs_m'))}m, p90={format_float(bin_row.get('p90_z_abs_m'))}m, n={bin_row['count']}" for bin_row in vertical_bins ) or "n/a", "yaw_bins_text": ";".join( f"{bin_row['depth_bin_label']}: mean={format_float(bin_row.get('mean_yaw_abs_deg'), 2)}deg, p90={format_float(bin_row.get('p90_yaw_abs_deg'), 2)}deg, n={bin_row['count']}" for bin_row in yaw_bins ) or "n/a", } per_class_interval_insights.append(per_class_insight) summary_payload = { "roi": roi_name, "model_path": bundle.spec.model_path, "yaw_compare_max_lateral_dist_m": float(yaw_compare_max_lateral_dist_m), "yaw_compare_max_longitudinal_dist_m": float(DEFAULT_YAW_COMPARE_MAX_LONGITUDINAL_DIST_M), "min_wh_px": float(min_wh_px), "configured_confidence_2d": float(configured_confidence_2d), "report_confidence_2d": float(report_confidence_2d), "horizontal_bad_threshold_m": float(horizontal_bad_threshold_m), "vertical_bad_threshold_m": float(vertical_bad_threshold_m), "yaw_bad_threshold_deg": float(yaw_bad_threshold_deg), "badcase_random_seed": int(badcase_random_seed), "per_class_badcases": int(per_class_badcases), "error_bin_badcases": int(error_bin_badcases), "error_bin_samples_per_bin": int(error_bin_samples_per_bin), "overall": overall_summary, "ap_summary_2d": ap_summary_2d, "threshold_advice_2d": threshold_advice_2d, "confidence_curve_paths_2d": ap_summary_2d.get("curve_paths", {}), "classification_summary_2d": classification_summary_2d, "face_selection_summary": face_selection_summary, "face_selection_csv": str(face_selection_csv), "face_selection_confusion_csv": str(face_selection_confusion_csv), "fake_class_summary": thresholded_2d.get("fake_class_summary", {}), "fake_class_csv": str(fake_class_csv), "occlusion_binary_summary": occlusion_binary_summary, "occlusion_binary_csv": str(occlusion_binary_csv), "confusion_matrix_plot_path_2d": thresholded_2d.get("confusion_matrix_plot_path_2d"), "focused_classification_summary_2d": focused_classification_summary_2d, "focused_confusion_matrix_plot_path_2d": thresholded_2d.get("focused_confusion_matrix_plot_path_2d"), "focused_confusion_filter": thresholded_2d.get("focused_confusion_filter", {}), "class_rows": class_rows, "horizontal_interval_rows": horizontal_interval_rows, "vertical_interval_rows": vertical_interval_rows, "yaw_interval_rows": yaw_interval_rows, "yaw_heading_interval_rows": yaw_heading_interval_rows, "yaw_horizontal_interval_rows": yaw_horizontal_interval_rows, "yaw_compare_lateral_rows": yaw_compare_lateral_rows, "yaw_compare_lateral_rows_by_face_visibility": yaw_compare_lateral_rows_by_face_visibility, "large_vehicle_compare": { "class_ids": sorted(int(cls_id) for cls_id in LARGE_VEHICLE_CLASS_IDS), "class_scope": LARGE_VEHICLE_CLASS_SCOPE_TEXT, "summary": large_vehicle_compare_summary, "class_rows": large_vehicle_class_rows, "yaw_compare_lateral_rows": large_vehicle_yaw_compare_lateral_rows, }, "horizontal_interval_csv": str(horizontal_interval_csv), "vertical_interval_csv": str(vertical_interval_csv), "yaw_interval_csv": str(yaw_interval_csv), "yaw_heading_interval_csv": str(yaw_heading_interval_csv), "yaw_horizontal_interval_csv": str(yaw_horizontal_interval_csv), "yaw_compare_lateral_csv": str(yaw_compare_lateral_csv), "yaw_compare_lateral_by_face_visibility_csv": str(yaw_compare_lateral_by_face_visibility_csv), "per_class_interval_insights": per_class_interval_insights, "yaw_compare_diagnostics": yaw_compare_diagnostics, "top_yaw_contributors": top_yaw_contributors, "top_horizontal_classes": top_horizontal_classes, "top_vertical_classes": top_vertical_classes, "worst_depth_bins": worst_depth_bins, "worst_bbox_bins": worst_bbox_bins, "worst_face_buckets": worst_face_buckets, "badcase_counts": { "yaw_saved_topk": len(bad_yaw_records), "horizontal_saved_topk": len(bad_x_records), "vertical_saved_topk": len(bad_y_records), "face_selection_saved_topk": len(bad_face_selection_records), **(thresholded_2d.get("badcase_counts") or {}), }, "badcase_manifest_paths": { "face_selection": str(roi_output / "visuals" / "face_selection" / "manifest.json"), **(thresholded_2d.get("badcase_manifest_paths") or {}), }, "error_bin_badcase_manifest_entries": error_bin_badcase_manifests, "class_badcase_manifest_paths": class_badcase_manifests, "badcase_manifest_sizes": { **(thresholded_2d.get("badcase_manifest_sizes") or {}), "yaw_by_error": sum(int(entry.get("count", 0)) for entry in error_bin_badcase_manifests["yaw"]), "horizontal_by_error": sum(int(entry.get("count", 0)) for entry in error_bin_badcase_manifests["horizontal"]), "vertical_by_error": sum(int(entry.get("count", 0)) for entry in error_bin_badcase_manifests["vertical"]), "face_selection": len(face_selection_manifest), }, } write_json(roi_output / "summary.json", summary_payload) write_markdown_summary( roi_output / "summary.md", roi_name=roi_name, payload=summary_payload, model_path=bundle.spec.model_path, split_file=split_file, ) write_markdown_summary_zh( roi_output / "summary_zh.md", roi_name=roi_name, payload=summary_payload, model_path=bundle.spec.model_path, split_file=split_file, horizontal_csv=horizontal_interval_csv, vertical_csv=vertical_interval_csv, yaw_depth_csv=yaw_interval_csv, yaw_heading_csv=yaw_heading_interval_csv, ) return summary_payload def write_incremental_combined_outputs( output_root: Path, data_yaml: Path, split_path: Path, image_root: Path, summary_by_roi: dict[str, Any], num_entries: int, args: argparse.Namespace, elapsed_minutes: float, portrait_payload: Optional[dict[str, Any]] = None, ) -> None: portrait_pending = bool(not args.skip_data_portrait and portrait_payload is None) combined_payload = { "data": str(data_yaml), "inference_config": str(args.inference_config), "split": args.split, "split_path": str(split_path), "image_root": str(image_root), "num_entries": int(num_entries), "elapsed_minutes": float(elapsed_minutes), "edge_yaw_max_lateral_dist_m": float(args.edge_yaw_max_lateral_dist), "yaw_compare_max_lateral_dist_m": float(args.yaw_compare_max_lateral_dist), "yaw_compare_max_longitudinal_dist_m": float(DEFAULT_YAW_COMPARE_MAX_LONGITUDINAL_DIST_M), "rois": summary_by_roi, } if portrait_payload is not None: portrait_summary = portrait_payload.get("summary") or {} combined_payload["data_portrait"] = { "split": portrait_payload.get("split", args.data_portrait_split), "summary_path": portrait_payload.get("summary_path"), "num_entries": int(portrait_summary.get("num_entries", 0) or 0), "vehicles": int(portrait_summary.get("vehicles", 0) or 0), "mapped_objects": int(portrait_summary.get("mapped_objects", 0) or 0), } write_json(output_root / "summary.json", combined_payload) lines = [ "# Two-ROI Validation Error Analysis", "", f"- data: `{data_yaml}`", f"- inference_config: `{args.inference_config}`", f"- split: `{args.split}`", f"- split_path: `{split_path}`", f"- image_root: `{image_root}`", f"- num_entries: {int(num_entries)}", f"- elapsed_minutes: {elapsed_minutes:.2f}", "", ] if portrait_payload is not None: portrait_summary = portrait_payload.get("summary") or {} lines.extend( [ "## Data Portrait", "", f"- status: ready", f"- split: `{portrait_payload.get('split', args.data_portrait_split)}`", f"- entries: {int(portrait_summary.get('num_entries', 0) or 0)}", f"- vehicles: {int(portrait_summary.get('vehicles', 0) or 0)}", f"- mapped_objects: {int(portrait_summary.get('mapped_objects', 0) or 0)}", f"- summary_json: `{portrait_payload.get('summary_path', 'n/a')}`", "", ] ) elif portrait_pending: lines.extend( [ "## Data Portrait", "", f"- status: building", f"- split: `{args.data_portrait_split}`", "", ] ) for roi_name, payload in summary_by_roi.items(): overall = payload["overall"] threshold_advice_2d = payload.get("threshold_advice_2d") or {} lines.extend( [ f"## {roi_name.upper()}", "", f"- matched_2d: {overall['matched_2d']}", f"- matched_3d: {overall['matched_3d']}", f"- recommended_2d_confidence: {format_float(to_float(threshold_advice_2d.get('recommended_confidence')), 3)}", f"- precision_recall_f1_2d: {format_percent(overall.get('precision_2d'))} / {format_percent(overall.get('recall_2d'))} / {format_percent(overall.get('f1_2d'))}", f"- yaw_mae_deg: {overall['yaw_mae_deg']}", f"- x_abs_mae_m: {overall['x_abs_mae_m']}", f"- z_abs_mae_m: {overall['z_abs_mae_m']}", f"- yaw_compare_count(gt_x in [-{args.yaw_compare_max_lateral_dist},{args.yaw_compare_max_lateral_dist})m, 5m bins): {overall['yaw_compare_count']}", f"- direct_regression_yaw_mae_deg: {overall['direct_regression_yaw_mae_deg']}", f"- edge_based_yaw_mae_deg: {overall['edge_based_yaw_mae_deg']}", f"- summary: `{output_root / roi_name / 'summary.md'}`", "", ] ) (output_root / "summary.md").write_text("\n".join(lines), encoding="utf-8") write_combined_summary_zh(output_root / "summary_zh.md", data_yaml=data_yaml, split_path=split_path, image_root=image_root, summary_by_roi=summary_by_roi) write_combined_html_report( output_root / "report.html", data_yaml=data_yaml, split_path=split_path, image_root=image_root, combined_payload=combined_payload, summary_by_roi=summary_by_roi, portrait_payload=portrait_payload, portrait_pending=portrait_pending, ) def write_run_manifest( output_root: Path, data_yaml: Path, split_path: Path, image_root: Path, num_entries: int, requested_rois: set[str], model_paths: dict[str, str], args: argparse.Namespace, portrait_payload: Optional[dict[str, Any]] = None, portrait_split_path: Optional[Path] = None, portrait_entries: Optional[list[tuple[str, str]]] = None, ) -> None: portrait_summary = portrait_payload.get("summary") or {} if portrait_payload is not None else {} portrait_num_entries = int(portrait_summary.get("num_entries", len(portrait_entries or [])) or 0) write_json( output_root / "run_manifest.json", { "data": str(data_yaml), "inference_config": str(args.inference_config), "split": args.split, "split_path": str(split_path), "image_root": str(image_root), "num_entries": int(num_entries), "analyze_rois": sorted(requested_rois), "models": model_paths, "edge_yaw_max_lateral_dist_m": float(args.edge_yaw_max_lateral_dist), "yaw_compare_max_lateral_dist_m": float(args.yaw_compare_max_lateral_dist), "yaw_compare_max_longitudinal_dist_m": float(DEFAULT_YAW_COMPARE_MAX_LONGITUDINAL_DIST_M), "data_portrait": ( None if portrait_payload is None else { "split": str(portrait_payload.get("split", args.data_portrait_split)), "split_path": str(portrait_split_path) if portrait_split_path is not None else None, "num_entries": portrait_num_entries, "summary_path": portrait_payload.get("summary_path"), } ), "args": vars(args), }, ) def main() -> None: args = parse_args() populate_two_roi_inference_args(args) if args.torch_threads and args.torch_threads > 0: torch.set_num_threads(int(args.torch_threads)) torch.set_num_interop_threads(max(1, min(int(args.torch_threads), 4))) data_yaml = Path(args.data).resolve() data_cfg = load_yaml(data_yaml) dataset_root = data_cfg.get("path") split_path = resolve_data_path(data_yaml, dataset_root, data_cfg.get(args.split)) entries = load_split_entries( split_path, max_samples=args.max_samples, sample_selection=str(args.sample_selection), sample_random_seed=int(args.sample_random_seed), ) image_root = resolve_data_path(data_yaml, None, dataset_root) args.roi0_data = args.roi0_data or args.data args.roi1_data = args.roi1_data or args.data requested_rois = {roi.lower() for roi in args.analyze_rois} context = build_two_roi_inference_context_from_args(args, requested_rois=requested_rois) output_root = Path(args.output_root).resolve() output_root.mkdir(parents=True, exist_ok=True) class_map = {str(key): int(value) for key, value in (data_cfg.get("class_map") or {}).items()} difficulty_weights = [float(v) for v in data_cfg.get("difficulty_weights", [1.0, 1.0, 1.0, 1.0])] face_3d_classes = set(int(v) for v in data_cfg.get("face_3d_classes", [])) complete_3d_classes = set(int(v) for v in data_cfg.get("complete_3d_classes", [])) include_classes = None if args.classes is None else set(int(v) for v in args.classes) min_wh_px = float(data_cfg.get("min_wh", 2.0)) ori_img_size_cfg = data_cfg.get("ori_img_size", [1920, 1080]) ori_img_size = ( int(ori_img_size_cfg[0]) if isinstance(ori_img_size_cfg, (list, tuple)) and len(ori_img_size_cfg) > 0 else 1920, int(ori_img_size_cfg[1]) if isinstance(ori_img_size_cfg, (list, tuple)) and len(ori_img_size_cfg) > 1 else 1080, ) class_names = infer_class_name_map(class_map, names_to_dict(context.roi_models[0].names) if context.roi_models else None) model_paths = {bundle.spec.name.lower(): bundle.spec.model_path for bundle in context.roi_models} start = time.time() summary_by_roi = {} for bundle in context.roi_models: print(f"\nAnalyzing {bundle.spec.name} on {len(entries)} {args.split} samples...") summary_by_roi[bundle.spec.name.lower()] = run_roi_analysis( bundle=bundle, entries=entries, image_root=image_root, class_map=class_map, difficulty_weights=difficulty_weights, face_3d_classes=face_3d_classes, complete_3d_classes=complete_3d_classes, classes=include_classes, min_wh_px=min_wh_px, face_visibility_score_thresh=float(args.face_visibility_score_thresh), yaw_bad_threshold_deg=float(args.yaw_bad_threshold_deg), edge_yaw_max_lateral_dist_m=float(args.edge_yaw_max_lateral_dist), yaw_compare_max_lateral_dist_m=float(args.yaw_compare_max_lateral_dist), horizontal_bad_threshold_m=float(args.horizontal_bad_threshold_m), vertical_bad_threshold_m=float(args.vertical_bad_threshold_m), topk_badcases=int(args.topk_badcases), per_class_badcases=int(args.per_class_badcases), error_bin_badcases=int(args.error_bin_badcases), error_bin_samples_per_bin=int(args.error_bin_samples_per_bin), badcase_random_seed=int(args.badcase_random_seed), output_root=output_root, log_every=int(args.log_every), split_file=str(split_path), inference_batch_size=int(args.inference_batch_size), ) write_incremental_combined_outputs( output_root=output_root, data_yaml=data_yaml, split_path=split_path, image_root=image_root, summary_by_roi=summary_by_roi, num_entries=len(entries), args=args, elapsed_minutes=(time.time() - start) / 60.0, portrait_payload=None, ) print(f"Partial combined report updated: {output_root / 'report.html'}") kpi_elapsed_minutes = (time.time() - start) / 60.0 write_incremental_combined_outputs( output_root=output_root, data_yaml=data_yaml, split_path=split_path, image_root=image_root, summary_by_roi=summary_by_roi, num_entries=len(entries), args=args, elapsed_minutes=kpi_elapsed_minutes, portrait_payload=None, ) write_run_manifest( output_root=output_root, data_yaml=data_yaml, split_path=split_path, image_root=image_root, num_entries=len(entries), requested_rois=requested_rois, model_paths=model_paths, args=args, ) print(f"\nKPI report ready: {output_root / 'report.html'}") portrait_payload = None portrait_split_path = None portrait_entries = None if not args.skip_data_portrait: portrait_payload = load_existing_data_portrait(output_root, str(args.data_portrait_split)) if portrait_payload is not None: portrait_split_path_raw = portrait_payload.get("split_path") portrait_split_path = None if not portrait_split_path_raw else Path(str(portrait_split_path_raw)) print( f"\nReusing existing {args.data_portrait_split} data portrait from " f"{portrait_payload.get('summary_path', output_root / f'{str(args.data_portrait_split).lower()}_portrait' / 'summary.json')}." ) else: portrait_split_value = data_cfg.get(args.data_portrait_split) if portrait_split_value is None: print(f"\nSkipping data portrait because split `{args.data_portrait_split}` is not defined in {data_yaml}.") else: portrait_split_path = resolve_data_path(data_yaml, dataset_root, portrait_split_value) portrait_entries = load_split_entries( portrait_split_path, max_samples=int(args.data_portrait_max_samples), sample_selection=str(args.data_portrait_sample_selection), sample_random_seed=int(args.data_portrait_random_seed), ) print(f"\nBuilding {args.data_portrait_split} data portrait on {len(portrait_entries)} samples...") portrait_payload = build_data_portrait( entries=portrait_entries, split_name=str(args.data_portrait_split), split_path=portrait_split_path, image_root=image_root, output_root=output_root, class_map=class_map, class_names=class_names, difficulty_weights=difficulty_weights, face_3d_classes=face_3d_classes, complete_3d_classes=complete_3d_classes, ori_img_size=ori_img_size, face_visibility_score_thresh=float(args.face_visibility_score_thresh), log_every=int(args.log_every), workers=int(args.data_portrait_workers), ) write_run_manifest( output_root=output_root, data_yaml=data_yaml, split_path=split_path, image_root=image_root, num_entries=len(entries), requested_rois=requested_rois, model_paths=model_paths, args=args, portrait_payload=portrait_payload, portrait_split_path=portrait_split_path, portrait_entries=portrait_entries, ) total_minutes = (time.time() - start) / 60.0 write_incremental_combined_outputs( output_root=output_root, data_yaml=data_yaml, split_path=split_path, image_root=image_root, summary_by_roi=summary_by_roi, num_entries=len(entries), args=args, elapsed_minutes=total_minutes, portrait_payload=portrait_payload, ) print("\nAnalysis complete.") print(f"Output root: {output_root}") print(f"HTML report: {output_root / 'report.html'}") if portrait_payload is not None: print(f"Data portrait summary: {portrait_payload.get('summary_path')}") for roi_name, payload in summary_by_roi.items(): overall = payload["overall"] threshold_advice_2d = payload.get("threshold_advice_2d") or {} print( f"[{roi_name}] matched_3d={overall['matched_3d']} yaw_mae_deg={overall['yaw_mae_deg']} " f"yaw_compare_count={overall['yaw_compare_count']} edge_based_mae_deg={overall['edge_based_yaw_mae_deg']} " f"x_bad={overall['x_bad_count']} z_bad={overall['z_bad_count']} " f"recommended_2d_conf={format_float(to_float(threshold_advice_2d.get('recommended_confidence')), 3)}" ) if __name__ == "__main__": main()