| |
| """ |
| Historical replay simulation through the full control loop. |
| |
| Reads IMS weather history and replays it through the control pipeline: |
| DayAheadPlanner → ControlLoop.tick() → log results |
| |
| Usage |
| ----- |
| # Replay a hot July week (default) |
| python scripts/run_control_simulation.py |
| |
| # Specify dates |
| python scripts/run_control_simulation.py --start 2025-07-01 --end 2025-07-07 |
| |
| # Dry run with verbose logging |
| python scripts/run_control_simulation.py --verbose |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import logging |
| import sys |
| from datetime import date, datetime, timedelta, timezone |
| from pathlib import Path |
|
|
| import pandas as pd |
|
|
| |
| sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) |
|
|
| from config.settings import ( |
| DP_SLOT_DURATION_MIN, |
| SIMULATION_LOG_DIR, |
| ) |
|
|
| logger = logging.getLogger("simulation") |
|
|
|
|
| |
| |
| |
|
|
| class ControlSimulation: |
| """Replay historical weather through the control pipeline. |
| |
| Parameters |
| ---------- |
| start_date : date |
| First day to simulate. |
| end_date : date |
| Last day to simulate (inclusive). |
| """ |
|
|
| def __init__(self, start_date: date, end_date: date): |
| self.start_date = start_date |
| self.end_date = end_date |
| self.results: list[dict] = [] |
|
|
| def run(self) -> list[dict]: |
| """Run the full simulation.""" |
| from src.control_loop import ControlLoop |
|
|
| loop = ControlLoop(dry_run=True) |
|
|
| current = self.start_date |
| day_count = 0 |
|
|
| while current <= self.end_date: |
| day_count += 1 |
| logger.info("=== Day %d: %s ===", day_count, current) |
|
|
| |
| plan = loop.load_plan(current) |
| if plan is None: |
| logger.warning("No plan for %s — skipping", current) |
| current += timedelta(days=1) |
| continue |
|
|
| n_slots = len(plan.get("slots", [])) |
| logger.info("Plan loaded: %d slots, stage=%s", |
| n_slots, plan.get("stage_id", "?")) |
|
|
| |
| for slot_idx in range(96): |
| hour = slot_idx // 4 |
| minute = (slot_idx % 4) * DP_SLOT_DURATION_MIN |
| ts = datetime( |
| current.year, current.month, current.day, |
| hour, minute, 0, |
| tzinfo=timezone.utc, |
| ) |
|
|
| try: |
| result = loop.tick(timestamp=ts) |
| entry = result.to_dict() |
| entry["sim_day"] = str(current) |
| self.results.append(entry) |
| except Exception as exc: |
| logger.error("Tick failed at %s: %s", ts, exc) |
| self.results.append({ |
| "timestamp": ts.isoformat(), |
| "slot_index": slot_idx, |
| "sim_day": str(current), |
| "error": str(exc), |
| }) |
|
|
| |
| day_ticks = [r for r in self.results if r.get("sim_day") == str(current)] |
| interventions = sum( |
| 1 for r in day_ticks |
| if (r.get("plan_offset_deg") or 0) > 0 |
| ) |
| overrides = sum(1 for r in day_ticks if r.get("live_override")) |
| logger.info( |
| "Day %s: %d ticks, %d interventions, %d live overrides", |
| current, len(day_ticks), interventions, overrides, |
| ) |
|
|
| current += timedelta(days=1) |
|
|
| logger.info("Simulation complete: %d total ticks over %d days", |
| len(self.results), day_count) |
| return self.results |
|
|
| def save(self, path: Path | None = None) -> Path: |
| """Save results to JSON.""" |
| if path is None: |
| SIMULATION_LOG_DIR.mkdir(parents=True, exist_ok=True) |
| path = SIMULATION_LOG_DIR / ( |
| f"sim_{self.start_date}_{self.end_date}.json" |
| ) |
| path.parent.mkdir(parents=True, exist_ok=True) |
| with open(path, "w") as f: |
| json.dump(self.results, f, indent=2, default=str) |
| logger.info("Saved %d results to %s", len(self.results), path) |
| return path |
|
|
| def print_summary(self) -> None: |
| """Print a human-readable summary.""" |
| if not self.results: |
| print("No results to summarise.") |
| return |
|
|
| total = len(self.results) |
| errors = sum(1 for r in self.results if "error" in r) |
| interventions = sum( |
| 1 for r in self.results |
| if (r.get("plan_offset_deg") or 0) > 0 |
| ) |
| overrides = sum(1 for r in self.results if r.get("live_override")) |
| dispatched = sum(1 for r in self.results if r.get("dispatch")) |
|
|
| print(f"\n{'='*60}") |
| print(f"Simulation: {self.start_date} → {self.end_date}") |
| print(f"{'='*60}") |
| print(f"Total ticks: {total}") |
| print(f"Errors: {errors}") |
| print(f"Interventions: {interventions} ({interventions/total*100:.1f}%)") |
| print(f"Live overrides: {overrides}") |
| print(f"Dispatched: {dispatched}") |
|
|
| |
| temps = [r["air_temp_c"] for r in self.results |
| if r.get("air_temp_c") is not None] |
| if temps: |
| print(f"\nTemperature: {min(temps):.1f}°C – {max(temps):.1f}°C " |
| f"(mean {sum(temps)/len(temps):.1f}°C)") |
|
|
| |
| days = sorted(set(r.get("sim_day", "") for r in self.results)) |
| print(f"\nPer-day breakdown:") |
| print(f"{'Date':<12} {'Ticks':>6} {'Interv':>7} {'Override':>9} {'MaxOff':>7}") |
| for day in days: |
| day_r = [r for r in self.results if r.get("sim_day") == day] |
| d_interv = sum(1 for r in day_r if (r.get("plan_offset_deg") or 0) > 0) |
| d_override = sum(1 for r in day_r if r.get("live_override")) |
| offsets = [r.get("plan_offset_deg", 0) or 0 for r in day_r] |
| max_off = max(offsets) if offsets else 0 |
| print(f"{day:<12} {len(day_r):>6} {d_interv:>7} {d_override:>9} {max_off:>6.0f}°") |
|
|
| print(f"{'='*60}\n") |
|
|
|
|
| |
| |
| |
|
|
| def main(): |
| parser = argparse.ArgumentParser( |
| description="Replay historical weather through the agrivoltaic control loop." |
| ) |
| _today = date.today() |
| _default_start = (_today - timedelta(days=7)).isoformat() |
| _default_end = _today.isoformat() |
| parser.add_argument( |
| "--start", type=str, default=_default_start, |
| help=f"Start date (YYYY-MM-DD). Default: {_default_start} (today − 7 d)", |
| ) |
| parser.add_argument( |
| "--end", type=str, default=_default_end, |
| help=f"End date (YYYY-MM-DD). Default: {_default_end} (today)", |
| ) |
| parser.add_argument( |
| "--output", type=str, default=None, |
| help="Output JSON path (default: Data/simulation_logs/sim_<start>_<end>.json)", |
| ) |
| parser.add_argument( |
| "--verbose", "-v", action="store_true", |
| help="Enable debug logging", |
| ) |
| args = parser.parse_args() |
|
|
| level = logging.DEBUG if args.verbose else logging.INFO |
| logging.basicConfig( |
| level=level, |
| format="%(asctime)s %(name)-15s %(levelname)-7s %(message)s", |
| datefmt="%H:%M:%S", |
| ) |
|
|
| start = date.fromisoformat(args.start) |
| end = date.fromisoformat(args.end) |
|
|
| sim = ControlSimulation(start, end) |
| sim.run() |
| sim.print_summary() |
|
|
| out_path = Path(args.output) if args.output else None |
| saved = sim.save(out_path) |
| print(f"Results saved to: {saved}") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
