""" Main evaluator module that orchestrates the evaluation process. """ import os import json import glob import numpy as np from pathlib import Path from tqdm import tqdm from PIL import Image from multiprocessing import Pool, cpu_count from concurrent.futures import ThreadPoolExecutor, as_completed from functools import partial from .parser import GroundTruthParser, DetectionParser from .matcher import Matcher2D from .metrics_2d import Metrics2D from .metrics_3d import Metrics3D, normalize_angle from .roi_processor import ROIProcessor from ..class_config import NUM_CLASSES class Evaluator: """Main evaluator class for 2D and 3D detection metrics.""" def __init__(self, config=None, iou_threshold=0.5, num_workers=None, save_detailed_matches=False): """ Initialize evaluator. Args: config: dict, configuration parameters (optional) iou_threshold: float, IoU threshold for matching num_workers: int, number of parallel workers (default: cpu_count()) save_detailed_matches: bool, whether to save detailed 3D match information """ self.config = config or {} self.iou_threshold = iou_threshold self.save_detailed_matches = save_detailed_matches self.coord_system = config.get('metrics_3d', {}).get('coordinate_system', 'camera') if self.coord_system not in ('camera', 'ego'): raise ValueError(f"Unsupported 3D coordinate system: {self.coord_system}") self.multi_roi_eval_config = self._build_multi_roi_eval_config(config) self.multi_roi_enabled = bool(self.multi_roi_eval_config) # Initialize ROI processor if config provided self.roi_processor = None roi_gt_config = config.get('roi_gt', {}) if roi_gt_config.get('enabled', False) and not self.multi_roi_enabled and roi_gt_config.get('roi_config') is not None: calib_root = roi_gt_config.get('calib_root') roi_config = roi_gt_config.get('roi_config') roi_bottom_offset = roi_gt_config.get('roi_bottom_offset', 0) roi_right_offset = roi_gt_config.get('roi_right_offset', 0) roi_use_true_vp_x = roi_gt_config.get('roi_use_true_vp_x', False) ori_img_size = config.get('image', {}).get('width', 1920), config.get('image', {}).get('height', 1080) self.roi_processor = ROIProcessor( calib_root=calib_root, roi_config=roi_config, ori_img_size=ori_img_size, roi_bottom_offset=roi_bottom_offset, roi_right_offset=roi_right_offset, roi_use_true_vp_x=roi_use_true_vp_x, ) print( "ROI processor enabled for GT filtering with config: " f"{roi_config}, roi_bottom_offset: {roi_bottom_offset}, " f"roi_right_offset: {roi_right_offset}, roi_use_true_vp_x: {roi_use_true_vp_x}" ) # Calculate minimum box size for GT filtering based on ROI and model input size self.min_box_size = self._calculate_min_box_size(config, roi_gt_config) self.default_min_box_size = self.min_box_size if self.multi_roi_enabled: self.default_min_box_size = min(roi_cfg['min_box_size'] for roi_cfg in self.multi_roi_eval_config.values()) print("Multi-ROI merged evaluation enabled with per-ROI min box sizes:") for roi_id, roi_cfg in sorted(self.multi_roi_eval_config.items()): print( f" ROI{roi_id}: min_box_size={roi_cfg['min_box_size']:.2f}, " f"roi_config={roi_cfg['roi_config']}, input_size={roi_cfg['input_size']}" ) # Initialize parsers parser_min_box_size = 0 if self.multi_roi_enabled else self.min_box_size self.gt_parser = GroundTruthParser(min_box_size=parser_min_box_size, coord_system=self.coord_system) self.det_parser = DetectionParser(min_box_size=parser_min_box_size, coord_system=self.coord_system) # Initialize matcher self.matcher = Matcher2D(iou_threshold=iou_threshold) # Initialize metrics calculators distance_ranges_2d = config.get('metrics_2d', {}).get('distance_ranges', None) lateral_roi_2d = config.get('metrics_2d', {}).get('lateral_roi', None) self.metrics_2d = Metrics2D(num_classes=NUM_CLASSES, distance_ranges=distance_ranges_2d, lateral_roi=lateral_roi_2d, coord_system=self.coord_system) self.distance_ranges_2d = distance_ranges_2d self.lateral_roi_2d = lateral_roi_2d # Get distance ranges from config distance_ranges = config.get('metrics_3d', {}).get('distance_ranges', None) lateral_distance_ranges = config.get('metrics_3d', {}).get('lateral_distance_ranges', None) heading_tolerance = config.get('metrics_3d', {}).get('heading_tolerance', 'strict') self.vehicle_size_split_config = config.get('metrics_3d', {}).get('vehicle_size_split', {}) self.metrics_3d = Metrics3D( distance_ranges=distance_ranges, lateral_distance_ranges=lateral_distance_ranges, heading_tolerance=heading_tolerance, coord_system=self.coord_system, vehicle_size_split=self.vehicle_size_split_config, ) self.heading_tolerance = heading_tolerance # Per-case metrics storage self.per_case_metrics_2d = {} self.per_case_metrics_3d = {} self.per_case_frame_stats = {} self.per_case_aggregate_stats = {} self.distance_ranges = distance_ranges self.lateral_distance_ranges = lateral_distance_ranges # Detailed 3D matches storage (for common match comparison) self.detailed_3d_matches = {} if save_detailed_matches else None # Image pairs to evaluate self.image_pairs = [] # Multiprocessing settings self.num_workers = num_workers if num_workers is not None else max(1, cpu_count() - 1) print(f"Using {self.num_workers} worker(s) for parallel processing") def _get_lateral_axis(self): return 0 if self.coord_system == 'camera' else 1 def _get_longitudinal_axis(self): return 2 if self.coord_system == 'camera' else 0 @staticmethod def _build_class_count_stats(match_result, require_3d=False): gts_filtered = match_result['gts_filtered'] dets_sorted = match_result['dets_sorted'] matches = match_result['matches'] def _obj_id(obj, fallback_prefix, fallback_idx): object_id = obj.get('id') if object_id is None: return f"{fallback_prefix}_{fallback_idx}" return str(object_id) if not require_3d: gt_ids = [_obj_id(gt, 'gt', idx) for idx, gt in enumerate(gts_filtered)] det_ids = [_obj_id(det, 'det', idx) for idx, det in enumerate(dets_sorted)] matched_pairs = [ { 'gt_id': _obj_id(gts_filtered[gt_idx], 'gt', gt_idx), 'det_id': _obj_id(dets_sorted[det_idx], 'det', det_idx), 'iou': float(iou), } for gt_idx, det_idx, iou in matches ] matched_gt_ids = {pair['gt_id'] for pair in matched_pairs} matched_det_ids = {pair['det_id'] for pair in matched_pairs} return { 'gt_count': len(gts_filtered), 'det_count': len(dets_sorted), 'match_count': len(matches), 'gt_ids': gt_ids, 'det_ids': det_ids, 'matched_pairs': matched_pairs, 'unmatched_gt_ids': [gt_id for gt_id in gt_ids if gt_id not in matched_gt_ids], 'unmatched_det_ids': [det_id for det_id in det_ids if det_id not in matched_det_ids], } valid_gt = [(idx, gt) for idx, gt in enumerate(gts_filtered) if gt.get('has_3d', False)] valid_det = [(idx, det) for idx, det in enumerate(dets_sorted) if det.get('3d_info') is not None] gt_ids = [_obj_id(gt, 'gt', idx) for idx, gt in valid_gt] det_ids = [_obj_id(det, 'det', idx) for idx, det in valid_det] matched_pairs = [ { 'gt_id': _obj_id(gts_filtered[gt_idx], 'gt', gt_idx), 'det_id': _obj_id(dets_sorted[det_idx], 'det', det_idx), 'iou': float(iou), } for gt_idx, det_idx, iou in matches if gts_filtered[gt_idx].get('has_3d', False) and dets_sorted[det_idx].get('3d_info') is not None ] matched_gt_ids = {pair['gt_id'] for pair in matched_pairs} matched_det_ids = {pair['det_id'] for pair in matched_pairs} return { 'gt_count': len(valid_gt), 'det_count': len(valid_det), 'match_count': len(matched_pairs), 'gt_ids': gt_ids, 'det_ids': det_ids, 'matched_pairs': matched_pairs, 'unmatched_gt_ids': [gt_id for gt_id in gt_ids if gt_id not in matched_gt_ids], 'unmatched_det_ids': [det_id for det_id in det_ids if det_id not in matched_det_ids], } @staticmethod def _aggregate_case_stats(frame_stats_by_frame): aggregate = {'2d': {}, '3d': {}} for frame_stats in frame_stats_by_frame.values(): for eval_type in ('2d', '3d'): for class_name, stats in frame_stats.get(eval_type, {}).items(): class_total = aggregate[eval_type].setdefault( class_name, {'gt_count': 0, 'det_count': 0, 'match_count': 0, 'num_frames': 0, 'gt_ids': [], 'det_ids': [], 'matched_pairs': []} ) class_total['gt_count'] += stats['gt_count'] class_total['det_count'] += stats['det_count'] class_total['match_count'] += stats['match_count'] class_total['num_frames'] += 1 class_total['gt_ids'].extend(stats.get('gt_ids', [])) class_total['det_ids'].extend(stats.get('det_ids', [])) class_total['matched_pairs'].extend(stats.get('matched_pairs', [])) return aggregate @staticmethod def _normalize_roi_id(roi_id): """Normalize ROI identifiers like 'roi0'/'0' to plain numeric strings.""" if roi_id is None: return None roi_id_str = str(roi_id).strip().lower() if roi_id_str.startswith('roi'): roi_id_str = roi_id_str[3:] return roi_id_str or None @classmethod def _infer_source_roi_from_path(cls, path_obj): """Infer ROI id from a directory path like predictions/roi0 or json_results/roi1.""" if path_obj is None: return None for part in reversed(Path(path_obj).parts): normalized = cls._normalize_roi_id(part) if normalized is not None and str(part).strip().lower().startswith('roi'): return normalized return None @staticmethod def _is_empty_gt_json_file(file_path): """ Check whether a GT JSON file is structurally empty. Empty GT frames should be skipped entirely during evaluation so their paired detections do not get counted as false positives. Returns: tuple[bool, str | None]: (is_empty, reason) """ try: with open(file_path, 'r', encoding='utf-8') as f: raw_content = f.read() except OSError: return False, None if not raw_content.strip(): return True, "blank file" try: data = json.loads(raw_content) except json.JSONDecodeError: return False, None if data is None: return True, "null root" if isinstance(data, dict) and not data: return True, "empty object" if isinstance(data, list) and not data: return True, "empty list" return False, None @staticmethod def _extract_roi_width(roi_config): """Extract ROI width from ROI config in either size or bounds form.""" if isinstance(roi_config, (list, tuple)): if len(roi_config) == 2: return roi_config[0] if len(roi_config) == 4: return roi_config[2] - roi_config[0] raise ValueError(f"Invalid ROI config format: {roi_config}") if isinstance(roi_config, dict): mode = roi_config.get('mode') if mode == 'size': return roi_config.get('width', 1920) if mode == 'bounds': return roi_config.get('x2', 1920) - roi_config.get('x1', 0) raise ValueError(f"Invalid ROI config mode: {mode}") raise ValueError(f"Unknown ROI config type: {type(roi_config)}") @classmethod def _calculate_min_box_size_from_params(cls, roi_config, model_input_width, min_box_at_input): """Calculate minimum box size at original image scale for one ROI branch.""" roi_width = cls._extract_roi_width(roi_config) return min_box_at_input * roi_width / model_input_width @classmethod def _build_multi_roi_eval_config(cls, config): """Build normalized per-ROI evaluation config for merged-result evaluation.""" roi_gt_config = config.get('roi_gt', {}) roi_entries = roi_gt_config.get('rois') if not isinstance(roi_entries, dict) or len(roi_entries) < 2: return None model_config = config.get('model', {}) input_size_by_roi = ( model_config.get('input_size_by_roi') or model_config.get('input_sizes_by_roi') or {} ) min_box_at_input_by_roi = model_config.get('min_box_size_at_input_scale_by_roi', {}) normalized = {} for raw_roi_id, roi_cfg in roi_entries.items(): roi_id = cls._normalize_roi_id(raw_roi_id) if roi_id is None: continue roi_config = roi_cfg.get('roi_config') if roi_config is None: raise ValueError(f"Missing roi_config for ROI entry: {raw_roi_id}") input_size = input_size_by_roi.get(raw_roi_id, input_size_by_roi.get(roi_id, model_config.get('input_size', 704))) min_box_at_input = min_box_at_input_by_roi.get( raw_roi_id, min_box_at_input_by_roi.get(roi_id, model_config.get('min_box_size_at_input_scale', 8)) ) min_box_size = cls._calculate_min_box_size_from_params(roi_config, input_size, min_box_at_input) normalized[roi_id] = { 'roi_config': roi_config, 'roi_bottom_offset': roi_cfg.get('roi_bottom_offset', 0), 'roi_right_offset': roi_cfg.get('roi_right_offset', 0), 'roi_use_true_vp_x': roi_cfg.get('roi_use_true_vp_x', False), 'calib_root': roi_cfg.get('calib_root', roi_gt_config.get('calib_root')), 'input_size': input_size, 'min_box_at_input': min_box_at_input, 'min_box_size': min_box_size, } return normalized or None @staticmethod def _bbox_meets_min_size(bbox, min_box_size): """Check whether a bbox is large enough under the configured min-box threshold.""" if min_box_size is None or min_box_size <= 0: return True return (bbox[2] - bbox[0]) >= min_box_size and (bbox[3] - bbox[1]) >= min_box_size @staticmethod def _create_roi_processor(roi_cfg, img_width, img_height): """Build a ROIProcessor instance from serialized config.""" return ROIProcessor( calib_root=roi_cfg.get('calib_root'), roi_config=roi_cfg.get('roi_config'), ori_img_size=(img_width, img_height), roi_bottom_offset=roi_cfg.get('roi_bottom_offset', 0), roi_right_offset=roi_cfg.get('roi_right_offset', 0), roi_use_true_vp_x=roi_cfg.get('roi_use_true_vp_x', False), ) @classmethod def _parse_ground_truths_for_pair(cls, pair, coord_system): """Parse and ROI-filter GTs for either single-ROI or merged multi-ROI evaluation.""" if 'multi_roi_eval_config' not in pair: gt_parser = GroundTruthParser(min_box_size=pair.get('min_box_size', 8), coord_system=coord_system) gts = gt_parser.parse_file(pair['gt_file'], pair['img_width'], pair['img_height']) if 'roi_processor_config' in pair: roi_processor = cls._create_roi_processor(pair['roi_processor_config'], pair['img_width'], pair['img_height']) gts, _ = roi_processor.process_case_frame( pair['case'], pair['frame'], gts, level1_name=pair.get('level1_name') ) return gts gt_parser = GroundTruthParser(min_box_size=0, coord_system=coord_system) raw_gts = gt_parser.parse_file(pair['gt_file'], pair['img_width'], pair['img_height']) for source_index, gt in enumerate(raw_gts): gt['_source_index'] = source_index merged_gts = {} for roi_id, roi_cfg in sorted(pair['multi_roi_eval_config'].items()): roi_processor = cls._create_roi_processor(roi_cfg, pair['img_width'], pair['img_height']) roi_gts, _ = roi_processor.process_case_frame( pair['case'], pair['frame'], raw_gts, level1_name=pair.get('level1_name') ) for ann in roi_gts: if not cls._bbox_meets_min_size(ann['bbox_2d'], roi_cfg['min_box_size']): continue source_index = ann.get('_source_index') if source_index is None: continue merged_ann = merged_gts.get(source_index) if merged_ann is None: merged_ann = ann.copy() merged_ann['bbox_2d_by_roi'] = {} merged_ann['valid_roi_ids'] = [] merged_gts[source_index] = merged_ann merged_ann['bbox_2d_by_roi'][roi_id] = ann['bbox_2d'] if roi_id not in merged_ann['valid_roi_ids']: merged_ann['valid_roi_ids'].append(roi_id) merged_ann['bbox_2d'] = ann['bbox_2d'] merged_ann['roi_filtered'] = True merged_ann['was_clipped'] = merged_ann.get('was_clipped', False) or ann.get('was_clipped', False) result = [] for source_index in sorted(merged_gts): ann = merged_gts[source_index] ann.pop('_source_index', None) result.append(ann) return result @classmethod def _parse_detections_for_pair(cls, pair, coord_system): """Parse and ROI-aware-filter detections for either single-ROI or merged evaluation.""" if 'multi_roi_eval_config' not in pair: det_parser = DetectionParser(min_box_size=pair.get('min_box_size', 8), coord_system=coord_system) dets = det_parser.parse_file(pair['det_file']) det_source_roi = pair.get('det_source_roi') if det_source_roi is not None: for det in dets: if cls._normalize_roi_id(det.get('roi_id')) is None: det['roi_id'] = det_source_roi det_roi_filter = pair.get('det_roi_filter') if det_roi_filter is not None: dets = [d for d in dets if cls._normalize_roi_id(d.get('roi_id')) == det_roi_filter] return dets det_parser = DetectionParser(min_box_size=0, coord_system=coord_system) dets = det_parser.parse_file(pair['det_file']) filtered_dets = [] default_min_box_size = pair.get('default_min_box_size', 0) for det in dets: roi_id = cls._normalize_roi_id(det.get('roi_id')) roi_cfg = pair['multi_roi_eval_config'].get(roi_id) min_box_size = roi_cfg['min_box_size'] if roi_cfg is not None else default_min_box_size if cls._bbox_meets_min_size(det['bbox_2d'], min_box_size): det['roi_id'] = roi_id filtered_dets.append(det) return filtered_dets def _calculate_min_box_size(self, config, roi_gt_config): """ Calculate minimum box size for GT filtering based on ROI and model configuration. Formula: min_box_size = min_box_size_at_input_scale * roi_width / model_input_width Examples: - ROI0 (1920->704): 8 * 1920 / 704 ≈ 21.82 - ROI1 (704->704): 8 * 704 / 704 = 8.0 Args: config: dict, full configuration roi_gt_config: dict, ROI ground truth configuration Returns: float, minimum box size in pixels at original image scale """ model_config = config.get('model', {}) model_input_width = model_config.get('input_size', 704) min_box_at_input = model_config.get('min_box_size_at_input_scale', 8) # If ROI is not enabled, use default if not roi_gt_config.get('enabled', False): print(f"ROI GT disabled, using default min_box_size: {min_box_at_input}") return min_box_at_input # Get ROI width from roi_config roi_config = roi_gt_config.get('roi_config') if roi_config is None: print(f"ROI config not found, using default min_box_size: {min_box_at_input}") return min_box_at_input try: roi_width = self._extract_roi_width(roi_config) except ValueError: print(f"Invalid ROI config, using default min_box_size: {min_box_at_input}") return min_box_at_input # Calculate minimum box size at original image scale min_box_size = min_box_at_input * roi_width / model_input_width print(f"Calculated min_box_size for GT filtering: {min_box_size:.2f} pixels " f"(ROI width: {roi_width}, Model input: {model_input_width}, " f"Min at input scale: {min_box_at_input})") return min_box_size def load_data_from_paths(self, det_root, gt_root, img_width=1920, img_height=1080, path_depth=1, det_format='auto', gt_format='auto'): """ Load detection and ground truth data from directory structure. Directory structure (path_depth=1): det_root/ case1/ json_results/ # JSON format frame001.json txt_results/ # TXT format frame001.txt case2/ ... gt_root/ case1/ labels_json/ # JSON format frame001.json labels/ # TXT format frame001.txt case2/ ... Directory structure (path_depth=2): det_root/ level1/ case1/ json_results/ frame001.json case2/ ... gt_root/ level1/ case1/ labels_json/ frame001.json case2/ ... Args: det_root: str, root directory containing detection results gt_root: str, root directory containing ground truth labels img_width: int, image width (default 1920) img_height: int, image height (default 1080) path_depth: int, directory depth (1 or 2, default 1) det_format: str, detection file format: 'auto' (default), 'json', or 'txt'. 'auto' probes json_results/ first then txt_results/. gt_format: str, ground truth file format: 'auto' (default), 'json', or 'txt'. 'auto' probes labels_json/ first then labels/. """ self.image_pairs = [] if self.coord_system == 'ego' and (det_format == 'txt' or gt_format == 'txt'): raise ValueError("Ego-coordinate evaluation requires JSON detection and JSON ground-truth inputs.") # Find all case directories based on path depth if path_depth == 1: # 1-level: det_root/case/txt_results det_cases = [d for d in Path(det_root).iterdir() if d.is_dir()] case_info_list = [(d, d.name, None) for d in det_cases] # (path, case_name, level1_name) elif path_depth == 2: # 2-level: det_root/level1/case/txt_results case_info_list = [] level1_dirs = [d for d in Path(det_root).iterdir() if d.is_dir()] for level1_dir in level1_dirs: level1_name = level1_dir.name case_dirs = [d for d in level1_dir.iterdir() if d.is_dir()] for case_dir in case_dirs: case_info_list.append((case_dir, case_dir.name, level1_name)) else: raise ValueError(f"Unsupported path_depth: {path_depth}. Must be 1 or 2.") print(f"Found {len(case_info_list)} case(s) in detection root: {det_root} (path_depth={path_depth})") skip_cases = frozenset([ '019b6bf2-01a6-7029-9232-fce2bbcd2d73', '019b6bf2-0124-7820-91b7-c5eb42150cd2', '019b8ddb-ae8a-70f3-b86f-055894c79724', ]) # Build ROI processor config once (passed into each worker) roi_processor_config = None if self.roi_processor is not None: roi_gt_config = self.config.get('roi_gt', {}) roi_processor_config = { 'calib_root': roi_gt_config.get('calib_root'), 'roi_config': roi_gt_config.get('roi_config'), 'roi_bottom_offset': roi_gt_config.get('roi_bottom_offset', 0), 'roi_right_offset': roi_gt_config.get('roi_right_offset', 0), 'roi_use_true_vp_x': roi_gt_config.get('roi_use_true_vp_x', False), } multi_roi_eval_config = self.multi_roi_eval_config if self.multi_roi_enabled else None det_subdir = self.config.get('dataset', {}).get('det_subdir') if det_subdir is not None: det_subdir = str(det_subdir).strip() if not det_subdir: det_subdir = None det_roi_filter = self.config.get('dataset', {}).get('det_roi_filter') if det_roi_filter is not None: det_roi_filter = str(det_roi_filter).strip() print(f"ROI filter: evaluating only roi_id={det_roi_filter!r} detections") if det_subdir is not None: print(f"Detection subdirectory override: {det_subdir}") # Parallel case scanning with threads (I/O-bound) scan_workers = min(self.num_workers, len(case_info_list)) if case_info_list else 1 all_pairs = [] with ThreadPoolExecutor(max_workers=scan_workers) as executor: futures = { executor.submit( self._scan_case, ci, gt_root, path_depth, img_width, img_height, self.min_box_size, skip_cases, roi_processor_config, multi_roi_eval_config, self.default_min_box_size, self.coord_system, det_format, gt_format, det_roi_filter, det_subdir ): ci for ci in case_info_list } for future in tqdm(as_completed(futures), total=len(futures), desc="Scanning cases"): pairs, warnings = future.result() for w in warnings: print(w) all_pairs.extend(pairs) # Sort for deterministic order (threads return out-of-order) self.image_pairs = sorted(all_pairs, key=lambda p: (p['level1_name'] or '', p['case'], p['frame'])) print(f"Loaded {len(self.image_pairs)} image pairs for evaluation") @staticmethod def _scan_case(case_info, gt_root, path_depth, img_width, img_height, min_box_size, skip_cases, roi_processor_config, multi_roi_eval_config=None, default_min_box_size=8, coord_system='camera', det_format='auto', gt_format='auto', det_roi_filter=None, det_subdir=None): """ Scan a single case directory and return its image pairs (thread worker). Supports both TXT and JSON file formats: - Detection: case/json_results/*.json or case/txt_results/*.txt - Ground truth: case/labels_json/*.json or case/labels/*.txt Args: det_format: 'auto' | 'json' | 'txt' — explicit format overrides auto-detection gt_format: 'auto' | 'json' | 'txt' — explicit format overrides auto-detection Returns: list of image pair dicts, and list of warning strings """ det_case_dir, case_name, level1_name = case_info pairs = [] warnings = [] if case_name in skip_cases: return pairs, warnings # Corresponding GT case directory if path_depth == 1: gt_case_dir = Path(gt_root) / case_name else: gt_case_dir = Path(gt_root) / level1_name / case_name if not gt_case_dir.exists(): warnings.append(f"Warning: GT case directory not found for {case_name}, skipping") return pairs, warnings # -------- Resolve detection directory / glob -------- det_json_dirs = [] if det_subdir is not None: det_json_dirs.append(Path(det_subdir) if Path(det_subdir).is_absolute() else det_case_dir / det_subdir) if det_roi_filter is not None: det_json_dirs.extend([ det_case_dir / "json_results" / f"roi{det_roi_filter}", det_case_dir / "predictions" / f"roi{det_roi_filter}", ]) det_json_dirs.extend([ det_case_dir / "json_results", det_case_dir / "predictions", ]) deduped_det_json_dirs = [] seen_det_dirs = set() for candidate_dir in det_json_dirs: candidate_key = str(candidate_dir) if candidate_key in seen_det_dirs: continue seen_det_dirs.add(candidate_key) deduped_det_json_dirs.append(candidate_dir) det_json_dirs = deduped_det_json_dirs det_txt_dir = det_case_dir / "txt_results" det_results_dir = None det_glob = None det_source_roi = None def _pick_json_dir(): nonlocal det_results_dir, det_glob, det_source_roi empty_existing_dirs = [] for candidate_dir in det_json_dirs: if not candidate_dir.exists(): continue matches = sorted(candidate_dir.glob("*.json")) if matches: det_results_dir = candidate_dir det_glob = "*.json" det_source_roi = Evaluator._infer_source_roi_from_path(candidate_dir) return empty_existing_dirs empty_existing_dirs.append(candidate_dir) return empty_existing_dirs if det_format == 'json': empty_existing_dirs = _pick_json_dir() if det_results_dir is None: if empty_existing_dirs: empty_dirs_str = ", ".join(str(d) for d in empty_existing_dirs) warnings.append( f"Warning: JSON detection directories exist but contain no *.json files for {case_name}: " f"{empty_dirs_str}, skipping" ) else: warnings.append( f"Warning: no JSON detection directory found for {case_name} " f"(checked: {', '.join(str(d) for d in det_json_dirs)}), skipping" ) return pairs, warnings elif det_format == 'txt': det_results_dir, det_glob = det_txt_dir, "*.txt" if not det_results_dir.exists(): warnings.append(f"Warning: txt_results not found in {case_name} (det_format=txt), skipping") return pairs, warnings else: # auto empty_existing_dirs = _pick_json_dir() if det_results_dir is not None: pass elif det_txt_dir.exists() and coord_system != 'ego': det_results_dir, det_glob = det_txt_dir, "*.txt" else: if coord_system == 'ego': if empty_existing_dirs: empty_dirs_str = ", ".join(str(d) for d in empty_existing_dirs) warnings.append( f"Warning: JSON detection directories exist but contain no *.json files for " f"{case_name} in ego-coordinate evaluation: {empty_dirs_str}, skipping" ) else: warnings.append( f"Warning: no JSON detection directory found in {case_name} for ego-coordinate " f"evaluation (checked: {', '.join(str(d) for d in det_json_dirs)}), skipping" ) else: if empty_existing_dirs: empty_dirs_str = ", ".join(str(d) for d in empty_existing_dirs) warnings.append( f"Warning: JSON detection directories exist but contain no *.json files for {case_name}: " f"{empty_dirs_str}; txt_results also not usable, skipping" ) else: warnings.append( f"Warning: neither JSON detection directories " f"({', '.join(str(d) for d in det_json_dirs)}) nor txt_results found in {case_name}, skipping" ) return pairs, warnings # -------- Resolve ground-truth directory / extension -------- gt_json_dirs = [ # gt_case_dir / "labels_20260320_with_ego", # gt_case_dir / "labels_20260320_json", gt_case_dir / "labels_json", ] gt_txt_dir = gt_case_dir / "labels" if gt_format == 'json': gt_labels_dir = next((d for d in gt_json_dirs if d.exists()), None) gt_ext = ".json" if gt_labels_dir is None: warnings.append(f"Warning: labels_json not found in {case_name} (gt_format=json), skipping") return pairs, warnings elif gt_format == 'txt': gt_labels_dir, gt_ext = gt_txt_dir, ".txt" if not gt_labels_dir.exists(): warnings.append(f"Warning: labels not found in {case_name} (gt_format=txt), skipping") return pairs, warnings else: # auto gt_labels_dir = next((d for d in gt_json_dirs if d.exists()), None) if gt_labels_dir is not None: gt_ext = ".json" elif gt_txt_dir.exists() and coord_system != 'ego': gt_labels_dir, gt_ext = gt_txt_dir, ".txt" else: if coord_system == 'ego': warnings.append(f"Warning: labels_20260320 not found in {case_name} for ego-coordinate evaluation, skipping") else: warnings.append(f"Warning: Neither labels_json nor labels directory found in {case_name}, skipping") return pairs, warnings # Pre-build a set of available GT frames to avoid per-file exists() calls gt_frames = {p.stem for p in gt_labels_dir.iterdir() if p.suffix == gt_ext} det_files = sorted(det_results_dir.glob(det_glob)) if not det_files: warnings.append( f"Warning: no detection files matched {det_glob} in {det_results_dir} for {case_name}, skipping" ) return pairs, warnings skipped_empty_gt_frames = [] for det_file in det_files: frame_name = det_file.stem if frame_name not in gt_frames: warnings.append(f"Warning: GT file not found for {frame_name} in {case_name}, skipping") continue gt_file = gt_labels_dir / f"{frame_name}{gt_ext}" if gt_ext == ".json": is_empty_gt_json, empty_reason = Evaluator._is_empty_gt_json_file(gt_file) if is_empty_gt_json: skipped_empty_gt_frames.append((frame_name, empty_reason)) continue pair = { 'case': case_name, 'frame': frame_name, 'det_file': str(det_file), 'gt_file': str(gt_file), 'img_width': img_width, 'img_height': img_height, 'min_box_size': min_box_size, 'level1_name': level1_name, 'coord_system': coord_system, } if roi_processor_config is not None: pair['roi_processor_config'] = roi_processor_config if multi_roi_eval_config is not None: pair['multi_roi_eval_config'] = multi_roi_eval_config pair['default_min_box_size'] = default_min_box_size if det_source_roi is not None: pair['det_source_roi'] = det_source_roi if det_roi_filter is not None: pair['det_roi_filter'] = det_roi_filter pairs.append(pair) if skipped_empty_gt_frames: preview = ", ".join( f"{frame} ({reason})" if reason else frame for frame, reason in skipped_empty_gt_frames[:5] ) remaining = len(skipped_empty_gt_frames) - min(len(skipped_empty_gt_frames), 5) if remaining > 0: preview = f"{preview}, ... (+{remaining} more)" warnings.append( f"Warning: skipped {len(skipped_empty_gt_frames)} frame(s) with empty GT JSON in " f"{case_name}: {preview}" ) return pairs, warnings @staticmethod def _process_frame_2d(pair, iou_threshold): """ Process a single frame for 2D evaluation (worker function). Args: pair: dict, image pair information (may contain roi_processor_config and min_box_size) iou_threshold: float, IoU threshold Returns: tuple: (case_name, frame_name, per_class_results) """ coord_system = pair.get('coord_system', 'camera') matcher = Matcher2D(iou_threshold=iou_threshold) gts = Evaluator._parse_ground_truths_for_pair(pair, coord_system) dets = Evaluator._parse_detections_for_pair(pair, coord_system) # Match and compute for each class class_results = {} for class_id in range(NUM_CLASSES): match_result = matcher.match(gts, dets, class_id) class_results[class_id] = { 'match_result': match_result, 'gts': gts, 'dets': dets } return (pair['case'], pair['frame'], class_results) @staticmethod def _process_frame_3d(pair, iou_threshold, distance_ranges, lateral_distance_ranges=None, save_detailed_matches=False, coord_system='camera', conf_threshold=0.0): """ Process a single frame for 3D evaluation (worker function). Args: pair: dict, image pair information (may contain roi_processor_config and min_box_size) iou_threshold: float, IoU threshold distance_ranges: list, distance ranges for 3D metrics lateral_distance_ranges: list, lateral distance ranges for 3D metrics save_detailed_matches: bool, whether to save detailed match information coord_system: str, coordinate system ('camera' or 'ego') conf_threshold: float, minimum confidence score for detections (default: 0.0, no filtering) Returns: tuple: (case_name, frame_name, 3d_samples, detailed_matches, frame_class_stats) """ import hashlib matcher = Matcher2D(iou_threshold=iou_threshold) lateral_axis = 0 if coord_system == 'camera' else 1 longitudinal_axis = 2 if coord_system == 'camera' else 0 gt_parser = GroundTruthParser(min_box_size=0 if 'multi_roi_eval_config' in pair else pair.get('min_box_size', 8), coord_system=coord_system) gts = Evaluator._parse_ground_truths_for_pair(pair, coord_system) dets = Evaluator._parse_detections_for_pair(pair, coord_system) if conf_threshold > 0.0: dets = [d for d in dets if d.get('confidence', 0.0) >= conf_threshold] # Collect 3D samples samples = [] detailed_matches = {} if save_detailed_matches else None frame_class_stats = {} for class_id in Metrics3D.CLASSES_3D: match_result = matcher.match(gts, dets, class_id) class_name = gt_parser.get_class_name(class_id) frame_class_stats[class_name] = Evaluator._build_class_count_stats(match_result, require_3d=True) if save_detailed_matches: detailed_matches[class_name] = [] for gt_idx, det_idx, iou in match_result['matches']: gt = match_result['gts_filtered'][gt_idx] det = match_result['dets_sorted'][det_idx] if gt['has_3d'] and det['3d_info'] is not None: samples.append((gt, det, class_id)) # Save detailed match information if save_detailed_matches: # Generate unique GT ID based on bbox coordinates gt_bbox_str = f"{gt['bbox_2d'][0]:.2f}_{gt['bbox_2d'][1]:.2f}_{gt['bbox_2d'][2]:.2f}_{gt['bbox_2d'][3]:.2f}" gt_id = hashlib.md5(gt_bbox_str.encode()).hexdigest()[:16] # Compute 3D errors # For vehicles, use the face center matching the detected face type # (same reference point as metrics_3d.add_sample uses) if class_id == 0: face_type = det['3d_info'].get('face_type', 'front') normalized_face = face_type.lower() if normalized_face in ('rear', 'tail'): normalized_face = 'back' faces = gt['3d_info'].get('faces') or {} face_data = faces.get(normalized_face) # face_data[7] is is_visible_from_camera; when 0, x3d/y3d/z3d # are sentinel -1.0 and must not be used as real coordinates. face_valid = ( face_data is not None and face_data[2] > 0 and not (len(face_data) >= 8 and face_data[7] == 0) ) gt_center = face_data[:3] if face_valid else gt['3d_info']['center'] else: gt_center = gt['3d_info']['center'] det_center = det['3d_info']['center'] lateral_error = abs(det_center[lateral_axis] - gt_center[lateral_axis]) longitudinal_error = abs(det_center[longitudinal_axis] - gt_center[longitudinal_axis]) # Use whole-object center z for relative error denominator and # distance routing, so that face centers with near-zero/negative z # (e.g. side faces at close range) don't produce astronomical # relative errors or get silently dropped from distance segments. gt_whole_longitudinal = gt['3d_info']['center'][longitudinal_axis] gt_depth = max(abs(gt_whole_longitudinal), 1e-6) longitudinal_relative_error = longitudinal_error / gt_depth # Heading error (using rotation field) gt_rotation = gt['3d_info'].get('rotation', 0) det_rotation = det['3d_info'].get('rotation', 0) heading_error = abs(normalize_angle(det_rotation - gt_rotation)) # Compute relaxed heading error if needed heading_error_relaxed = None is_reversal = False if heading_error > np.pi / 2: heading_error_relaxed = np.pi - heading_error is_reversal = True else: heading_error_relaxed = heading_error error_dict = { 'lateral': float(lateral_error), 'longitudinal': float(longitudinal_error), 'longitudinal_relative': float(longitudinal_relative_error), 'heading': float(heading_error), 'heading_relaxed': float(heading_error_relaxed), 'is_reversal': bool(is_reversal) } detailed_matches[class_name].append({ 'gt_id': gt_id, 'gt_bbox': gt['bbox_2d'], 'gt_center_3d': gt_center, 'gt_rotation': float(gt_rotation), 'det_bbox': det['bbox_2d'], 'det_center_3d': det_center, 'det_rotation': float(det_rotation), 'iou': float(iou), 'confidence': float(det.get('confidence', 0)), 'errors': error_dict, 'distance': { 'longitudinal': float(gt_whole_longitudinal), 'lateral': float(gt_center[lateral_axis]) } }) return (pair['case'], pair['frame'], samples, detailed_matches, frame_class_stats) def evaluate_2d(self): """ Evaluate 2D detection metrics with multiprocessing support. Returns: dict, 2D evaluation summary """ print("\n" + "="*50) print("Evaluating 2D Detection Metrics") print("="*50) # Group pairs by case (use unique case identifier for 2-level paths) cases = {} for pair in self.image_pairs: # Create unique case identifier: "level1/case" for 2-level, "case" for 1-level level1_name = pair.get('level1_name') case_name = pair['case'] if level1_name: case_key = f"{level1_name}/{case_name}" else: case_key = case_name if case_key not in cases: cases[case_key] = [] cases[case_key].append(pair) # Evaluate each case separately for case_idx, (case_key, case_pairs) in enumerate(cases.items(), 1): print(f"\nProcessing case [{case_idx}/{len(cases)}]: {case_key} ({len(case_pairs)} frames)") # Create metrics for this case case_metrics_2d = Metrics2D(num_classes=NUM_CLASSES, distance_ranges=self.distance_ranges_2d, lateral_roi=self.lateral_roi_2d, coord_system=self.coord_system) case_frame_stats = {} if self.num_workers > 1 and len(case_pairs) > 1: # Multiprocessing worker_func = partial( self._process_frame_2d, iou_threshold=self.iou_threshold ) with Pool(processes=self.num_workers) as pool: results = list(tqdm( pool.imap(worker_func, case_pairs), total=len(case_pairs), desc=f" {case_key}" )) # Aggregate results for case, frame, class_results in results: frame_stats = {'2d': {}, '3d': {}} for class_id, result_data in class_results.items(): match_result = result_data['match_result'] gts = result_data['gts'] dets = result_data['dets'] class_name = self.gt_parser.get_class_name(class_id) frame_stats['2d'][class_name] = self._build_class_count_stats(match_result, require_3d=False) case_metrics_2d.add_image_results(match_result, gts, dets, class_id) self.metrics_2d.add_image_results(match_result, gts, dets, class_id) case_frame_stats[frame] = frame_stats else: # Single process fallback for pair in tqdm(case_pairs, desc=f" {case_key}"): _, frame, class_results = self._process_frame_2d( pair, self.iou_threshold ) frame_stats = {'2d': {}, '3d': {}} for class_id, result_data in class_results.items(): match_result = result_data['match_result'] gts = result_data['gts'] dets = result_data['dets'] class_name = self.gt_parser.get_class_name(class_id) frame_stats['2d'][class_name] = self._build_class_count_stats(match_result, require_3d=False) case_metrics_2d.add_image_results(match_result, gts, dets, class_id) self.metrics_2d.add_image_results(match_result, gts, dets, class_id) case_frame_stats[frame] = frame_stats # Store case results (use case_key as identifier) self.per_case_metrics_2d[case_key] = case_metrics_2d.get_summary( conf_threshold=self.config.get('metrics_2d', {}).get('conf_threshold', 0.5), ap_method=self.config.get('metrics_2d', {}).get('ap_method', 'voc2010') ) existing = self.per_case_frame_stats.get(case_key, {}) for frame, stats in case_frame_stats.items(): existing.setdefault(frame, {'2d': {}, '3d': {}}) existing[frame]['2d'] = stats.get('2d', {}) self.per_case_frame_stats[case_key] = existing self.per_case_aggregate_stats[case_key] = self._aggregate_case_stats(existing) # Get overall summary summary = self.metrics_2d.get_summary( conf_threshold=self.config.get('metrics_2d', {}).get('conf_threshold', 0.5), ap_method=self.config.get('metrics_2d', {}).get('ap_method', 'voc2010') ) return summary def evaluate_3d(self): """ Evaluate 3D detection metrics with multiprocessing support. Returns: dict, 3D evaluation summary """ print("\n" + "="*50) print("Evaluating 3D Detection Metrics") print("="*50) # Group pairs by case (use unique case identifier for 2-level paths) cases = {} for pair in self.image_pairs: # Create unique case identifier: "level1/case" for 2-level, "case" for 1-level level1_name = pair.get('level1_name') case_name = pair['case'] if level1_name: case_key = f"{level1_name}/{case_name}" else: case_key = case_name if case_key not in cases: cases[case_key] = [] cases[case_key].append(pair) # Evaluate each case separately for case_idx, (case_key, case_pairs) in enumerate(cases.items(), 1): print(f"\nProcessing case [{case_idx}/{len(cases)}]: {case_key} ({len(case_pairs)} frames)") # Create metrics for this case case_metrics_3d = Metrics3D( distance_ranges=self.distance_ranges, lateral_distance_ranges=self.lateral_distance_ranges, heading_tolerance=self.heading_tolerance, coord_system=self.coord_system, vehicle_size_split=self.vehicle_size_split_config, ) case_frame_stats = self.per_case_frame_stats.get(case_key, {}) conf_threshold_3d = self.config.get('metrics_3d', {}).get( 'conf_threshold', self.config.get('metrics_2d', {}).get('conf_threshold', 0.0), ) if self.num_workers > 1 and len(case_pairs) > 1: # Multiprocessing worker_func = partial( self._process_frame_3d, iou_threshold=self.iou_threshold, distance_ranges=self.distance_ranges, lateral_distance_ranges=self.lateral_distance_ranges, save_detailed_matches=self.save_detailed_matches, coord_system=self.coord_system, conf_threshold=conf_threshold_3d, ) with Pool(processes=self.num_workers) as pool: results = list(tqdm( pool.imap(worker_func, case_pairs), total=len(case_pairs), desc=f" {case_key}" )) # Aggregate results if self.save_detailed_matches: self.detailed_3d_matches[case_key] = {} for case, frame, samples, detailed_matches, frame_class_stats in results: for gt, det, class_id in samples: case_metrics_3d.add_sample(gt, det, class_id) self.metrics_3d.add_sample(gt, det, class_id) case_frame_stats.setdefault(frame, {'2d': {}, '3d': {}}) case_frame_stats[frame]['3d'] = frame_class_stats # Save detailed matches if self.save_detailed_matches and detailed_matches: self.detailed_3d_matches[case_key][frame] = detailed_matches else: # Single process fallback if self.save_detailed_matches: self.detailed_3d_matches[case_key] = {} for pair in tqdm(case_pairs, desc=f" {case_key}"): _, frame, samples, detailed_matches, frame_class_stats = self._process_frame_3d( pair, self.iou_threshold, self.distance_ranges, self.lateral_distance_ranges, self.save_detailed_matches, self.coord_system, conf_threshold_3d, ) for gt, det, class_id in samples: case_metrics_3d.add_sample(gt, det, class_id) self.metrics_3d.add_sample(gt, det, class_id) case_frame_stats.setdefault(frame, {'2d': {}, '3d': {}}) case_frame_stats[frame]['3d'] = frame_class_stats # Save detailed matches if self.save_detailed_matches and detailed_matches: self.detailed_3d_matches[case_key][frame] = detailed_matches # Store case results (use case_key as identifier) self.per_case_metrics_3d[case_key] = case_metrics_3d.get_summary() self.per_case_frame_stats[case_key] = case_frame_stats self.per_case_aggregate_stats[case_key] = self._aggregate_case_stats(case_frame_stats) # Get overall summary summary = self.metrics_3d.get_summary() return summary def evaluate(self): """ Run complete evaluation (2D and 3D). Returns: dict with both 2d and 3d evaluation results """ results = {} # Evaluate 2D if self.config.get('eval_2d', True): results['2d_evaluation'] = self.evaluate_2d() # Evaluate 3D if self.config.get('eval_3d', True): results['3d_evaluation'] = self.evaluate_3d() return results def generate_report(self, results, output_dir): """ Generate evaluation report. Args: results: dict, evaluation results output_dir: str, output directory path """ os.makedirs(output_dir, exist_ok=True) # Add per-case results to the report if self.per_case_metrics_2d: results['per_case_2d'] = self.per_case_metrics_2d if self.per_case_metrics_3d: results['per_case_3d'] = self.per_case_metrics_3d # Add evaluation configuration to the report results['evaluation_config'] = { 'conf_threshold': self.config.get('metrics_2d', {}).get('conf_threshold', 0.5), 'iou_threshold': self.iou_threshold, 'ap_method': self.config.get('metrics_2d', {}).get('ap_method', 'voc2010'), 'vehicle_size_split_3d': { 'enabled': self.metrics_3d.vehicle_size_split_enabled, 'large_length_threshold': self.metrics_3d.vehicle_large_length_threshold, 'large_height_threshold': self.metrics_3d.vehicle_large_height_threshold, }, 'per_case_reports_dir': 'per_case_reports', 'main_json_excludes_detailed_case_stats': True, } # Save JSON report json_path = os.path.join(output_dir, 'evaluation_report.json') with open(json_path, 'w') as f: json.dump(results, f, indent=2) # Save detailed 3D matches if enabled if self.save_detailed_matches and self.detailed_3d_matches: matches_path = os.path.join(output_dir, 'detailed_3d_matches.json') with open(matches_path, 'w') as f: json.dump(self.detailed_3d_matches, f, indent=2) print(f"\nDetailed 3D matches saved to: {matches_path}") print(f"\nJSON report saved to: {json_path}") # Save text report txt_path = os.path.join(output_dir, 'evaluation_report.txt') with open(txt_path, 'w') as f: self._write_text_report(results, f) print(f"Text report saved to: {txt_path}") # Save per-case reports if self.per_case_metrics_2d or self.per_case_metrics_3d: self._write_per_case_reports(output_dir) # Print summary to console self._print_summary(results) def _write_text_report(self, results, file): """Write human-readable text report.""" file.write("=" * 80 + "\n") file.write("EVALUATION REPORT\n") file.write("=" * 80 + "\n\n") # Evaluation Configuration file.write("EVALUATION CONFIGURATION\n") file.write("-" * 80 + "\n") conf_threshold = self.config.get('metrics_2d', {}).get('conf_threshold', 0.5) file.write(f"Confidence Threshold (for P/R/F1): {conf_threshold:.3f}\n") file.write("Note: Precision, Recall, and F1 Score are calculated at the specified\n") file.write(" confidence threshold. AP (Average Precision) is computed from the\n") file.write(" complete Precision-Recall curve and is independent of this threshold.\n") file.write("\n") # 2D Results if '2d_evaluation' in results: file.write("2D DETECTION METRICS\n") file.write("-" * 80 + "\n") eval_2d = results['2d_evaluation'] # Per-class results file.write("\nPer-Class Metrics:\n") file.write(f"{'Class':<15} {'Precision':<12} {'Recall':<12} {'F1 Score':<12} {'AP':<12} {'TP':<8} {'FP':<8} {'FN':<8} {'GT':<8}\n") file.write("-" * 100 + "\n") for class_name, metrics in sorted(eval_2d['per_class'].items()): file.write(f"{class_name:<15} " f"{metrics['precision']:<12.4f} " f"{metrics['recall']:<12.4f} " f"{metrics['f1_score']:<12.4f} " f"{metrics['ap']:<12.4f} " f"{metrics['tp']:<8} " f"{metrics['fp']:<8} " f"{metrics['fn']:<8} " f"{metrics['num_gt']:<8}\n") # Overall results file.write("\nOverall Metrics:\n") overall = eval_2d['overall'] file.write(f" Precision: {overall['precision']:.4f}\n") file.write(f" Recall: {overall['recall']:.4f}\n") file.write(f" F1 Score: {overall['f1_score']:.4f}\n") file.write(f" mAP: {overall['map']:.4f}\n") file.write(f" Total TP: {overall['tp']}\n") file.write(f" Total FP: {overall['fp']}\n") file.write(f" Total FN: {overall['fn']}\n") file.write("\n") # Per-distance-range 2D metrics — helper def _write_dist_table_2d(dist_data, header_lines): for h in header_lines: file.write(h) all_range_keys = [] for cls_data in dist_data.values(): for rk in cls_data: if rk not in all_range_keys: all_range_keys.append(rk) for class_name_str in sorted(dist_data.keys()): cls_data = dist_data[class_name_str] file.write(f" {class_name_str.upper()}:\n") file.write(f" {'Range':<14} {'AP':<10} {'Precision':<12} {'Recall':<10} " f"{'F1':<10} {'TP':<6} {'FP':<6} {'FN':<6} {'GT':<6}\n") file.write(f" {'-'*80}\n") for rk in all_range_keys: if rk not in cls_data: continue m = cls_data[rk] file.write(f" {rk:<14} {m['ap']:<10.4f} {m['precision']:<12.4f} " f"{m['recall']:<10.4f} {m['f1_score']:<10.4f} " f"{m['tp']:<6} {m['fp']:<6} {m['fn']:<6} {m['num_gt']:<6}\n") file.write("\n") if 'per_class_by_distance' in eval_2d and eval_2d['per_class_by_distance']: _write_dist_table_2d( eval_2d['per_class_by_distance'], [ "\n全横向 + 纵向分段 2D Metrics (3D-capable classes only):\n", "Note: TPs are bucketed by matched GT z3d; FPs are bucketed by the\n", " detection's own predicted z3d. 2D-only classes are not shown.\n\n", ] ) if 'per_class_by_distance_lat_roi' in eval_2d and eval_2d['per_class_by_distance_lat_roi']: lat_roi = eval_2d.get('lateral_roi', self.lateral_roi_2d) lat_str = f"{lat_roi[0]}m ~ {lat_roi[1]}m" if lat_roi else "" _write_dist_table_2d( eval_2d['per_class_by_distance_lat_roi'], [ f"\n限定横向范围 [{lat_str}] + 纵向分段 2D Metrics (3D-capable classes only):\n", "Note: Objects outside the lateral ROI are excluded from all counts.\n\n", ] ) # 3D Results if '3d_evaluation' in results: file.write("\n3D DETECTION METRICS\n") file.write("-" * 80 + "\n") eval_3d = results['3d_evaluation'] for class_name, class_metrics in sorted(eval_3d.items()): file.write(f"\n{class_name.upper()}:\n") # Check if this is distance-range based or simple format if 'overall' in class_metrics: # Distance range based format for range_key, metrics in sorted(class_metrics.items()): if metrics['num_samples'] == 0 and range_key != 'overall': continue range_label = "OVERALL" if range_key == 'overall' else range_key file.write(f"\n [{range_label}]:\n") file.write(f" Samples: {metrics['num_samples']}\n") if metrics['num_samples'] > 0: file.write(f" Lateral Error (m):\n") file.write(f" Mean: {metrics['lateral_error']['mean']:.4f}\n") file.write(f" Median: {metrics['lateral_error']['median']:.4f}\n") file.write(f" Std: {metrics['lateral_error']['std']:.4f}\n") file.write(f" 90%: {metrics['lateral_error']['percentile_90']:.4f}\n") file.write(f" Longitudinal Error (m):\n") file.write(f" Mean: {metrics['longitudinal_error']['mean']:.4f}\n") file.write(f" Median: {metrics['longitudinal_error']['median']:.4f}\n") file.write(f" Std: {metrics['longitudinal_error']['std']:.4f}\n") file.write(f" 90%: {metrics['longitudinal_error']['percentile_90']:.4f}\n") file.write(f" Longitudinal Relative Error:\n") file.write(f" Mean: {metrics['longitudinal_relative_error']['mean']:.4f}\n") file.write(f" Median: {metrics['longitudinal_relative_error']['median']:.4f}\n") file.write(f" Std: {metrics['longitudinal_relative_error']['std']:.4f}\n") file.write(f" 90%: {metrics['longitudinal_relative_error']['percentile_90']:.4f}\n") file.write(f" Heading Error (rad):\n") file.write(f" Mean: {metrics['heading_error']['mean']:.4f}\n") file.write(f" Median: {metrics['heading_error']['median']:.4f}\n") file.write(f" Std: {metrics['heading_error']['std']:.4f}\n") file.write(f" 90%: {metrics['heading_error']['percentile_90']:.4f}\n") # Add relaxed heading error if available if 'heading_error_relaxed' in metrics: file.write(f" Heading Error Relaxed (rad) [180° symmetry]:\n") file.write(f" Mean: {metrics['heading_error_relaxed']['mean']:.4f}\n") file.write(f" Median: {metrics['heading_error_relaxed']['median']:.4f}\n") file.write(f" Std: {metrics['heading_error_relaxed']['std']:.4f}\n") file.write(f" 90%: {metrics['heading_error_relaxed']['percentile_90']:.4f}\n") file.write(f" Reversal Cases: {metrics['reversal_count']} ({metrics['reversal_percentage']:.1f}%)\n") else: # Legacy simple format (backward compatibility) metrics = class_metrics file.write(f" Samples: {metrics['num_samples']}\n") if metrics['num_samples'] > 0: file.write(f" Lateral Error (m):\n") file.write(f" Mean: {metrics['lateral_error']['mean']:.4f}\n") file.write(f" Median: {metrics['lateral_error']['median']:.4f}\n") file.write(f" Std: {metrics['lateral_error']['std']:.4f}\n") file.write(f" 90%: {metrics['lateral_error']['percentile_90']:.4f}\n") file.write(f" Longitudinal Error (m):\n") file.write(f" Mean: {metrics['longitudinal_error']['mean']:.4f}\n") file.write(f" Median: {metrics['longitudinal_error']['median']:.4f}\n") file.write(f" Std: {metrics['longitudinal_error']['std']:.4f}\n") file.write(f" 90%: {metrics['longitudinal_error']['percentile_90']:.4f}\n") file.write(f" Longitudinal Relative Error:\n") file.write(f" Mean: {metrics['longitudinal_relative_error']['mean']:.4f}\n") file.write(f" Median: {metrics['longitudinal_relative_error']['median']:.4f}\n") file.write(f" Std: {metrics['longitudinal_relative_error']['std']:.4f}\n") file.write(f" 90%: {metrics['longitudinal_relative_error']['percentile_90']:.4f}\n") file.write(f" Heading Error (rad):\n") file.write(f" Mean: {metrics['heading_error']['mean']:.4f}\n") file.write(f" Median: {metrics['heading_error']['median']:.4f}\n") file.write(f" Std: {metrics['heading_error']['std']:.4f}\n") file.write(f" 90%: {metrics['heading_error']['percentile_90']:.4f}\n") # Add relaxed heading error if available if 'heading_error_relaxed' in metrics: file.write(f" Heading Error Relaxed (rad) [180° symmetry]:\n") file.write(f" Mean: {metrics['heading_error_relaxed']['mean']:.4f}\n") file.write(f" Median: {metrics['heading_error_relaxed']['median']:.4f}\n") file.write(f" Std: {metrics['heading_error_relaxed']['std']:.4f}\n") file.write(f" 90%: {metrics['heading_error_relaxed']['percentile_90']:.4f}\n") file.write(f" Reversal Cases: {metrics['reversal_count']} ({metrics['reversal_percentage']:.1f}%)\n") def _write_per_case_reports(self, output_dir): """Write per-case evaluation reports.""" per_case_dir = os.path.join(output_dir, 'per_case_reports') os.makedirs(per_case_dir, exist_ok=True) # Get all case names case_names = set() if self.per_case_metrics_2d: case_names.update(self.per_case_metrics_2d.keys()) if self.per_case_metrics_3d: case_names.update(self.per_case_metrics_3d.keys()) for case_name in sorted(case_names): # Replace "/" with "_" in case_name for file system compatibility safe_case_name = case_name.replace('/', '_') case_report_path = os.path.join(per_case_dir, f'{safe_case_name}_report.txt') case_stats_json_path = os.path.join(per_case_dir, f'{safe_case_name}_frame_stats.json') case_stats_txt_path = os.path.join(per_case_dir, f'{safe_case_name}_frame_stats.txt') with open(case_report_path, 'w') as f: f.write("=" * 80 + "\n") f.write(f"EVALUATION REPORT - {case_name}\n") f.write("=" * 80 + "\n\n") # 2D metrics for this case if case_name in self.per_case_metrics_2d: eval_2d = self.per_case_metrics_2d[case_name] f.write("2D DETECTION METRICS\n") f.write("-" * 80 + "\n\n") overall = eval_2d['overall'] f.write(f"Overall Metrics:\n") f.write(f" Precision: {overall['precision']:.4f}\n") f.write(f" Recall: {overall['recall']:.4f}\n") f.write(f" mAP: {overall['map']:.4f}\n") f.write(f" Total TP: {overall['tp']}\n") f.write(f" Total FP: {overall['fp']}\n") f.write(f" Total FN: {overall['fn']}\n\n") f.write("Per-Class Metrics:\n") f.write(f"{'Class':<15} {'Precision':<12} {'Recall':<12} {'F1 Score':<12} {'AP':<12} {'TP':<8} {'FP':<8} {'FN':<8}\n") f.write("-" * 100 + "\n") for class_name_str, metrics in sorted(eval_2d['per_class'].items()): if metrics['num_gt'] > 0 or metrics['tp'] > 0 or metrics['fp'] > 0: f.write(f"{class_name_str:<15} " f"{metrics['precision']:<12.4f} " f"{metrics['recall']:<12.4f} " f"{metrics['f1_score']:<12.4f} " f"{metrics['ap']:<12.4f} " f"{metrics['tp']:<8} " f"{metrics['fp']:<8} " f"{metrics['fn']:<8}\n") # 3D metrics for this case if case_name in self.per_case_metrics_3d: eval_3d = self.per_case_metrics_3d[case_name] f.write("\n\n3D DETECTION METRICS\n") f.write("-" * 80 + "\n") for class_name_str, class_metrics in sorted(eval_3d.items()): # Check if this is distance-range based or simple format if 'overall' in class_metrics: overall_metrics = class_metrics['overall'] if overall_metrics['num_samples'] > 0: f.write(f"\n{class_name_str.upper()} [overall]:\n") f.write(f" Samples: {overall_metrics['num_samples']}\n") f.write(f" Lateral Error: {overall_metrics['lateral_error']['mean']:.4f}m (±{overall_metrics['lateral_error']['std']:.4f})\n") f.write(f" Longitudinal Error: {overall_metrics['longitudinal_error']['mean']:.4f}m (±{overall_metrics['longitudinal_error']['std']:.4f})\n") f.write(f" Longitudinal Relative Error: {overall_metrics['longitudinal_relative_error']['mean']:.4f} (±{overall_metrics['longitudinal_relative_error']['std']:.4f})\n") f.write(f" Heading Error: {overall_metrics['heading_error']['mean']:.4f}rad (±{overall_metrics['heading_error']['std']:.4f})\n") if 'heading_error_relaxed' in overall_metrics: f.write(f" Heading Error Relaxed: {overall_metrics['heading_error_relaxed']['mean']:.4f}rad (±{overall_metrics['heading_error_relaxed']['std']:.4f})\n") f.write(f" Reversal Cases: {overall_metrics['reversal_count']} ({overall_metrics['reversal_percentage']:.1f}%)\n") # Show distance ranges if any for range_key, metrics in sorted(class_metrics.items()): if range_key != 'overall' and metrics['num_samples'] > 0: f.write(f"\n [{range_key}]: n={metrics['num_samples']}\n") f.write(f" Lateral: {metrics['lateral_error']['mean']:.4f}m\n") f.write(f" Longitudinal: {metrics['longitudinal_error']['mean']:.4f}m\n") f.write(f" Longitudinal Relative Error: {metrics['longitudinal_relative_error']['mean']:.4f}\n") f.write(f" Heading: {metrics['heading_error']['mean']:.4f}rad\n") if 'heading_error_relaxed' in metrics: f.write(f" Heading Relaxed: {metrics['heading_error_relaxed']['mean']:.4f}rad\n") f.write(f" Reversals: {metrics['reversal_count']} ({metrics['reversal_percentage']:.1f}%)\n") else: # Legacy simple format if class_metrics['num_samples'] > 0: f.write(f"\n{class_name_str.upper()}:\n") f.write(f" Samples: {class_metrics['num_samples']}\n") f.write(f" Lateral Error: {class_metrics['lateral_error']['mean']:.4f}m (±{class_metrics['lateral_error']['std']:.4f})\n") f.write(f" Longitudinal Error: {class_metrics['longitudinal_error']['mean']:.4f}m (±{class_metrics['longitudinal_error']['std']:.4f})\n") f.write(f" Longitudinal Relative Error: {class_metrics['longitudinal_relative_error']['mean']:.4f} (±{class_metrics['longitudinal_relative_error']['std']:.4f})\n") f.write(f" Heading Error: {class_metrics['heading_error']['mean']:.4f}rad (±{class_metrics['heading_error']['std']:.4f})\n") if 'heading_error_relaxed' in class_metrics: f.write(f" Heading Error Relaxed: {class_metrics['heading_error_relaxed']['mean']:.4f}rad (±{class_metrics['heading_error_relaxed']['std']:.4f})\n") f.write(f" Reversal Cases: {class_metrics['reversal_count']} ({class_metrics['reversal_percentage']:.1f}%)\n") case_frame_stats = self.per_case_frame_stats.get(case_name, {}) case_aggregate_stats = self.per_case_aggregate_stats.get(case_name, {'2d': {}, '3d': {}}) with open(case_stats_json_path, 'w') as f: json.dump({ 'case_name': case_name, 'frame_stats': case_frame_stats, 'aggregate_stats': case_aggregate_stats, }, f, indent=2) with open(case_stats_txt_path, 'w') as f: f.write("=" * 80 + "\n") f.write(f"FRAME-LEVEL EVALUATION STATS - {case_name}\n") f.write("=" * 80 + "\n\n") f.write("CASE AGGREGATE STATS\n") f.write("-" * 80 + "\n") for eval_type in ('2d', '3d'): f.write(f"\n[{eval_type.upper()}]\n") f.write(f"{'Class':<15} {'GT':<10} {'Det':<10} {'Match':<10} {'Frames':<10}\n") f.write("-" * 60 + "\n") for class_name_str, stats in sorted(case_aggregate_stats.get(eval_type, {}).items()): f.write( f"{class_name_str:<15} {stats['gt_count']:<10} {stats['det_count']:<10} " f"{stats['match_count']:<10} {stats['num_frames']:<10}\n" ) if stats.get('gt_ids'): f.write(f" GT IDs: {stats['gt_ids']}\n") if stats.get('det_ids'): f.write(f" Det IDs: {stats['det_ids']}\n") if stats.get('matched_pairs'): f.write(f" Matched Pairs: {stats['matched_pairs']}\n") f.write("\n\nFRAME STATS\n") f.write("-" * 80 + "\n") for frame_name in sorted(case_frame_stats.keys()): f.write(f"\n[{frame_name}]\n") for eval_type in ('2d', '3d'): f.write(f" {eval_type.upper()}:\n") f.write(f" {'Class':<13} {'GT':<8} {'Det':<8} {'Match':<8}\n") f.write(f" {'-' * 45}\n") for class_name_str, stats in sorted(case_frame_stats[frame_name].get(eval_type, {}).items()): f.write( f" {class_name_str:<13} {stats['gt_count']:<8} {stats['det_count']:<8} " f"{stats['match_count']:<8}\n" ) if stats.get('gt_ids'): f.write(f" GT IDs: {stats['gt_ids']}\n") if stats.get('det_ids'): f.write(f" Det IDs: {stats['det_ids']}\n") if stats.get('matched_pairs'): f.write(f" Matched Pairs: {stats['matched_pairs']}\n") if stats.get('unmatched_gt_ids'): f.write(f" Unmatched GT IDs: {stats['unmatched_gt_ids']}\n") if stats.get('unmatched_det_ids'): f.write(f" Unmatched Det IDs: {stats['unmatched_det_ids']}\n") print(f"Per-case reports saved to: {per_case_dir}/ ({len(case_names)} cases)") def _print_summary(self, results): """Print summary to console.""" print("\n" + "=" * 80) print("EVALUATION SUMMARY - OVERALL") print("=" * 80) # Print configuration conf_threshold = self.config.get('metrics_2d', {}).get('conf_threshold', 0.5) print(f"\nConfiguration:") print(f" Confidence Threshold (for P/R/F1): {conf_threshold:.3f}") print(f" Note: AP is independent of confidence threshold") if '2d_evaluation' in results: overall = results['2d_evaluation']['overall'] print(f"\n2D Metrics:") print(f" Precision: {overall['precision']:.4f}") print(f" Recall: {overall['recall']:.4f}") print(f" F1 Score: {overall['f1_score']:.4f}") print(f" mAP: {overall['map']:.4f}") # Per-distance-range 2D summary (console) def _print_dist_table_2d(dist_data, title): print(f"\n {title}") for class_name_str in sorted(dist_data.keys()): cls_data = dist_data[class_name_str] parts = [] for rk, m in cls_data.items(): if m['num_gt'] > 0: parts.append(f"{rk}: R={m['recall']:.3f} AP={m['ap']:.3f} (n={m['num_gt']})") if parts: print(f" {class_name_str}: " + " | ".join(parts)) dist_data = results['2d_evaluation'].get('per_class_by_distance', {}) if dist_data: _print_dist_table_2d(dist_data, "2D 全横向 + 纵向分段 (Recall | AP):") dist_data_lat = results['2d_evaluation'].get('per_class_by_distance_lat_roi', {}) if dist_data_lat: lat_roi = results['2d_evaluation'].get('lateral_roi', self.lateral_roi_2d) lat_str = f"{lat_roi[0]}m ~ {lat_roi[1]}m" if lat_roi else "" _print_dist_table_2d(dist_data_lat, f"2D 限定横向 [{lat_str}] + 纵向分段 (Recall | AP):") if '3d_evaluation' in results: print(f"\n3D Metrics:") for class_name, class_metrics in sorted(results['3d_evaluation'].items()): # Check if this is distance-range based or simple format if 'overall' in class_metrics: # Distance range based: show overall first overall_metrics = class_metrics['overall'] if overall_metrics['num_samples'] > 0: heading_str = f"Head={overall_metrics['heading_error']['mean']:.3f}rad" if 'heading_error_relaxed' in overall_metrics: heading_str += f" (relaxed={overall_metrics['heading_error_relaxed']['mean']:.3f}rad, rev={overall_metrics['reversal_count']})" print(f" {class_name} [overall]: " f"Lat={overall_metrics['lateral_error']['mean']:.3f}m, " f"Long={overall_metrics['longitudinal_error']['mean']:.3f}m, " f"LongRel={overall_metrics['longitudinal_relative_error']['mean']:.3f}, " f"{heading_str} " f"(n={overall_metrics['num_samples']})") # Show distance ranges for range_key, metrics in sorted(class_metrics.items()): if range_key != 'overall' and metrics['num_samples'] > 0: print(f" [{range_key}]: " f"Lat={metrics['lateral_error']['mean']:.3f}m, " f"Long={metrics['longitudinal_error']['mean']:.3f}m, " f"LongRel={metrics['longitudinal_relative_error']['mean']:.3f}, " f"Head={metrics['heading_error']['mean']:.3f}rad " f"(n={metrics['num_samples']})") else: # Legacy simple format if class_metrics['num_samples'] > 0: heading_str = f"Head={class_metrics['heading_error']['mean']:.3f}rad" if 'heading_error_relaxed' in class_metrics: heading_str += f" (relaxed={class_metrics['heading_error_relaxed']['mean']:.3f}rad, rev={class_metrics['reversal_count']})" print(f" {class_name}: " f"Lat={class_metrics['lateral_error']['mean']:.3f}m, " f"Long={class_metrics['longitudinal_error']['mean']:.3f}m, " f"LongRel={class_metrics['longitudinal_relative_error']['mean']:.3f}, " f"{heading_str} " f"(n={class_metrics['num_samples']})") # Print per-case summary if available if 'per_case_2d' in results and results['per_case_2d']: print("\n" + "=" * 80) print("PER-CASE SUMMARY") print("=" * 80) for case_name in sorted(results['per_case_2d'].keys()): case_2d = results['per_case_2d'].get(case_name, {}) case_3d = results.get('per_case_3d', {}).get(case_name, {}) print(f"\n[{case_name}]") if case_2d and 'overall' in case_2d: overall = case_2d['overall'] print(f" 2D: P={overall['precision']:.3f}, R={overall['recall']:.3f}, F1={overall['f1_score']:.3f}, mAP={overall['map']:.3f}") if case_3d: # Show 3D metrics for vehicle class as representative for class_name in ['vehicle', 'vehicle_large', 'vehicle_small', 'bus', 'truck', 'tanker', 'unknown', 'pedestrian', 'bicycle', 'motorcyclist', 'tricycle']: if class_name in case_3d: class_metrics = case_3d[class_name] if 'overall' in class_metrics: metrics = class_metrics['overall'] else: metrics = class_metrics if metrics['num_samples'] > 0: print(f" 3D {class_name}: " f"Lat={metrics['lateral_error']['mean']:.3f}m, " f"Long={metrics['longitudinal_error']['mean']:.3f}m " f"(n={metrics['num_samples']})") print("=" * 80 + "\n")