Spaces:
Sleeping
Sleeping
File size: 3,286 Bytes
30bdd62 | 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 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | """Deterministic benchmark runner for EcoGrid policies."""
from __future__ import annotations
import argparse
import json
import random
from pathlib import Path
from statistics import mean
from baseline import heuristic_agent
from env.environment import EcoGridEnv
from env.tasks import (
BasicGridBalanceGrader,
CarbonConstrainedGrader,
RenewableVariabilityGrader,
)
from models.schemas import GridAction
TASKS = ("easy", "medium", "hard")
AGENTS = ("random", "heuristic")
# Historical reference numbers from pre-fix evaluation snapshot.
REFERENCE_MEAN = {
"easy": {"random": 0.269, "heuristic": 0.748},
"medium": {"random": 0.251, "heuristic": 0.376},
"hard": {"random": 0.001, "heuristic": 0.001},
}
def grade_episode(task: str, episode_log):
if task == "easy":
return BasicGridBalanceGrader.grade(episode_log).score
if task == "medium":
return RenewableVariabilityGrader.grade(episode_log).score
return CarbonConstrainedGrader.grade(episode_log).score
def choose_action(agent: str, task: str, state, rng: random.Random) -> GridAction:
if agent == "heuristic":
return heuristic_agent(state, task)
renewable_ratio = rng.random()
fossil_ratio = rng.random() * (1.0 - renewable_ratio)
battery_action = rng.uniform(-1.0, 1.0)
return GridAction(
renewable_ratio=renewable_ratio,
fossil_ratio=fossil_ratio,
battery_action=battery_action,
)
def run_episode(task: str, agent: str, seed: int) -> float:
rng = random.Random(seed)
env = EcoGridEnv()
state = env.reset(task=task, seed=seed)
while not env.is_done:
action = choose_action(agent, task, state, rng)
result = env.step(action)
state = result.observation
return grade_episode(task, env.get_episode_log())
def run_benchmarks(seeds: list[int]) -> dict:
out = {
"metadata": {"seeds": seeds, "agents": list(AGENTS), "tasks": list(TASKS)},
"results": {},
}
for task in TASKS:
out["results"][task] = {}
for agent in AGENTS:
scores = [run_episode(task, agent, seed) for seed in seeds]
avg = float(mean(scores))
ref = REFERENCE_MEAN[task][agent]
out["results"][task][agent] = {
"scores": [round(x, 6) for x in scores],
"mean": round(avg, 6),
"reference_mean": ref,
"delta_vs_reference": round(avg - ref, 6),
}
return out
def main():
parser = argparse.ArgumentParser(description="Run EcoGrid reproducible benchmark suite.")
parser.add_argument("--seeds", default="1,2,3,4,5", help="Comma-separated integer seeds")
parser.add_argument(
"--out",
default="logs/benchmark_results.json",
help="Path to save benchmark results JSON",
)
args = parser.parse_args()
seeds = [int(x.strip()) for x in args.seeds.split(",") if x.strip()]
results = run_benchmarks(seeds)
out_path = Path(args.out)
out_path.parent.mkdir(parents=True, exist_ok=True)
out_path.write_text(json.dumps(results, indent=2), encoding="utf-8")
print(json.dumps(results, indent=2))
print(f"\nSaved benchmark report to {out_path}")
if __name__ == "__main__":
main()
|