run_experiments.py

#!/usr/bin/env python3
"""
实验数据生成脚本
用于生成报告所需的真实对比数据
"""
import sys
import time
import json
import random
import math
import numpy as np
from typing import Dict, List, Any, Optional

# 添加当前目录到路径
sys.path.insert(0, '.')

from src.agent_base import Position
from src.environment import Environment
from src.drone_agent import DroneAgent, DroneSpec
from src.coordinator_agent import CoordinatorAgent
from src.communication import MessageBus
from src.scenarios import ScenarioManager

class ExperimentRunner:
    def __init__(self):
        self.results = {}

    def _trial_seed(self, experiment_id: int, variant_id: int, trial_index: int) -> int:
        return 202500 + experiment_id * 100000 + variant_id * 10000 + trial_index * 97

    def setup_simulation(self, scenario: str, num_drones: int, use_vortex: bool, use_entropy: bool):
        """初始化单次模拟环境"""
        map_size = 120
        message_bus = MessageBus()
        environment = Environment(width=map_size, height=map_size)
        center = map_size / 2
        environment.base_position = Position(center, center)
        
        # 固定随机种子以保证场景一致
        state = random.getstate()
        random.seed(42) 
        ScenarioManager.create_scenario(scenario, environment)
        random.setstate(state) # 恢复之前的随机状态
        
        coordinator = CoordinatorAgent(
            agent_id="coord_001",
            name="指挥中心",
            message_bus=message_bus,
            use_entropy=use_entropy
        )
        coordinator.position = Position(center, center)
        
        drones = []
        for i in range(num_drones):
            angle = (2 * math.pi * i) / num_drones
            dist = 5
            start_pos = Position(
                center + dist * math.cos(angle),
                center + dist * math.sin(angle)
            )
            
            drone = DroneAgent(
                agent_id=f"drone_{i}",
                name=f"Drone-{i}",
                position=start_pos,
                spec=DroneSpec(max_speed=6.0, perception_range=15.0),
                message_bus=message_bus,
                use_vortex=use_vortex
            )
            drones.append(drone)
            coordinator.register_drone(drone)
            
        return environment, coordinator, drones

    def run_trial(
        self,
        setup_args: Dict,
        max_steps: Optional[int] = 500,
        stop_on_all_targets: bool = True,
        suppress_mission_complete: bool = False,
        plateau_window: Optional[int] = None,
        plateau_epsilon: float = 1e-4,
        seed: Optional[int] = None,
    ) -> Dict:
        """运行单次实验"""
        if seed is not None:
            random.seed(seed)
            np.random.seed(seed)
        else:
            random.seed(time.time())
        
        env, coord, drones = self.setup_simulation(**setup_args)
        coord.start_mission(env)

        inferred_max_steps = None
        if max_steps is None:
            num_drones = int(setup_args.get("num_drones", 1)) or 1
            inferred_max_steps = int((env.coverage_map.rows * env.coverage_map.cols * 4) / num_drones)
            max_steps = max(400, inferred_max_steps)
        
        coverage_history = []
        targets_found_history = []
        
        completed_step = None
        completed_by_targets = False
        best_coverage = -1.0
        since_improve = 0
        
        for step in range(max_steps):
            # Update Drones
            for drone in drones:
                drone.update(env)
                perception = drone.perceive(env)
                for target in perception.get('visible_targets', []):
                    env.mark_target_found(target.target_id, drone.agent_id)
            
            # Update Coordinator
            coord_perception = coord.perceive(env)
            if suppress_mission_complete:
                stats = coord_perception.get("environment_stats")
                if isinstance(stats, dict):
                    stats["total_targets"] = int(stats.get("found_targets", 0)) + 1
            action = coord.decide(coord_perception)
            if action:
                coord.act(action)
                
            env.step()
            
            # Record stats
            stats = env.get_statistics()
            coverage_history.append(stats['coverage_ratio'])
            targets_found_history.append(stats['found_targets'])
            
            current_cov = float(stats['coverage_ratio'])
            if current_cov > best_coverage + plateau_epsilon:
                best_coverage = current_cov
                since_improve = 0
            else:
                since_improve += 1

            if stop_on_all_targets and stats['found_targets'] == stats['total_targets'] and completed_step is None:
                completed_step = step
                completed_by_targets = True
                break

            if plateau_window is not None and since_improve >= plateau_window:
                completed_step = step
                break
                
        return {
            "total_steps": completed_step if completed_step is not None else max_steps,
            "success": bool(completed_by_targets),
            "final_coverage": coverage_history[-1],
            "final_targets": targets_found_history[-1],
            "total_targets": stats['total_targets'],
            "coverage_history": coverage_history,
            "max_steps": max_steps,
            "safety_cap_steps": inferred_max_steps if inferred_max_steps is not None else max_steps,
        }

    def run_obstacle_experiment(self):
        """
        实验 1: 避障性能对比 (APF vs VPF)
        场景: Extreme (密集障碍物)
        """
        print("Running Obstacle Avoidance Experiment (APF vs VPF)...")
        scenario = "extreme" # 使用 extreme 场景
        num_drones = 4
        trials = 3 # 运行3次取平均
        
        results = {"APF": [], "VPF": []}
        
        for method in ["APF", "VPF"]:
            use_vortex = (method == "VPF")
            variant_id = 1 if method == "VPF" else 0
            for i in range(trials):
                print(f"  Running {method} Trial {i+1}/{trials}...")
                res = self.run_trial({
                    "scenario": scenario,
                    "num_drones": num_drones,
                    "use_vortex": use_vortex,
                    "use_entropy": True
                }, max_steps=800, seed=self._trial_seed(1, variant_id, i))
                results[method].append(res)
                
        # 统计分析
        summary = {}
        for method, data in results.items():
            avg_targets = np.mean([d['final_targets'] for d in data])
            avg_steps = np.mean([d['total_steps'] for d in data])
            success_rate = np.mean([1 if d['success'] else 0 for d in data])
            summary[method] = {
                "avg_targets_found": float(avg_targets),
                "avg_steps": float(avg_steps),
                "success_rate": float(success_rate)
            }
            
        self.results["obstacle_avoidance"] = summary
        print(f"Obstacle Experiment Result: {json.dumps(summary, indent=2)}")

    def run_coverage_experiment(self):
        """
        实验 2: 覆盖效率对比 (Random vs Info-Driven)
        场景: City (规则街区)
        """
        print("\nRunning Coverage Efficiency Experiment (Random vs Info-Driven)...")
        scenario = "city"
        num_drones = 2 # 减少无人机数量，凸显策略差异
        trials = 10
        
        results = {"Random": [], "Info-Driven": []}
        
        for method in ["Random", "Info-Driven"]:
            use_entropy = (method == "Info-Driven")
            variant_id = 1 if method == "Info-Driven" else 0
            for i in range(trials):
                print(f"  Running {method} Trial {i+1}/{trials}...")
                res = self.run_trial({
                    "scenario": scenario,
                    "num_drones": num_drones,
                    "use_vortex": True,
                    "use_entropy": use_entropy
                }, max_steps=None, stop_on_all_targets=False, suppress_mission_complete=True, plateau_window=200, seed=self._trial_seed(2, variant_id, i))
                results[method].append(res)

        # 统计分析
        summary = {}
        for method, data in results.items():
            def _mean_std(values: List[Optional[float]]):
                clean = [v for v in values if v is not None]
                if not clean:
                    return None, None
                return float(np.mean(clean)), float(np.std(clean, ddof=1)) if len(clean) > 1 else 0.0

            steps_to_70 = []
            steps_to_90 = []
            reach_70 = 0
            reach_90 = 0
            auc_coverages = []
            final_coverages = []
            for run in data:
                cov_hist = run.get('coverage_history', [])
                final_coverages.append(run['final_coverage'])
                auc_coverages.append(float(np.mean(cov_hist)) if cov_hist else float(run['final_coverage']))
                found = False
                for idx, cov in enumerate(cov_hist):
                    if cov >= 0.7:
                        steps_to_70.append(idx)
                        reach_70 += 1
                        break
                else:
                    steps_to_70.append(None)

                for idx, cov in enumerate(cov_hist):
                    if cov >= 0.9:
                        steps_to_90.append(idx)
                        reach_90 += 1
                        found = True
                        break
                if not found:
                    steps_to_90.append(None)

            avg_final_coverage, std_final_coverage = _mean_std(final_coverages)
            avg_auc_coverage, std_auc_coverage = _mean_std(auc_coverages)
            avg_steps_70, std_steps_70 = _mean_std(steps_to_70)
            avg_steps_90, std_steps_90 = _mean_std(steps_to_90)
            
            summary[method] = {
                "max_steps": int(max([r.get("max_steps", 0) for r in data] or [0])),
                "safety_cap_steps": int(max([r.get("safety_cap_steps", 0) for r in data] or [0])),
                "trials": int(len(data)),
                "avg_final_coverage": avg_final_coverage,
                "std_final_coverage": std_final_coverage,
                "avg_auc_coverage": avg_auc_coverage,
                "std_auc_coverage": std_auc_coverage,
                "reach_70_rate": float(reach_70 / len(data)) if data else 0.0,
                "avg_steps_to_70_coverage": avg_steps_70,
                "std_steps_to_70_coverage": std_steps_70,
                "reach_90_rate": float(reach_90 / len(data)) if data else 0.0,
                "avg_steps_to_90_coverage": avg_steps_90,
                "std_steps_to_90_coverage": std_steps_90
            }
            
        self.results["coverage_efficiency"] = summary
        print(f"Coverage Experiment Result: {json.dumps(summary, indent=2)}")

    def run_scalability_experiment(self):
        """
        实验 3: 集群规模扩展性分析
        场景: Medium, 考察 3, 5, 8 架无人机的效率
        """
        print("\nRunning Scalability Experiment (3 vs 5 vs 8 Drones)...")
        scenario = "medium"
        drone_counts = [3, 5, 8]
        trials = 3
        
        summary = {}
        
        for count in drone_counts:
            label = f"{count} Drones"
            data = []
            variant_id = count
            for i in range(trials):
                print(f"  Running {label} Trial {i+1}/{trials}...")
                res = self.run_trial({
                    "scenario": scenario,
                    "num_drones": count,
                    "use_vortex": True,
                    "use_entropy": True
                }, max_steps=600, seed=self._trial_seed(3, variant_id, i))
                data.append(res)
            
            avg_steps = np.mean([d['total_steps'] for d in data])
            avg_coverage = np.mean([d['final_coverage'] for d in data])
            summary[label] = {
                "avg_steps_to_complete": float(avg_steps),
                "avg_final_coverage": float(avg_coverage)
            }
            
        self.results["scalability"] = summary
        print(f"Scalability Experiment Result: {json.dumps(summary, indent=2)}")

    def run_robustness_experiment(self):
        """
        实验 4: 极端环境鲁棒性测试
        场景: Hard vs Extreme, 考察 VPF 在极限情况下的表现
        """
        print("\nRunning Robustness Experiment (Hard vs Extreme)...")
        scenarios = ["hard", "extreme"]
        num_drones = 5
        trials = 3
        
        summary = {}
        
        for scenario in scenarios:
            data = []
            variant_id = 1 if scenario == "extreme" else 0
            for i in range(trials):
                print(f"  Running {scenario} Scenario Trial {i+1}/{trials}...")
                res = self.run_trial({
                    "scenario": scenario,
                    "num_drones": num_drones,
                    "use_vortex": True,
                    "use_entropy": True
                }, max_steps=1200, seed=self._trial_seed(4, variant_id, i))
                data.append(res)
            
            success_rate = np.mean([1 if d['success'] else 0 for d in data])
            successful_steps = [d['total_steps'] for d in data if d.get('success')]
            avg_steps_success = float(np.mean(successful_steps)) if successful_steps else None
            avg_steps_capped = float(np.mean([d['total_steps'] for d in data])) if data else None
            summary[scenario] = {
                "success_rate": float(success_rate),
                "avg_steps_to_complete": avg_steps_success,
                "avg_steps_capped": avg_steps_capped
            }
            
        self.results["robustness"] = summary
        print(f"Robustness Experiment Result: {json.dumps(summary, indent=2)}")

    def save_results(self):
        with open("experiment_results.json", "w") as f:
            json.dump(self.results, f, indent=2)
        print("\nResults saved to experiment_results.json")

if __name__ == "__main__":
    runner = ExperimentRunner()
    runner.run_obstacle_experiment()
    runner.run_coverage_experiment()
    runner.run_scalability_experiment()
    runner.run_robustness_experiment()
    runner.save_results()