Grid Simulator Mechanics

{
  "time": 42,                       # Timestep index (0..287)
  "demand": 145.2,                  # Current city demand (MW)   - this step only
  "solar": 80.5,                    # Current solar generation (MW) - this step only
  "soc": 0.45,                      # Battery state of charge (0..1) - this step only
  "price": 120.0,                   # Current import tariff ($/MWh)
  "features": {                     # Which mechanics are live this scenario
    "battery": True, "fcas": True, "ids": True
  },
  "forecast": {                     # Lookahead arrays (≈16 steps). NOISY - see Tips.
    "demand": [146.1, 150.2, ...],
    "solar":  [82.1, 85.0, ...],
    "price":  [125.0, 150.0, ...]
  },
  "alerts": [                       # Narrative events active RIGHT NOW (see Reacting to Events)
    {"id": "ids_w1", "type": "qualitative_alert", "severity": "critical",
     "title": "IDS Alert wave 1", "description": "SECURITY BREACH ...",
     "at_step": 30, "end_step": 42}
  ],
  "fcas_events_upcoming": [         # Scheduled FCAS dispatch calls (pre-position SOC!)
    {"at_step": 152, "end_step": 154, "magnitude_mw": 18.0}
  ],
  "ids_signal_node_a": 0.87,        # Attack-probability hint [0..1] - only if you subscribed,
  "ids_signal_node_b": 0.64,        #   otherwise None (see Cyber & Phishing)
  "battery_throughput_remaining_mwh": 500.0,   # If a throughput budget is active
  "peak_import_mw": 92.0,           # Your biggest grid import so far (drives the demand charge)
  "agent_plan": { ... }             # Whatever your plan()/replan() returned (see Agentic)
}

You control the grid for the upcoming timestep by returning a dictionary with any of these keys. If you leave a key out, it just defaults to 0.

def controller(state):
    # Your logic here...

    return {
        # Positive = discharge to grid, Negative = charge from grid.
        # Bounded by: inverter limit, SOC, and throughput budget.
        "battery_flow_mw": 10.5,

        # MW of emergency diesel generation to dispatch [0, max_limit]
        "emergency_generator": 0.0,

        # MW of solar to intentionally disconnect (prevents overvoltage)
        "curtail_solar": 0.0,

        # MW of inverter capacity to lock for frequency control.
        # This capacity cannot be used for charging/discharging this step!
        "fcas_reserve_mw": 5.0,

        # Pay for the IDS probability signal THIS step (cyber scenarios).
        # Populates state["ids_signal_node_a"/"_b"] on the NEXT step.
        "subscribe_ids": False,

        # Per-step acknowledgements the engine reads from YOUR ACTION.
        # (containment_ack / anomaly_ack - see Cyber & Phishing.)
        "agent_plan": {},
    }

price_history = []          # defined at the TOP of the file = "module level"

def controller(state):
    price_history.append(state["price"])         # appending is fine, no keyword needed
    recent = price_history[-12:]
    avg = sum(recent) / len(recent)
    return {"battery_flow_mw": 20.0 if state["price"] > avg else -20.0}

This works because your file is imported once, so the module-level price_history is shared across every call.

import re

class Strategy:
    def __init__(self):
        self.seen = set()
        self.attacks = {}        # attack_id -> (start_step, end_step), learned from prose
        self.ema_a = 0.0         # smoothed IDS node A
        self.ema_b = 0.0         # smoothed IDS node B

    def replan(self, state, alerts):
        # Parse each NEW alert once. A critical alert names the attack id to ack;
        # a decoy alert names an id but tells you NOT to ack it.
        for a in alerts:
            if a["id"] in self.seen:
                continue
            self.seen.add(a["id"])
            m = re.search(r"`containment_ack`:\s*`([^`]+)`", a.get("description", ""))
            if m and a.get("severity") == "critical" and "decoy" not in a["description"].lower():
                self.attacks[m.group(1)] = (a.get("at_step"), a.get("end_step"))
        return {}

    def step(self, state):
        t = int(state["time"])

        # 1) Smooth the (noisy) IDS signal. Subscribe whenever an attack window is open.
        in_window = any(s <= t <= e for (s, e) in self.attacks.values())
        a = state.get("ids_signal_node_a") or 0.0
        b = state.get("ids_signal_node_b") or 0.0
        self.ema_a = 0.5 * self.ema_a + 0.5 * a
        self.ema_b = 0.5 * self.ema_b + 0.5 * b
        corroborated = self.ema_a >= 0.4 and self.ema_b >= 0.4   # BOTH nodes agree -> real

        # 2) Acknowledge ONLY a confirmed real attack, using the id from the prose.
        agent_plan = {}
        for aid, (s, e) in self.attacks.items():
            if s is not None and s <= t <= e and corroborated:
                agent_plan["containment_ack"] = aid

        # 3) When data looks spoofed, steer by the forecast instead of the live sensor.
        demand = state["demand"]; solar = state["solar"]
        fc = state.get("forecast", {})
        if corroborated and fc.get("demand") and fc.get("solar"):
            demand, solar = fc["demand"][0], fc["solar"][0]

        net = demand - solar
        flow = max(-50.0, min(50.0, net)) if state["soc"] > 0.1 else 0.0
        return {
            "battery_flow_mw": flow,
            "subscribe_ids": in_window,     # only pay for IDS while a window is open
            "agent_plan": agent_plan,       # containment_ack read from HERE
        }

Each brief is a constraint in plain English - a reserve floor, an export cap - with a step window. Parse them once in replan(), enforce the active ones in step(). The traps this scenario is built around: fractional words ("four-fifths of capacity"), rescissions ("the prior directive is RESCINDED"), and life-safety overrides that win an overlap.

import re

MIN_SOC = re.compile(r">=\s*(\d+)\s*%.*?steps?\s+(\d+)\s+through\s+(\d+)", re.I | re.S)
EXPORT  = re.compile(r"export.{0,40}?(?:cap|limit|reduced).{0,80}?(\d+)\s*MW.*?steps?\s+(\d+)\s+through\s+(\d+)", re.I | re.S)

class Strategy:
    def __init__(self):
        self.rules, self.seen = [], set()

    def replan(self, state, alerts):
        for a in alerts:
            if a["id"] in self.seen:
                continue
            self.seen.add(a["id"])
            text = a.get("title", "") + "\n" + a.get("description", "")
            pr = 100 if "life-safety" in text.lower() else 50   # life-safety wins overlaps
            m = MIN_SOC.search(text)
            if m:
                self.rules.append({"min_soc": int(m.group(1)) / 100,
                    "start": int(m.group(2)), "end": int(m.group(3)), "pr": pr})
            m = EXPORT.search(text)
            if m:
                self.rules.append({"max_export": float(m.group(1)),
                    "start": int(m.group(2)), "end": int(m.group(3)), "pr": pr})
        return {}

    def step(self, state):
        t = int(state["time"])
        active = [c for c in self.rules if c["start"] <= t <= c["end"]]
        if active:                                   # only the top-priority tier applies
            top = max(c["pr"] for c in active)
            active = [c for c in active if c["pr"] == top]
        min_soc = max((c["min_soc"] for c in active if "min_soc" in c), default=None)
        # ...then hold SOC >= min_soc and cap exports at max_export this step.
        return {"battery_flow_mw": 0.0}

Regex catches the clean AEMO-style phrasings; the fractional-word and rescission cases are where an LLM parse pulls ahead - that cost gap is the lesson of this scenario.