"""Reproducible Phase 2.1 SUMO controller evaluation. The four evaluated controllers are intentionally kept separate: * fixed_time: SUMO's configured signal program, with no intervention; * density_based: highest current movement demand phase; * predictive_adaptive: highest phase-specific GNN prediction; * integrated_ai: the verified Phase2SUMOController. Every run uses a fresh SUMO process, the same network/configuration, route file, simulation window, and explicit seed. Metrics are collected directly from TraCI and are never substituted when a run fails. """ from __future__ import annotations import argparse import csv import json import math import os import statistics import sys from collections import defaultdict from datetime import datetime from pathlib import Path import numpy as np PROJECT_ROOT = Path(__file__).resolve().parents[2] SRC_ROOT = PROJECT_ROOT / "src" if str(SRC_ROOT) not in sys.path: sys.path.insert(0, str(SRC_ROOT)) if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) import traci from integration.decision_engine import DecisionEngine from integration.fallback_controller import FallbackController from integration.phase_generator import PhaseGenerator from integration.prediction_adapter import TrafficPredictionAdapter from integration.safety_gate import SafetyGate from integration.sumo_conflict_adapter import SUMOConflictAdapter from integration.sumo_decision_controller import ( HISTORY_LENGTH, NUM_SENSORS, Phase2SUMOController, _phase_objects, _phase_sensor_ids, ) from integration.sumo_demand import SUMODemandCalculator from integration.sumo_phase_adapter import SUMOPhaseAdapter from integration.sumo_topology_adapter import SUMOTopologyAdapter from integration.traffic_history import TrafficHistory from integration.sumo_prediction_controller import SUMOTrafficStateBuilder CONTROLLERS = ( "fixed_time", "density_based", "predictive_adaptive", "integrated_ai", ) METRICS = ( "average_waiting_time", "total_waiting_time", "maximum_waiting_time", "average_queue_length", "maximum_queue_length", "throughput", "average_travel_time", "total_travel_time", "number_of_stops", "fuel_consumption_mg", "co2_emissions_mg", ) class MetricAccumulator: """Collect raw TraCI measurements and calculate documented metrics.""" def __init__(self, traci_module=traci): self.traci = traci_module self.departure_time = {} self.last_waiting = {} self.last_speed = {} self.completed_waiting = [] self.completed_travel = [] self.queue_samples = [] self.total_stops = 0 self.total_fuel_mg = 0.0 self.total_co2_mg = 0.0 self.total_distance_m = 0.0 self.samples = 0 self.max_queue = 0 self.max_waiting = 0.0 def record_sample(self, queue_length, fuel_mg=0.0, co2_mg=0.0, stops=0, distance_m=0.0): """Record one simulation-timestep observation for unit tests/runtime.""" queue_length = int(queue_length) self.queue_samples.append(queue_length) self.max_queue = max(self.max_queue, queue_length) self.total_fuel_mg += float(fuel_mg) self.total_co2_mg += float(co2_mg) self.total_stops += int(stops) self.total_distance_m += float(distance_m) self.samples += 1 def record_vehicle_completion(self, travel_time, waiting_time): self.completed_travel.append(float(travel_time)) self.completed_waiting.append(float(waiting_time)) self.max_waiting = max(self.max_waiting, float(waiting_time)) @staticmethod def _vehicle_value(getter, vehicle_id, default=0.0): try: return float(getter(vehicle_id)) except Exception: return float(default) def sample(self): """Sample active vehicles and arrivals after one TraCI step.""" now = float(self.traci.simulation.getTime()) try: delta_t = float(self.traci.simulation.getDeltaT()) / 1000.0 except Exception: delta_t = 1.0 departed = list(self.traci.simulation.getDepartedIDList()) for vehicle_id in departed: self.departure_time.setdefault(vehicle_id, now) active_ids = list(self.traci.vehicle.getIDList()) queue_length = 0 fuel = 0.0 co2 = 0.0 distance = 0.0 stops = 0 for vehicle_id in active_ids: self.departure_time.setdefault(vehicle_id, now) speed = self._vehicle_value(self.traci.vehicle.getSpeed, vehicle_id) waiting = self._vehicle_value( self.traci.vehicle.getAccumulatedWaitingTime, vehicle_id, ) self.last_waiting[vehicle_id] = waiting self.max_waiting = max(self.max_waiting, waiting) if speed <= 0.1: queue_length += 1 if vehicle_id in self.last_speed and self.last_speed[vehicle_id] > 0.1 and speed <= 0.1: stops += 1 self.last_speed[vehicle_id] = speed fuel += self._vehicle_value(self.traci.vehicle.getFuelConsumption, vehicle_id) * delta_t co2 += self._vehicle_value(self.traci.vehicle.getCO2Emission, vehicle_id) * delta_t distance += max(0.0, speed * delta_t) self.record_sample(queue_length, fuel, co2, stops, distance) for vehicle_id in self.traci.simulation.getArrivedIDList(): depart = self.departure_time.pop(vehicle_id, now) self.record_vehicle_completion( travel_time=max(0.0, now - depart), waiting_time=self.last_waiting.pop(vehicle_id, 0.0), ) self.last_speed.pop(vehicle_id, None) def finalize(self): completed = len(self.completed_travel) return { "average_waiting_time": ( sum(self.completed_waiting) / completed if completed else None ), "total_waiting_time": sum(self.completed_waiting), "maximum_waiting_time": self.max_waiting, "average_queue_length": ( sum(self.queue_samples) / len(self.queue_samples) if self.queue_samples else None ), "maximum_queue_length": self.max_queue, "throughput": completed, "average_travel_time": ( sum(self.completed_travel) / completed if completed else None ), "total_travel_time": sum(self.completed_travel), "number_of_stops": self.total_stops, "fuel_consumption_mg": self.total_fuel_mg, "co2_emissions_mg": self.total_co2_mg, "total_distance_m": self.total_distance_m, "completed_vehicles": completed, "measurement_samples": self.samples, } def _sumo_command(seed, config): executable = Phase2SUMOController().sumo_cmd[0] return [executable, "-c", str(config), "--seed", str(int(seed))] def _runtime_metadata(connection): topology_adapter = SUMOTopologyAdapter(connection) phase_adapter = SUMOPhaseAdapter(connection) conflict_adapter = SUMOConflictAdapter(phase_adapter) runtimes = {} for junction_id in topology_adapter.discover_signalized_junctions(): topology = topology_adapter.discover_junction(junction_id) graph = conflict_adapter.build_conflict_graph( junction_id, topology, topology_adapter.sumo_metadata ) representations = phase_adapter.build_phase_representation( junction_id, topology_adapter.sumo_metadata ) runtimes[junction_id] = { "topology": topology, "generator": PhaseGenerator(graph), "phases": representations, "phase_objects": _phase_objects(representations), "phase_by_id": {p["phase_id"]: p for p in representations}, "last_control": -float("inf"), } return topology_adapter, phase_adapter, runtimes def _safe_baseline_phase(junction_id, phase_id, runtime, phase_adapter, now, last_control): phase = runtime["phase_by_id"][phase_id] current = phase_adapter.get_current_phase(junction_id) current_type = phase_adapter.get_phase_type(junction_id, current) timing_valid = ( phase["phase_index"] == current or current_type != "green" or now - last_control >= 10.0 ) safety = SafetyGate(min_green_seconds=10.0).validate( phase_id, topology_valid=bool(runtime["topology"].roads and runtime["topology"].movements), conflict_free=runtime["generator"].is_compatible(phase["movements"]), timing_valid=timing_valid, pedestrian_clear=True, downstream_available=True, emergency_safe=True, confidence=1.0, ) return safety def _select_density_phase(runtime, demand, current_phase): values = { phase.phase_id: sum(demand.get(movement_id, 0) for movement_id in phase.movement_ids) for phase in runtime["phase_objects"] } selected = max(values, key=values.get) return selected, values def _select_predictive_phase(runtime, predictions, sensor_builder): values = {} for phase in runtime["phases"]: sensor_ids = _phase_sensor_ids(phase, sensor_builder, runtime["topology"]) values[phase["phase_id"]] = ( float(np.mean(np.maximum(predictions[sensor_ids], 0.0))) if sensor_ids else 0.0 ) return max(values, key=values.get), values def _run_baseline(controller_name, seed, config, steps, warmup, control_interval): metrics = MetricAccumulator() traci.start(_sumo_command(seed, config)) try: topology_adapter, phase_adapter, runtimes = _runtime_metadata(traci) if controller_name == "predictive_adaptive": sensor_builder = SUMOTrafficStateBuilder( num_sensors=NUM_SENSORS, target_junctions=list(runtimes), ) history = TrafficHistory(HISTORY_LENGTH, NUM_SENSORS) predictor = TrafficPredictionAdapter() predictions = np.zeros(NUM_SENSORS, dtype=np.float32) else: sensor_builder = None history = None predictor = None predictions = None executor = Phase2SUMOController( traci_connection=traci, step_observer=metrics.sample, ) executor._observe_steps = True last_control = {junction_id: -float("inf") for junction_id in runtimes} for _ in range(int(warmup)): traci.simulationStep() for step in range(int(steps)): traci.simulationStep() metrics.sample() now = float(traci.simulation.getTime()) if history is not None: history.add_state(sensor_builder.build_state()) if controller_name == "fixed_time": continue if predictor is not None and step % 10 == 0: predictions = np.asarray( predictor.predict(history.get_padded_history()), dtype=np.float32, ) for junction_id, runtime in runtimes.items(): if now - last_control[junction_id] < control_interval: continue demand = SUMODemandCalculator().calculate_movement_demand( runtime["topology"], topology_adapter.sumo_metadata ) if controller_name == "density_based": selected, values = _select_density_phase( runtime, demand, phase_adapter.get_current_phase(junction_id) ) else: selected, values = _select_predictive_phase( runtime, predictions, sensor_builder ) safety = _safe_baseline_phase( junction_id, selected, runtime, phase_adapter, now, last_control[junction_id], ) if safety["approved"]: executor._execute( junction_id, runtime["phase_by_id"][selected]["phase_index"], phase_adapter, ) last_control[junction_id] = float(traci.simulation.getTime()) return metrics.finalize() finally: traci.close() def _run_integrated(seed, config, steps, warmup, control_interval): metrics = MetricAccumulator() controller = Phase2SUMOController( sumo_cmd=_sumo_command(seed, config), total_steps=steps, warmup_steps=warmup, control_interval=control_interval, step_observer=metrics.sample, ) controller.run() return metrics.finalize() def run_one(controller_name, seed, config, steps=200, warmup=10, control_interval=10.0): """Run exactly one controller/seed pair and preserve failures.""" started = datetime.utcnow().isoformat(timespec="seconds") + "Z" try: if controller_name == "integrated_ai": metrics = _run_integrated(seed, config, steps, warmup, control_interval) else: metrics = _run_baseline( controller_name, seed, config, steps, warmup, control_interval ) return { "controller": controller_name, "seed": int(seed), "status": "success", "started_at": started, **metrics, } except Exception as exc: try: traci.close() except Exception: pass return { "controller": controller_name, "seed": int(seed), "status": "failed", "started_at": started, "error": f"{type(exc).__name__}: {exc}", } def aggregate_results(results): summary = [] grouped = defaultdict(list) for result in results: if result.get("status") == "success": grouped[result["controller"]].append(result) for controller in CONTROLLERS: runs = grouped.get(controller, []) row = {"controller": controller, "successful_runs": len(runs)} for metric in METRICS: values = [r[metric] for r in runs if r.get(metric) is not None] row[f"{metric}_mean"] = statistics.mean(values) if values else None row[f"{metric}_std"] = statistics.stdev(values) if len(values) > 1 else 0.0 if values else None summary.append(row) return summary def _write_csv(path, rows): if not rows: return fields = [] for row in rows: for key in row: if key not in fields: fields.append(key) with path.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=fields) writer.writeheader() writer.writerows(rows) def _write_summary_markdown(path, summary): labels = { "fixed_time": "Fixed-Time", "density_based": "Density-Based", "predictive_adaptive": "Predictive Adaptive", "integrated_ai": "Integrated AI", } with path.open("w", encoding="utf-8") as handle: handle.write("# Phase 2.1 Controller Evaluation\n\n") handle.write("| Metric | Fixed-Time | Density-Based | Predictive Adaptive | Integrated AI |\n") handle.write("|---|---:|---:|---:|---:|\n") by_controller = {row["controller"]: row for row in summary} for metric in METRICS: values = [] for controller in CONTROLLERS: row = by_controller[controller] value = row.get(f"{metric}_mean") deviation = row.get(f"{metric}_std") if value is None: values.append("unavailable") elif deviation: values.append(f"{value:.3f} ± {deviation:.3f}") else: values.append(f"{value:.3f}") label = metric.replace("_", " ").title() handle.write(f"| {label} | " + " | ".join(values) + " |\n") handle.write("\nMetrics are raw SUMO/TraCI measurements; no unavailable values were fabricated.\n") def _write_plots(output_dir, summary): try: import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt except Exception as exc: (output_dir / "plots_unavailable.txt").write_text(str(exc), encoding="utf-8") return False labels = ["Fixed-Time", "Density-Based", "Predictive", "Integrated AI"] for metric in ( "average_waiting_time", "average_queue_length", "throughput", "average_travel_time", "fuel_consumption_mg", "co2_emissions_mg", ): values = [ row.get(f"{metric}_mean") or 0.0 for row in summary ] figure, axis = plt.subplots(figsize=(8, 4.5)) axis.bar(labels, values) axis.set_title(metric.replace("_", " ").title()) axis.set_ylabel("SUMO units") axis.tick_params(axis="x", rotation=20) figure.tight_layout() figure.savefig(output_dir / f"{metric}.png", dpi=140) plt.close(figure) return True def run_evaluation(output_dir, seeds=(1,), config=None, steps=200, warmup=10, control_interval=10.0, make_plots=True): config = Path(config or PROJECT_ROOT / "simulation" / "configs" / "corridor.sumocfg") output_dir = Path(output_dir) raw_dir = output_dir / "raw" summary_dir = output_dir / "summaries" plots_dir = output_dir / "plots" raw_dir.mkdir(parents=True, exist_ok=True) summary_dir.mkdir(parents=True, exist_ok=True) plots_dir.mkdir(parents=True, exist_ok=True) results = [] for seed in seeds: for controller_name in CONTROLLERS: print(f"\n=== {controller_name} seed={seed} ===") result = run_one(controller_name, seed, config, steps, warmup, control_interval) results.append(result) if result["status"] == "failed": print(f"FAILED: {result['error']}") else: print(f"completed: throughput={result['throughput']}") (raw_dir / f"{controller_name}_seed{seed}.json").write_text( json.dumps(result, indent=2), encoding="utf-8" ) summary = aggregate_results(results) _write_csv(output_dir / "raw_results.csv", results) _write_csv(summary_dir / "summary.csv", summary) (summary_dir / "summary.json").write_text(json.dumps(summary, indent=2), encoding="utf-8") _write_summary_markdown(summary_dir / "summary.md", summary) plots_created = _write_plots(plots_dir, summary) if make_plots else False (output_dir / "manifest.json").write_text( json.dumps({ "config": str(config), "seeds": list(seeds), "controllers": list(CONTROLLERS), "steps": steps, "warmup_steps": warmup, "control_interval": control_interval, "plots_created": plots_created, "metric_definitions": { "average_waiting_time": "completed-vehicle accumulated waiting time mean", "total_waiting_time": "completed-vehicle accumulated waiting time sum", "maximum_waiting_time": "maximum observed accumulated waiting time", "average_queue_length": "time-average active vehicles at speed <= 0.1 m/s", "maximum_queue_length": "maximum active vehicles at speed <= 0.1 m/s", "throughput": "number of arrived vehicles", "average_travel_time": "mean arrival time minus departure time", "total_travel_time": "sum of completed-vehicle travel times", "number_of_stops": "speed transition from >0.1 to <=0.1 m/s", "fuel_consumption_mg": "time-integrated SUMO fuel rate", "co2_emissions_mg": "time-integrated SUMO CO2 rate", }, }, indent=2), encoding="utf-8" ) return results, summary def main(argv=None): parser = argparse.ArgumentParser(description="Run the Phase 2.1 SUMO evaluation") parser.add_argument("--output-dir", default=None) parser.add_argument("--seeds", nargs="+", type=int, default=[1]) parser.add_argument("--steps", type=int, default=200) parser.add_argument("--warmup", type=int, default=10) parser.add_argument("--control-interval", type=float, default=10.0) parser.add_argument("--no-plots", action="store_true") args = parser.parse_args(argv) output_dir = args.output_dir or str( PROJECT_ROOT / "results" / "evaluation" / datetime.now().strftime("run_%Y%m%d_%H%M%S") ) results, summary = run_evaluation( output_dir, seeds=args.seeds, steps=args.steps, warmup=args.warmup, control_interval=args.control_interval, make_plots=not args.no_plots, ) print(f"\nResults: {output_dir}") print(f"Runs: {len(results)}; successful: {sum(r['status'] == 'success' for r in results)}") print((Path(output_dir) / "summaries" / "summary.md").read_text(encoding="utf-8")) return 0 if all(r["status"] == "success" for r in results) else 1 if __name__ == "__main__": raise SystemExit(main())