1764 lines
88 KiB
Python
Executable File
1764 lines
88 KiB
Python
Executable File
"""
|
|
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")
|