Spaces:
Sleeping
Sleeping
File size: 2,792 Bytes
1b72fa2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 | from __future__ import annotations
from pathlib import Path
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
def _ensure_dir(path: str | Path) -> Path:
p = Path(path)
p.mkdir(parents=True, exist_ok=True)
return p
def plot_training_history(history: dict[str, list[float]], output_dir: str | Path = "outputs") -> dict[str, str]:
out = _ensure_dir(output_dir)
fig, axes = plt.subplots(2, 2, figsize=(12, 8), constrained_layout=True)
axes[0, 0].plot(history.get("episode_reward", []), color="#1f77b4")
axes[0, 0].set_title("Episode Reward")
axes[0, 0].set_xlabel("Episode")
axes[0, 1].plot(history.get("avg_queue", []), color="#d62728")
axes[0, 1].set_title("Average Queue Length")
axes[0, 1].set_xlabel("Episode")
axes[1, 0].plot(history.get("throughput", []), color="#2ca02c")
axes[1, 0].set_title("Throughput")
axes[1, 0].set_xlabel("Episode")
axes[1, 1].plot(history.get("epsilon", []), color="#9467bd")
axes[1, 1].set_title("Exploration (Epsilon)")
axes[1, 1].set_xlabel("Episode")
path = out / "training_history.png"
fig.savefig(path, dpi=160)
plt.close(fig)
return {"training_history": str(path)}
def plot_comparison(
baseline: dict[str, float],
rl: dict[str, float],
output_dir: str | Path = "outputs",
) -> dict[str, str]:
out = _ensure_dir(output_dir)
metrics = ["avg_waiting_time", "avg_queue_length", "throughput", "ambulance_clearances"]
labels = ["Avg Wait", "Avg Queue", "Throughput", "Ambulance Clears"]
x = range(len(metrics))
width = 0.35
fig, ax = plt.subplots(figsize=(10, 5), constrained_layout=True)
ax.bar([i - width / 2 for i in x], [baseline[m] for m in metrics], width, label="Fixed")
ax.bar([i + width / 2 for i in x], [rl[m] for m in metrics], width, label="RL")
ax.set_xticks(list(x))
ax.set_xticklabels(labels)
ax.set_title("RL vs Fixed-Time Controller")
ax.legend()
path = out / "policy_comparison.png"
fig.savefig(path, dpi=160)
plt.close(fig)
return {"policy_comparison": str(path)}
def plot_trajectory(trace: dict[str, list[float]], output_dir: str | Path = "outputs", name: str = "trajectory") -> dict[str, str]:
out = _ensure_dir(output_dir)
fig, axes = plt.subplots(3, 1, figsize=(10, 8), constrained_layout=True)
axes[0].plot(trace.get("queue_sum", []), color="#ff7f0e")
axes[0].set_title("Queue Sum")
axes[1].plot(trace.get("phase", []), color="#17becf")
axes[1].set_title("Signal Phase")
axes[2].plot(trace.get("reward", []), color="#1f77b4")
axes[2].set_title("Reward")
axes[2].set_xlabel("Step")
path = out / f"{name}.png"
fig.savefig(path, dpi=160)
plt.close(fig)
return {name: str(path)}
|