Spaces:
Sleeping
Sleeping
| """ | |
| Performance Analytics & Episode History | |
| Tracks episode metrics over time and provides analysis tools. | |
| """ | |
| from typing import List, Dict, Any, Optional | |
| from dataclasses import dataclass, field | |
| from datetime import datetime | |
| import json | |
| import os | |
| class EpisodeMetrics: | |
| """Metrics for a single episode.""" | |
| episode_id: str | |
| task_id: str | |
| start_time: datetime | |
| end_time: Optional[datetime] = None | |
| steps: int = 0 | |
| total_reward: float = 0.0 | |
| total_vehicles: int = 0 | |
| total_emergency: int = 0 | |
| total_waiting_time: float = 0.0 | |
| total_collisions: int = 0 | |
| phase_changes: int = 0 | |
| avg_queue_length: List[float] = field(default_factory=lambda: [0.0, 0.0, 0.0, 0.0]) | |
| rewards_history: List[float] = field(default_factory=list) | |
| decisions: List[Dict[str, Any]] = field(default_factory=list) | |
| def duration_seconds(self) -> float: | |
| if self.end_time: | |
| return (self.end_time - self.start_time).total_seconds() | |
| return 0.0 | |
| def throughput_per_step(self) -> float: | |
| if self.steps > 0: | |
| return self.total_vehicles / self.steps | |
| return 0.0 | |
| def avg_reward_per_step(self) -> float: | |
| if self.steps > 0: | |
| return self.total_reward / self.steps | |
| return 0.0 | |
| class EpisodeHistory: | |
| """Store and analyze episode history.""" | |
| def __init__(self, max_episodes: int = 100): | |
| self.episodes: List[EpisodeMetrics] = [] | |
| self.max_episodes = max_episodes | |
| self._current: Optional[EpisodeMetrics] = None | |
| def start_episode(self, episode_id: str, task_id: str) -> EpisodeMetrics: | |
| """Start tracking a new episode.""" | |
| episode = EpisodeMetrics( | |
| episode_id=episode_id, | |
| task_id=task_id, | |
| start_time=datetime.now(), | |
| ) | |
| self._current = episode | |
| return episode | |
| def record_step( | |
| self, | |
| step: int, | |
| reward: float, | |
| action: Dict[str, Any], | |
| observation: Dict[str, Any], | |
| ) -> None: | |
| """Record a step in the current episode.""" | |
| if self._current: | |
| self._current.steps = step | |
| self._current.total_reward += reward | |
| self._current.rewards_history.append(reward) | |
| # Track queue lengths for averaging | |
| queues = observation.get("queue_lengths", [0, 0, 0, 0]) | |
| for i, q in enumerate(queues): | |
| self._current.avg_queue_length[i] = ( | |
| self._current.avg_queue_length[i] * (step - 1) + q | |
| ) / step | |
| # Record decision | |
| self._current.decisions.append({ | |
| "step": step, | |
| "action": action, | |
| "phase": observation.get("current_phase"), | |
| "queues": queues, | |
| "emergency_queues": observation.get("emergency_queue", [0, 0, 0, 0]), | |
| }) | |
| def record_state(self, state: Dict[str, Any]) -> None: | |
| """Record final state metrics.""" | |
| if self._current: | |
| self._current.total_vehicles = state.get("total_vehicles_passed", 0) | |
| self._current.total_emergency = state.get("total_emergency_passed", 0) | |
| self._current.total_waiting_time = state.get("total_waiting_time", 0.0) | |
| self._current.total_collisions = state.get("total_collisions", 0) | |
| self._current.phase_changes = state.get("total_phase_changes", 0) | |
| def end_episode(self) -> EpisodeMetrics: | |
| """Finalize the current episode.""" | |
| if self._current: | |
| self._current.end_time = datetime.now() | |
| self.episodes.append(self._current) | |
| # Trim old episodes | |
| if len(self.episodes) > self.max_episodes: | |
| self.episodes = self.episodes[-self.max_episodes:] | |
| result = self._current | |
| self._current = None | |
| return result | |
| raise RuntimeError("No active episode to end") | |
| def get_summary(self, task_id: Optional[str] = None) -> Dict[str, Any]: | |
| """Get summary statistics.""" | |
| episodes = self.episodes | |
| if task_id: | |
| episodes = [e for e in episodes if e.task_id == task_id] | |
| if not episodes: | |
| return {"message": "No episodes recorded yet"} | |
| total_episodes = len(episodes) | |
| avg_reward = sum(e.avg_reward_per_step for e in episodes) / total_episodes | |
| avg_throughput = sum(e.throughput_per_step for e in episodes) / total_episodes | |
| avg_duration = sum(e.duration_seconds for e in episodes) / total_episodes | |
| # Find best episode | |
| best_idx = max(range(total_episodes), key=lambda i: episodes[i].avg_reward_per_step) | |
| best = episodes[best_idx] | |
| return { | |
| "total_episodes": total_episodes, | |
| "avg_reward_per_step": round(avg_reward, 4), | |
| "avg_throughput_per_step": round(avg_throughput, 4), | |
| "avg_duration_seconds": round(avg_duration, 2), | |
| "best_episode": { | |
| "episode_id": best.episode_id, | |
| "task_id": best.task_id, | |
| "reward_per_step": round(best.avg_reward_per_step, 4), | |
| "total_reward": round(best.total_reward, 2), | |
| "steps": best.steps, | |
| }, | |
| "recent_performance": [ | |
| { | |
| "episode_id": e.episode_id, | |
| "task_id": e.task_id, | |
| "reward_per_step": round(e.avg_reward_per_step, 4), | |
| "steps": e.steps, | |
| } | |
| for e in episodes[-10:] | |
| ], | |
| } | |
| def get_episode_details(self, episode_id: str) -> Optional[Dict[str, Any]]: | |
| """Get detailed metrics for a specific episode.""" | |
| for episode in self.episodes: | |
| if episode.episode_id == episode_id: | |
| return { | |
| "episode_id": episode.episode_id, | |
| "task_id": episode.task_id, | |
| "start_time": episode.start_time.isoformat(), | |
| "end_time": episode.end_time.isoformat() if episode.end_time else None, | |
| "duration_seconds": episode.duration_seconds, | |
| "steps": episode.steps, | |
| "total_reward": round(episode.total_reward, 2), | |
| "total_vehicles": episode.total_vehicles, | |
| "total_emergency": episode.total_emergency, | |
| "total_collisions": episode.total_collisions, | |
| "phase_changes": episode.phase_changes, | |
| "avg_queue_lengths": [round(q, 2) for q in episode.avg_queue_length], | |
| "throughput_per_step": round(episode.throughput_per_step, 4), | |
| "reward_per_step": round(episode.avg_reward_per_step, 4), | |
| "rewards_history": [round(r, 3) for r in episode.rewards_history], | |
| "decision_count": len(episode.decisions), | |
| } | |
| return None | |
| def export_to_json(self, filepath: str) -> None: | |
| """Export all episodes to JSON file.""" | |
| data = { | |
| "export_time": datetime.now().isoformat(), | |
| "total_episodes": len(self.episodes), | |
| "episodes": [ | |
| { | |
| "episode_id": e.episode_id, | |
| "task_id": e.task_id, | |
| "steps": e.steps, | |
| "total_reward": e.total_reward, | |
| "total_vehicles": e.total_vehicles, | |
| "avg_reward_per_step": e.avg_reward_per_step, | |
| } | |
| for e in self.episodes | |
| ], | |
| } | |
| with open(filepath, 'w') as f: | |
| json.dump(data, f, indent=2) | |
| # Global history instance | |
| _episode_history = EpisodeHistory() | |
| def get_history() -> EpisodeHistory: | |
| """Get the global episode history instance.""" | |
| return _episode_history | |