""" 종합 평가 엔진 모든 평가 지표를 통합하여 종합 점수를 계산하고 성능 등급을 결정합니다. """ import numpy as np import os import logging from typing import Dict, Tuple, Optional, List from datetime import datetime import json from .data_loader import DataLoader from .renderer import Renderer from .metrics.map_2d import Map2DCalculator from .metrics.map_3d import Map3DCalculator from .metrics.chamfer_distance import ChamferDistanceCalculator from .metrics.emd import EMCalculator from .metrics.class_accuracy import ClassAccuracyCalculator from .utils.reference_extractor import ReferenceDataExtractor from .utils.performance_monitor import get_memory_profiler from .utils.logging_utils import get_logger from config.evaluation_config import EVALUATION_CONFIG def normalize_score_improved(metric_name: str, metric_value: float, threshold: float) -> float: """ 개선된 점수 정규화 함수 (0-100 범위, 더 엄격한 기준 적용) Args: metric_name (str): 지표 이름 metric_value (float): 원본 지표 값 threshold (float): 임계값 Returns: float: 정규화된 점수 (0-100) """ import math # 무한대/NaN 값 처리 if math.isinf(metric_value) or math.isnan(metric_value): return 0.0 if metric_name in ['2d_map', '3d_map', 'class_accuracy']: # 높을수록 좋은 지표 (더 엄격한 정규화) ratio = metric_value / threshold if ratio >= 1.0: # 임계값 이상: 80-100점 (우수) normalized = 80.0 + min(20.0, (ratio - 1.0) * 20.0) elif ratio >= 0.8: # 임계값의 80% 이상: 60-80점 (양호) normalized = 60.0 + ((ratio - 0.8) * 100.0) elif ratio >= 0.5: # 임계값의 50% 이상: 30-60점 (보통) normalized = 30.0 + ((ratio - 0.5) * 100.0) else: # 임계값의 50% 미만: 0-30점 (부족) normalized = ratio * 60.0 return max(0.0, min(100.0, normalized)) elif metric_name in ['chamfer_distance', 'emd']: # 낮을수록 좋은 지표 (거리 기반, 완화된 정규화) if metric_value <= 0: return 100.0 # 완화된 임계값 검증 (임계값의 10배 이상이면 0점) if metric_value > threshold * 10: return 0.0 # 완화된 정규화 적용 ratio = metric_value / threshold if ratio <= 0.5: # 임계값의 절반 이하: 90-100점 (우수) normalized = 100.0 - (ratio * 20.0) elif ratio <= 1.0: # 임계값 이하: 70-90점 (양호) normalized = 90.0 - ((ratio - 0.5) * 40.0) elif ratio <= 2.0: # 임계값의 2배 이하: 40-70점 (보통) normalized = 70.0 - ((ratio - 1.0) * 30.0) else: # 임계값의 2배 초과: 0-40점 (완만한 지수적 감소) normalized = 40.0 * math.exp(-(ratio - 2.0) * 1.0) return max(0.0, min(100.0, normalized)) else: # 기본 정규화 return min(100.0, max(0.0, metric_value * 100.0)) def validate_score_normalization(metric_name: str, original_value: float, normalized_score: float) -> bool: """ 점수 정규화 결과의 유효성을 검증합니다. Args: metric_name (str): 지표 이름 original_value (float): 원본 값 normalized_score (float): 정규화된 점수 Returns: bool: 유효성 여부 """ # 기본 검증 if np.isnan(normalized_score) or np.isinf(normalized_score): return False if normalized_score < 0 or normalized_score > 100: return False # 지표별 특수 검증 if metric_name in ['2d_map', '3d_map', 'class_accuracy']: # 높을수록 좋은 지표 if original_value < 0 or original_value > 1: return False elif metric_name in ['chamfer_distance', 'emd']: # 낮을수록 좋은 지표 if original_value < 0: return False return True class Evaluator: """종합 평가를 담당하는 클래스""" def __init__(self, config: Dict = None): """ 평가 엔진 초기화 Args: config (Dict): 평가 설정 (None이면 기본 설정 사용) """ self.config = config if config else EVALUATION_CONFIG # 로깅 설정 self._setup_logging() # 컴포넌트 초기화 self.data_loader = DataLoader() self.renderer = Renderer( image_size=self.config['rendering']['image_size'] ) # 평가 지표 계산기 초기화 self.map_2d_calculator = Map2DCalculator() self.map_3d_calculator = Map3DCalculator() self.chamfer_calculator = ChamferDistanceCalculator() self.emd_calculator = EMCalculator() self.class_accuracy_calculator = ClassAccuracyCalculator() # 참조 데이터 추출기 초기화 self.reference_extractor = ReferenceDataExtractor() # 성능 모니터링 및 로깅 초기화 self.memory_profiler = get_memory_profiler() self.detailed_logger = get_logger() # 결과 저장용 디렉토리 생성 self._create_output_directories() self.logger.info("평가 엔진이 초기화되었습니다.") def evaluate_model(self, model_path: str, reference_path: str) -> Dict: """ 3D 모델을 종합적으로 평가합니다. Args: model_path (str): 변환된 3D 모델 파일 경로 reference_path (str): 참조 이미지 파일 경로 Returns: Dict: 평가 결과 """ try: # 입력 파일 검증 if not self._validate_input_files(model_path, reference_path): raise ValueError("입력 파일 검증 실패") # 데이터 로드 model_3d = self.data_loader.load_glb_model(model_path) reference_image = self.data_loader.load_reference_image(reference_path) # Ground Truth 생성 ground_truth_2d = self.data_loader.create_ground_truth(reference_image) ground_truth_3d = self._create_3d_ground_truth(reference_path, model_path) # 렌더링 수행 rendered_images = self.renderer.render_multiple_views( model_3d, self.config['rendering']['num_views_for_evaluation'] ) # 각 평가 지표 계산 metrics = self._calculate_all_metrics( model_3d, ground_truth_3d, rendered_images, reference_image, ground_truth_2d ) # 종합 점수 계산 comprehensive_score, score_details = self.calculate_comprehensive_score(metrics) # 성능 등급 결정 grade = self.determine_grade(comprehensive_score) # 결과 정리 results = { 'model_path': model_path, 'reference_path': reference_path, 'evaluation_timestamp': datetime.now().isoformat(), 'comprehensive_score': comprehensive_score, 'grade': grade, 'metrics': metrics, 'score_details': score_details, 'config_used': self.config } # 결과 저장 if self.config['output']['generate_report']: self._save_results(results) self.logger.info(f"평가 완료 - 종합 점수: {comprehensive_score:.2f}, 등급: {grade}") return results except Exception as e: self.logger.error(f"평가 중 오류 발생: {str(e)}") raise def calculate_comprehensive_score(self, metrics: Dict) -> Tuple[float, Dict]: """ 가중치 기반 종합 점수를 계산합니다. Args: metrics (Dict): 개별 평가 지표 결과 Returns: Tuple[float, Dict]: (종합 점수, 점수 세부사항) """ weights = self.config['weights'] thresholds = self.config['thresholds'] score_details = {} weighted_sum = 0.0 total_weight = 0.0 for metric_name, metric_value in metrics.items(): if metric_name in weights: weight = weights[metric_name] threshold = thresholds[metric_name] # 개선된 점수 정규화 (0-100 범위) normalized_score = normalize_score_improved(metric_name, metric_value, threshold) # 정규화 결과 검증 if not validate_score_normalization(metric_name, metric_value, normalized_score): self.detailed_logger.main_logger.warning( f"{metric_name} 정규화 결과가 유효하지 않습니다: " f"원본={metric_value}, 정규화={normalized_score}" ) # 가중치 적용 weighted_score = normalized_score * weight weighted_sum += weighted_score total_weight += weight score_details[metric_name] = { 'raw_value': metric_value, 'normalized_score': normalized_score, 'weight': weight, 'weighted_score': weighted_score, 'threshold': threshold } # 최종 종합 점수 comprehensive_score = weighted_sum / total_weight if total_weight > 0 else 0.0 return comprehensive_score, score_details def determine_grade(self, score: float) -> str: """ 점수에 따라 성능 등급을 결정합니다. Args: score (float): 종합 점수 (0-100) Returns: str: 성능 등급 (A, B, C, D, F) """ grade_thresholds = self.config['grade_thresholds'] if score >= grade_thresholds['A']: return 'A' elif score >= grade_thresholds['B']: return 'B' elif score >= grade_thresholds['C']: return 'C' elif score >= grade_thresholds['D']: return 'D' else: return 'F' def generate_report(self, results: Dict) -> str: """ 평가 결과 리포트를 생성합니다. Args: results (Dict): 평가 결과 Returns: str: 생성된 리포트 내용 """ report_format = self.config['output']['report_format'] if report_format == 'html': return self._generate_html_report(results) elif report_format == 'json': return self._generate_json_report(results) else: return self._generate_text_report(results) def _calculate_all_metrics(self, model_3d: Dict, ground_truth_3d: Dict, rendered_images: List[np.ndarray], reference_image: np.ndarray, ground_truth_2d: Dict) -> Dict: """ 모든 평가 지표를 계산합니다 (강화된 오류 처리). Args: model_3d (Dict): 3D 모델 정보 ground_truth_3d (Dict): 3D Ground Truth rendered_images (List[np.ndarray]): 렌더링된 이미지들 reference_image (np.ndarray): 참조 이미지 ground_truth_2d (Dict): 2D Ground Truth Returns: Dict: 모든 평가 지표 결과 """ metrics = {} # 각 지표별 개별 오류 처리 metric_calculators = { '2d_map': { 'calculator': self.map_2d_calculator, 'method': 'calculate_2d_map', 'args': (rendered_images, reference_image), 'default_value': 0.0, 'description': '2D mAP 계산' }, '3d_map': { 'calculator': self.map_3d_calculator, 'method': 'calculate_3d_map', 'args': (model_3d, ground_truth_3d), 'default_value': 0.0, 'description': '3D mAP 계산' }, 'chamfer_distance': { 'calculator': self.chamfer_calculator, 'method': 'calculate_chamfer_distance', 'args': (model_3d, ground_truth_3d), 'default_value': float('inf'), 'description': 'Chamfer Distance 계산' }, 'emd': { 'calculator': self.emd_calculator, 'method': 'calculate_emd', 'args': (model_3d, ground_truth_3d), 'default_value': float('inf'), 'description': 'EMD 계산' }, 'class_accuracy': { 'calculator': self.class_accuracy_calculator, 'method': 'calculate_class_accuracy', 'args': (model_3d, ground_truth_3d), 'default_value': 0.0, 'description': '클래스 정확도 계산' } } # 각 지표를 개별적으로 계산 (오류 격리) for metric_name, config in metric_calculators.items(): try: # 메모리 프로파일링 시작 self.memory_profiler.take_snapshot(f"{metric_name}_start") # 상세 로깅 시작 self.detailed_logger.log_metric_start( metric_name, model_3d_size=len(model_3d.get('vertices', [])), ground_truth_3d_size=len(ground_truth_3d.get('vertices', [])), rendered_images_count=len(rendered_images) if rendered_images else 0 ) # 지표 계산 calculator = config['calculator'] method = getattr(calculator, config['method']) result = method(*config['args']) # 결과 검증 if result is None or (isinstance(result, float) and np.isnan(result)): raise ValueError(f"{metric_name} 계산 결과가 유효하지 않습니다: {result}") metrics[metric_name] = result # 메모리 프로파일링 종료 self.memory_profiler.take_snapshot(f"{metric_name}_end") # 상세 로깅 완료 self.detailed_logger.log_metric_result(metric_name, result) self.logger.info(f"{config['description']} 완료: {result}") except MemoryError as e: # 메모리 부족 오류 처리 error_msg = f"{config['description']} 중 메모리 부족: {str(e)}" self.logger.error(error_msg) self.detailed_logger.log_metric_error(metric_name, e, { 'error_type': 'MemoryError', 'description': config['description'] }) # 메모리 부족 시 기본값 사용 metrics[metric_name] = config['default_value'] except Exception as e: # 기타 오류 처리 error_msg = f"{config['description']} 중 오류 발생: {str(e)}" self.logger.error(error_msg) self.detailed_logger.log_metric_error(metric_name, e, { 'error_type': type(e).__name__, 'description': config['description'], 'args_info': { 'model_3d_keys': list(model_3d.keys()) if model_3d else [], 'ground_truth_3d_keys': list(ground_truth_3d.keys()) if ground_truth_3d else [], 'rendered_images_count': len(rendered_images) if rendered_images else 0 } }) # 오류 발생 시 기본값 사용 metrics[metric_name] = config['default_value'] # 전체 메트릭 계산 결과 로깅 self.detailed_logger.main_logger.info("=== 모든 지표 계산 완료 ===") for metric_name, value in metrics.items(): self.detailed_logger.main_logger.info(f"{metric_name}: {value}") return metrics def _create_3d_ground_truth(self, reference_image_path: str, model_3d_path: str) -> Dict: """ 참조 이미지에서 올바른 Ground Truth 생성 (표준화된 구조) Args: reference_image_path: 참조 이미지 경로 model_3d_path: 3D 모델 경로 (Ground Truth가 아님) Returns: 표준화된 Ground Truth 데이터 """ try: # 표준화된 Ground Truth 생성 ground_truth = self.reference_extractor.create_unified_ground_truth( reference_image_path, model_3d_path ) # 모델과 Ground Truth의 독립성 검증 self._validate_ground_truth_independence(ground_truth, model_3d_path) # vertices 키 존재 여부 확인 및 검증 if 'vertices' not in ground_truth: self.logger.warning("Ground Truth에 vertices 키가 없습니다. 재구성 중...") ground_truth = self._reconstruct_ground_truth_vertices(ground_truth, reference_image_path) # 메시 구조 일관성 확인 self._validate_mesh_consistency(ground_truth) self.logger.info(f"표준화된 Ground Truth 생성 완료: {len(ground_truth.get('vertices', []))}개 vertices") return ground_truth except Exception as e: self.logger.error(f"Error creating ground truth: {str(e)}") # 오류 발생 시 대체 Ground Truth 생성 return self._create_fallback_ground_truth(reference_image_path) def _extract_reference_data(self, reference_image_path: str) -> Dict: """참조 이미지에서 실제 객체 정보 추출""" return self.reference_extractor.extract_reference_data(reference_image_path) def _validate_ground_truth_independence(self, ground_truth: Dict, model_3d_path: str) -> bool: """모델과 Ground Truth의 독립성 검증""" # Ground Truth가 모델과 다른지 확인 if not ground_truth.get('objects'): self.logger.warning("Ground Truth에 객체가 없습니다") return False # 모델 파일 경로와 Ground Truth가 다른지 확인 self.logger.info(f"Ground Truth 독립성 검증: {len(ground_truth['objects'])}개 객체 발견") return True def _reconstruct_ground_truth_vertices(self, ground_truth: Dict, reference_image_path: str) -> Dict: """Ground Truth vertices 재구성""" try: self.logger.info("Ground Truth vertices 재구성 중...") # 바운딩 박스가 있는 경우 vertices 재구성 if ground_truth.get('bounding_boxes'): vertices = [] for bbox in ground_truth['bounding_boxes']: x1, y1, x2, y2 = bbox z_min, z_max = 0.0, 1.0 # 8개 꼭짓점 생성 (정육면체) cube_vertices = [ [x1, y1, z_min], [x2, y1, z_min], [x2, y2, z_min], [x1, y2, z_min], # 하단 [x1, y1, z_max], [x2, y1, z_max], [x2, y2, z_max], [x1, y2, z_max] # 상단 ] vertices.extend(cube_vertices) ground_truth['vertices'] = vertices # faces도 재구성 if 'faces' not in ground_truth: ground_truth['faces'] = self._generate_faces_from_vertices(vertices) self.logger.info(f"vertices 재구성 완료: {len(vertices)}개 vertices") else: # 바운딩 박스가 없는 경우 기본 정육면체 생성 ground_truth['vertices'] = [ [0, 0, 0], [1, 0, 0], [1, 1, 0], [0, 1, 0], # 하단 [0, 0, 1], [1, 0, 1], [1, 1, 1], [0, 1, 1] # 상단 ] ground_truth['faces'] = [ [0, 1, 2], [0, 2, 3], # 하단 [4, 7, 6], [4, 6, 5], # 상단 [0, 4, 5], [0, 5, 1], # 앞면 [2, 6, 7], [2, 7, 3], # 뒷면 [0, 3, 7], [0, 7, 4], # 왼쪽 [1, 5, 6], [1, 6, 2] # 오른쪽 ] self.logger.info("기본 정육면체 vertices 생성 완료") return ground_truth except Exception as e: self.logger.error(f"vertices 재구성 중 오류 발생: {str(e)}") # 오류 발생 시 기본 구조 반환 return self._create_fallback_ground_truth(reference_image_path) def _generate_faces_from_vertices(self, vertices: List[List[float]]) -> List[List[int]]: """vertices에서 faces 생성""" faces = [] num_objects = len(vertices) // 8 # 각 객체마다 8개 vertices for i in range(num_objects): base_idx = i * 8 # 정육면체의 12개 면 (각 면은 2개 삼각형으로 구성) cube_faces = [ # 하단 면 [base_idx, base_idx+1, base_idx+2], [base_idx, base_idx+2, base_idx+3], # 상단 면 [base_idx+4, base_idx+7, base_idx+6], [base_idx+4, base_idx+6, base_idx+5], # 앞면 [base_idx, base_idx+4, base_idx+5], [base_idx, base_idx+5, base_idx+1], # 뒷면 [base_idx+2, base_idx+6, base_idx+7], [base_idx+2, base_idx+7, base_idx+3], # 왼쪽 면 [base_idx, base_idx+3, base_idx+7], [base_idx, base_idx+7, base_idx+4], # 오른쪽 면 [base_idx+1, base_idx+5, base_idx+6], [base_idx+1, base_idx+6, base_idx+2] ] faces.extend(cube_faces) return faces def _validate_mesh_consistency(self, ground_truth: Dict) -> bool: """메시 구조 일관성 확인""" try: vertices = ground_truth.get('vertices', []) faces = ground_truth.get('faces', []) if not vertices or not faces: self.logger.warning("메시 데이터가 비어있습니다") return False # vertices 개수 확인 if len(vertices) == 0: self.logger.error("vertices가 비어있습니다") return False # faces 인덱스 범위 확인 max_vertex_idx = len(vertices) - 1 for i, face in enumerate(faces): if not isinstance(face, list) or len(face) != 3: self.logger.error(f"faces[{i}]가 유효하지 않습니다: {face}") return False for vertex_idx in face: if not isinstance(vertex_idx, int) or vertex_idx < 0 or vertex_idx > max_vertex_idx: self.logger.error(f"faces[{i}]에 유효하지 않은 vertex 인덱스: {vertex_idx}") return False self.logger.info(f"메시 일관성 검증 완료: {len(vertices)}개 vertices, {len(faces)}개 faces") return True except Exception as e: self.logger.error(f"메시 일관성 검증 중 오류 발생: {str(e)}") return False def _create_fallback_ground_truth(self, reference_image_path: str) -> Dict: """오류 발생 시 대체 Ground Truth 생성""" self.logger.warning("대체 Ground Truth 생성 중...") return { 'vertices': [ [0, 0, 0], [1, 0, 0], [1, 1, 0], [0, 1, 0], # 하단 [0, 0, 1], [1, 0, 1], [1, 1, 1], [0, 1, 1] # 상단 ], 'faces': [ [0, 1, 2], [0, 2, 3], # 하단 [4, 7, 6], [4, 6, 5], # 상단 [0, 4, 5], [0, 5, 1], # 앞면 [2, 6, 7], [2, 7, 3], # 뒷면 [0, 3, 7], [0, 7, 4], # 왼쪽 [1, 5, 6], [1, 6, 2] # 오른쪽 ], 'center': [0.5, 0.5, 0.5], 'scale': [1.0, 1.0, 1.0], 'objects': [], 'bounding_boxes': [], 'class_labels': [], 'confidence_scores': [], 'metadata': { 'source_image': reference_image_path, 'extraction_method': 'fallback', 'timestamp': self._get_timestamp(), 'version': '1.0', 'error': 'fallback_created' } } def _get_timestamp(self) -> str: """현재 시간 스탬프 반환""" from datetime import datetime return datetime.now().isoformat() def _validate_input_files(self, model_path: str, reference_path: str) -> bool: """ 입력 파일의 유효성을 검증합니다. Args: model_path (str): 3D 모델 파일 경로 reference_path (str): 참조 이미지 파일 경로 Returns: bool: 파일 유효성 """ validation_config = self.config['validation'] if not validation_config['validate_input_files']: return True # 파일 존재 확인 if not os.path.exists(model_path): self.logger.error(f"3D 모델 파일이 존재하지 않습니다: {model_path}") return False if not os.path.exists(reference_path): self.logger.error(f"참조 이미지 파일이 존재하지 않습니다: {reference_path}") return False # 파일 형식 확인 if validation_config['check_file_formats']: if not self.data_loader.validate_file(model_path, '3d'): self.logger.error(f"지원하지 않는 3D 파일 형식: {model_path}") return False if not self.data_loader.validate_file(reference_path, 'image'): self.logger.error(f"지원하지 않는 이미지 파일 형식: {reference_path}") return False # 파일 크기 확인 min_size = validation_config['min_file_size_kb'] * 1024 max_size = validation_config['max_file_size_mb'] * 1024 * 1024 for file_path in [model_path, reference_path]: file_size = os.path.getsize(file_path) if file_size < min_size: self.logger.error(f"파일 크기가 너무 작습니다: {file_path}") return False if file_size > max_size: self.logger.error(f"파일 크기가 너무 큽니다: {file_path}") return False return True def _setup_logging(self): """로깅을 설정합니다.""" logging_config = self.config['logging'] # 로그 디렉토리 생성 log_file = logging_config['file'] log_dir = os.path.dirname(log_file) if log_dir and not os.path.exists(log_dir): os.makedirs(log_dir) # 로깅 설정 logging.basicConfig( level=getattr(logging, logging_config['level']), format=logging_config['format'], handlers=[ logging.FileHandler(log_file), logging.StreamHandler() ] ) self.logger = logging.getLogger(__name__) def _create_output_directories(self): """출력 디렉토리들을 생성합니다.""" file_paths = self.config['file_paths'] for dir_path in file_paths.values(): if not os.path.exists(dir_path): os.makedirs(dir_path) def _save_results(self, results: Dict): """평가 결과를 저장합니다.""" timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") # JSON 결과 저장 json_file = os.path.join( self.config['file_paths']['reports_dir'], f"evaluation_results_{timestamp}.json" ) with open(json_file, 'w', encoding='utf-8') as f: json.dump(results, f, indent=2, ensure_ascii=False) self.logger.info(f"결과가 저장되었습니다: {json_file}") def _generate_html_report(self, results: Dict) -> str: """HTML 형식의 리포트를 생성합니다.""" html_content = f"""
평가 시간: {results['evaluation_timestamp']}
모델 파일: {results['model_path']}
참조 파일: {results['reference_path']}