evaluation.py

import os
import pandas as pd
import numpy as np
from abc import ABC, abstractmethod
from typing import List, Dict, Optional, Tuple, Set
from dataclasses import dataclass
from collections import defaultdict
from loader import TSPInstance, TSPInstanceFactory
from visualization import save_tour_gif

class TSPSolver(ABC):
    """TSP求解器抽象基类"""
    
    def __init__(self, name: str):
        self.name = name
    
    @abstractmethod
    def solve(self, instance: TSPInstance, time_limit: float = 60.0) -> Tuple[List[int], float]:
        """
        求解TSP
        返回: (路径, 运行时间)
        """
        pass

class Metric(ABC):
    """评估指标抽象基类"""
    
    def __init__(self, name: str):
        self.name = name
    
    @abstractmethod
    def calculate(self, tour: List[int], runtime: float, 
                  instance: TSPInstance) -> float:
        """计算指标值"""
        pass

@dataclass
class EvaluationResult:
    """单次评估结果"""
    solver: str
    instance: str
    category: str
    size_label: str
    type_label: str
    num_cities: int
    metrics: Dict[str, float]
    tour: List[int]
    
class TSPEvaluator:
    def __init__(self, results_path: str = "evaluation_results_progress.csv", best_results_path: str = "evaluation_results_best.csv", auto_save: bool = True):
        self.metrics: List[Metric] = []
        self.results: List[EvaluationResult] = []
        self._runs_data: Dict[str, List[Dict]] = defaultdict(list)  # 用于稳定性统计
        self.results_path = results_path
        self.best_results_path = best_results_path
        self.auto_save = auto_save
        self._completed_runs: Dict[Tuple[str, str], int] = defaultdict(int)
        self._loaded_results: Dict[Tuple[str, str], List[EvaluationResult]] = defaultdict(list)
        self._best_results: Dict[Tuple[str, str], EvaluationResult] = {}
        self._load_existing_results()

    @staticmethod
    def _split_category(category: str) -> Tuple[str, str]:
        parts = str(category).split("_", 1)
        size_label = parts[0] if parts else "Unknown"
        type_label = parts[1] if len(parts) > 1 else "Unknown"
        return size_label, type_label

    def _load_existing_results(self) -> None:
        if not self.results_path or not os.path.exists(self.results_path):
            return
        try:
            df = pd.read_csv(self.results_path)
        except Exception:
            return

        fixed_cols = {"solver", "instance", "category", "size_label", "type_label", "num_cities"}
        for _, row in df.iterrows():
            solver = row.get("solver")
            instance = row.get("instance")
            category = row.get("category")
            if pd.isna(solver) or pd.isna(instance) or pd.isna(category):
                continue
            size_label = row.get("size_label") if "size_label" in row else None
            type_label = row.get("type_label") if "type_label" in row else None
            if not size_label or not type_label:
                size_label, type_label = self._split_category(category)

            metrics = {
                col: row[col]
                for col in df.columns
                if col not in fixed_cols and col in row
            }

            result = EvaluationResult(
                solver=str(solver),
                instance=str(instance),
                category=str(category),
                size_label=str(size_label),
                type_label=str(type_label),
                num_cities=int(row.get("num_cities", 0)) if not pd.isna(row.get("num_cities", 0)) else 0,
                metrics=metrics,
                tour=[]
            )
            key = (result.solver, result.instance)
            self._loaded_results[key].append(result)
            self._completed_runs[key] += 1
            self._update_best_result(key, result)

        self.results = list(self._best_results.values())

    @staticmethod
    def _get_tour_length(result: EvaluationResult) -> Optional[float]:
        if "TourLength" in result.metrics:
            return float(result.metrics["TourLength"])
        return None

    def _is_better(self, candidate: EvaluationResult, current: Optional[EvaluationResult]) -> bool:
        if current is None:
            return True
        cand_len = self._get_tour_length(candidate)
        curr_len = self._get_tour_length(current)
        if cand_len is not None and curr_len is not None:
            return cand_len < curr_len
        if "OptimalityGap(%)" in candidate.metrics and "OptimalityGap(%)" in current.metrics:
            return float(candidate.metrics["OptimalityGap(%)"]) < float(current.metrics["OptimalityGap(%)"])
        if "Runtime(s)" in candidate.metrics and "Runtime(s)" in current.metrics:
            return float(candidate.metrics["Runtime(s)"]) < float(current.metrics["Runtime(s)"])
        return True

    def _update_best_result(self, key: Tuple[str, str], result: EvaluationResult) -> None:
        current = self._best_results.get(key)
        if self._is_better(result, current):
            self._best_results[key] = result
            if self.auto_save:
                self._write_best_results()

    def _append_result(self, result: EvaluationResult) -> None:
        if not self.results_path:
            return
        row = {
            "solver": result.solver,
            "instance": result.instance,
            "category": result.category,
            "size_label": result.size_label,
            "type_label": result.type_label,
            "num_cities": result.num_cities,
            **result.metrics
        }
        df = pd.DataFrame([row])
        header = not os.path.exists(self.results_path)
        df.to_csv(self.results_path, mode="a", header=header, index=False)

    def _write_best_results(self) -> None:
        if not self.best_results_path:
            return
        rows = []
        for r in self._best_results.values():
            rows.append({
                "solver": r.solver,
                "instance": r.instance,
                "category": r.category,
                "size_label": r.size_label,
                "type_label": r.type_label,
                "num_cities": r.num_cities,
                **r.metrics
            })
        df = pd.DataFrame(rows)
        df.to_csv(self.best_results_path, index=False)

    def add_metric(self, metric: Metric) -> 'TSPEvaluator':
        self.metrics.append(metric)
        return self
    
    def evaluate(self, solver: TSPSolver, instance: TSPInstance,
                 time_limit: float = 36000.0, runs: int = 1,
                 save_gif: bool = True, gif_dir: str = "outputs/gifs") -> List[EvaluationResult]:
        """评估求解器，支持多次运行计算稳定性"""
        print(f"\n评估: {solver.name} on {instance.instance_id} (runs={runs})")
        
        run_results = []
        best_tour = None
        best_tour_len = float("inf")

        key = (solver.name, instance.instance_id)
        already_done = self._completed_runs.get(key, 0)
        if already_done >= runs:
            cached = self._loaded_results.get(key, [])
            print(f"  已完成 {already_done} 次评估，跳过。")
            return cached[:runs]
        
        for run in range(already_done, runs):
            tour, runtime, memory = solver.solve(instance, time_limit)
            tour_len = instance.calculate_path_distance(tour)
            
            # 计算所有指标
            metric_values = {}
            for metric in self.metrics:
                value = metric.calculate(tour, runtime, memory, instance)
                metric_values[metric.name] = value
            if "TourLength" not in metric_values:
                metric_values["TourLength"] = tour_len
            
            # 存储单次结果
            self._runs_data[f"{solver.name}_{instance.instance_id}"].append(metric_values)

            if tour_len < best_tour_len:
                best_tour_len = tour_len
                best_tour = tour
            
            result = EvaluationResult(
                solver=solver.name,
                instance=instance.instance_id,
                category=instance.category,
                size_label=self._split_category(instance.category)[0],
                type_label=self._split_category(instance.category)[1],
                num_cities=instance.num_cities,
                metrics=metric_values,
                tour=tour
            )
            run_results.append(result)
            self._completed_runs[key] += 1
            self._loaded_results[key].append(result)
            if self.auto_save:
                self._append_result(result)
            self._update_best_result(key, result)
        
        if save_gif and best_tour is not None:
            safe_solver = solver.name.replace(" ", "_")
            safe_instance = instance.instance_id.replace(" ", "_")
            out_path = os.path.join(
                gif_dir,
                safe_solver,
                f"{safe_instance}_best_len{best_tour_len:.2f}.gif"
            )
            save_tour_gif(instance, best_tour, out_path)

        # 打印结果
        self._print_result(run_results, runs)
        self.results = list(self._best_results.values())
        
        return run_results
    
    def _print_result(self, run_results: List[EvaluationResult], runs: int):
        """打印评估结果"""
        if runs == 1:
            r = run_results[0]
            print(f"  结果: ", end="")
            for name, value in r.metrics.items():
                if "Rate" in name or "Gap" in name:
                    print(f"{name}={value:.2f}% ", end="")
                elif "Memory" in name:
                    print(f"{name}={value:.2f}B ", end="")
                elif "Runtime" in name:
                    print(f"{name}={value:.4f}s ", end="")
                else:
                    print(f"{name}={value:.2f} ", end="")
            print()
        else:
            # 多次运行：计算统计值
            print(f"  多次运行统计 (n={runs}):")
            metric_names = list(run_results[0].metrics.keys())
            for name in metric_names:
                values = [r.metrics[name] for r in run_results]
                mean_val = np.mean(values)
                std_val = np.std(values)
                
                # 4. 稳定性指标：显示平均值±标准差
                unit = "%" if "Rate" in name or "Gap" in name else "B" if "Memory" in name else "s" if "Runtime" in name else ""
                print(f"    {name}: {mean_val:.4f}±{std_val:.4f}{unit}")
    
    def compare(self, solvers: List[TSPSolver], instance: TSPInstance,
                runs: int = 1) -> Dict[str, List[EvaluationResult]]:
        """对比多个求解器"""
        print(f"\n{'='*60}")
        print(f"对比求解器 on {instance.instance_id} ({instance.category}, {instance.num_cities} cities)")
        print(f"{'='*60}")
        
        comparison = {}
        for solver in solvers:
            results = self.evaluate(solver, instance, runs=runs)
            comparison[solver.name] = results
        
        return comparison
    
    def benchmark(self, solver: TSPSolver, factory: TSPInstanceFactory,
                  categories: Optional[List[str]] = None,
                  runs: int = 1) -> pd.DataFrame:
        """批量基准测试，支持5.不同规模 和 6.不同结构的影响分析"""
        print(f"\n{'='*60}")
        print(f"批量基准测试: {solver.name}")
        print(f"{'='*60}")
        
        instances = factory.get_all()
        if categories:
            instances = [inst for inst in instances if inst.category in categories]
        
        records = []
        for inst in instances:
            try:
                results = self.evaluate(solver, inst, runs=runs)
                size_label, type_label = self._split_category(inst.category)
                # 取平均（如果多次运行）
                avg_metrics = {}
                for name in results[0].metrics.keys():
                    values = [r.metrics[name] for r in results]
                    avg_metrics[name] = np.mean(values)
                    avg_metrics[f"{name}_std"] = np.std(values)
                
                records.append({
                    'instance_id': inst.instance_id,
                    'category': inst.category,
                    'size_label': size_label,
                    'type_label': type_label,
                    'num_cities': inst.num_cities,
                    **avg_metrics
                })
            except Exception as e:
                print(f"  错误: {e}")
        
        df = pd.DataFrame(records)
        
        # 分析不同规模和结构的影响
        self._analyze_impact(df)
        
        return df
    
    def _analyze_impact(self, df: pd.DataFrame):
        """分析规模和结构影响"""
        print(f"\n{'='*60}")
        print("影响分析")
        print(f"{'='*60}")
        
        gap_col = [c for c in df.columns if 'OptimalityGap' in c and '_std' not in c][0]
        time_col = [c for c in df.columns if 'Runtime' in c and '_std' not in c][0]

        # 5. 不同输入规模的影响
        print("\n【规模影响】按规模标签分组:")
        if 'size_label' in df.columns:
            size_groups = df.groupby('size_label')
        else:
            size_groups = df.groupby(pd.cut(df['num_cities'], bins=[0, 30, 100, 1000, 10000],
                                           labels=['Small(<30)', 'Medium(30-100)',
                                                   'Large(100-1000)', 'XLarge(>1000)']))
        for name, group in size_groups:
            if not group.empty:
                print(f"  {name}: n={len(group)}, "
                      f"AvgGap={group[gap_col].mean():.2f}%, "
                      f"AvgTime={group[time_col].mean():.4f}s")

        # 6. 不同实例结构的影响
        print("\n【结构影响】按类型标签分组:")
        if 'type_label' in df.columns:
            type_groups = df.groupby('type_label')
        else:
            type_groups = df.groupby('category')
        for name, group in type_groups:
            if not group.empty:
                print(f"  {name}: n={len(group)}, "
                      f"AvgGap={group[gap_col].mean():.2f}%, "
                      f"AvgTime={group[time_col].mean():.4f}s")
    
    def export_results(self, filename: str = "evaluation_results.csv"):
        """导出结果到CSV"""
        if not self.results:
            print("无结果可导出")
            return
        
        data = []
        for r in self.results:
            row = {
                'solver': r.solver,
                'instance': r.instance,
                'category': r.category,
                'size_label': r.size_label,
                'type_label': r.type_label,
                'num_cities': r.num_cities,
                **r.metrics
            }
            data.append(row)
        
        df = pd.DataFrame(data)
        df.to_csv(filename, index=False)
        print(f"\n结果已导出: {filename}")