diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..1bdfed6 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,30 @@ +# Simulator — Agent Rules + +Rules in this file apply to all AI coding agents working in this repository. + +## Energy System Encapsulation + +The `span_panel_simulator.energy` package is the **sole authority** for all +energy and power-flow calculations. This boundary was deliberately established +to replace scattered inline logic and must not be eroded. + +**Rules:** + +- The engine (`engine.py`) provides **raw measurements** to the energy module + (PV power, load power, grid status). It must never pre-compute, resolve, or + override energy scheduling, dispatch, or balance decisions. +- `PowerInputs` carries only observable state — never derived energy decisions + like BESS scheduled state. +- All BESS scheduling (charge mode logic, TOU hour resolution, islanding + overrides, forced-offline behavior) lives inside `EnergySystem.tick()` and + `BESSUnit`. The engine must not call `resolve_scheduled_state()` or read + `effective_state` to feed back into inputs. +- PV curtailment, GFE throttling, SOE enforcement, and bus balancing are + energy-module concerns. The engine consumes `SystemState` results — it does + not participate in producing them. +- New energy behaviors (e.g. demand response, rate optimization) must be added + inside the energy package, not grafted onto the engine. + +**Test discipline:** Tests drive BESS behavior through `BESSConfig` +(charge_mode, charge_hours, discharge_hours), not by injecting state into +`PowerInputs`. diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..ae2b7ab --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,4 @@ +# Simulator — Claude Rules + +All agent-wide rules are in [AGENTS.md](AGENTS.md). This file is for +Claude-specific configuration only. diff --git a/docs/superpowers/plans/2026-03-28-component-energy-system.md b/docs/superpowers/plans/2026-03-28-component-energy-system.md new file mode 100644 index 0000000..f713185 --- /dev/null +++ b/docs/superpowers/plans/2026-03-28-component-energy-system.md @@ -0,0 +1,1999 @@ +# Component-Based Energy System Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Replace the scattered energy balance logic in the simulator engine with a modular, component-based energy system that enforces conservation of energy and correctly models GFE, hybrid inverter, and islanding behavior. + +**Architecture:** Physical components (GridMeter, PVSource, BESSUnit, LoadGroup) resolve on a PanelBus in role order (LOAD → SOURCE → STORAGE → SLACK). The EnergySystem is a pure value object — constructed from config, ticked with PowerInputs, returns SystemState. The engine delegates all energy math to it. + +**Tech Stack:** Python 3.14, dataclasses, pytest. No new dependencies. + +**Spec:** `docs/superpowers/specs/2026-03-28-component-energy-system-design.md` + +--- + +## File Structure + +| File | Action | Responsibility | +|------|--------|---------------| +| `src/span_panel_simulator/energy/__init__.py` | Create | Public API re-exports | +| `src/span_panel_simulator/energy/types.py` | Create | ComponentRole enum, PowerContribution, BusState, SystemState, PowerInputs, config dataclasses | +| `src/span_panel_simulator/energy/components.py` | Create | Component base class, GridMeter, PVSource, BESSUnit, LoadGroup | +| `src/span_panel_simulator/energy/bus.py` | Create | PanelBus — role-ordered resolution, conservation check | +| `src/span_panel_simulator/energy/system.py` | Create | EnergySystem — from_config factory, tick entry point | +| `tests/test_energy/__init__.py` | Create | Test package init | +| `tests/test_energy/test_components.py` | Create | Layer 1: component unit tests | +| `tests/test_energy/test_bus.py` | Create | Layer 2: bus integration / conservation tests | +| `tests/test_energy/test_scenarios.py` | Create | Layer 3: topology / scenario tests | +| `src/span_panel_simulator/engine.py` | Modify | Wire EnergySystem into snapshot, modeling, and dashboard paths; remove old energy balance code | +| `src/span_panel_simulator/circuit.py` | Modify | Battery direction reads from SystemState instead of behavior engine | +| `src/span_panel_simulator/bsee.py` | Delete | Absorbed into BESSUnit | + +--- + +## Phase 1: Build Energy System in Isolation + +### Task 1: Types Module + +**Files:** +- Create: `src/span_panel_simulator/energy/__init__.py` +- Create: `src/span_panel_simulator/energy/types.py` +- Test: `tests/test_energy/__init__.py` +- Test: `tests/test_energy/test_components.py` + +- [ ] **Step 1: Create package structure** + +```bash +mkdir -p src/span_panel_simulator/energy tests/test_energy +``` + +- [ ] **Step 2: Write types module** + +Create `src/span_panel_simulator/energy/__init__.py`: + +```python +"""Component-based energy system for the SPAN panel simulator.""" +``` + +Create `tests/test_energy/__init__.py`: + +```python +"""Energy system tests.""" +``` + +Create `src/span_panel_simulator/energy/types.py` with all core types from the spec: +- `ComponentRole` enum: `LOAD`, `SOURCE`, `STORAGE`, `SLACK` +- `PowerContribution` dataclass: `demand_w`, `supply_w` (both non-negative) +- `BusState` dataclass: `total_demand_w`, `total_supply_w`, `storage_contribution_w`, `grid_power_w`, `net_deficit_w` property, `is_balanced()` method +- `PowerInputs` dataclass: `pv_available_w`, `bess_requested_w`, `bess_scheduled_state`, `load_demand_w`, `grid_connected` +- `SystemState` dataclass: `grid_power_w`, `pv_power_w`, `bess_power_w`, `bess_state`, `load_power_w`, `soe_kwh`, `soe_percentage`, `balanced` +- Config dataclasses (all frozen): `GridConfig`, `PVConfig`, `BESSConfig`, `LoadConfig`, `EnergySystemConfig` + +Implementation details from spec: +- `BusState.net_deficit_w` = `total_demand_w - total_supply_w` +- `BusState.is_balanced()` checks `abs(total_demand_w - total_supply_w - grid_power_w) < 0.01` +- `BESSConfig.initial_soe_kwh` defaults to `None` (factory computes 50% of nameplate) +- `EnergySystemConfig.loads` uses `field(default_factory=list)` + +- [ ] **Step 3: Write type smoke tests** + +Create `tests/test_energy/test_components.py` with initial type tests: + +```python +"""Layer 1: Component unit tests for the energy system.""" + +from __future__ import annotations + +from span_panel_simulator.energy.types import ( + BESSConfig, + BusState, + ComponentRole, + EnergySystemConfig, + GridConfig, + LoadConfig, + PowerContribution, + PowerInputs, + PVConfig, + SystemState, +) + + +class TestPowerContribution: + def test_default_zero(self) -> None: + pc = PowerContribution() + assert pc.demand_w == 0.0 + assert pc.supply_w == 0.0 + + def test_demand_only(self) -> None: + pc = PowerContribution(demand_w=5000.0) + assert pc.demand_w == 5000.0 + assert pc.supply_w == 0.0 + + +class TestBusState: + def test_net_deficit_positive(self) -> None: + bs = BusState(total_demand_w=5000.0, total_supply_w=3000.0) + assert bs.net_deficit_w == 2000.0 + + def test_net_deficit_negative_means_excess(self) -> None: + bs = BusState(total_demand_w=2000.0, total_supply_w=5000.0) + assert bs.net_deficit_w == -3000.0 + + def test_balanced_when_grid_absorbs_residual(self) -> None: + bs = BusState( + total_demand_w=5000.0, + total_supply_w=3000.0, + grid_power_w=2000.0, + ) + assert bs.is_balanced() + + def test_not_balanced_when_residual_exists(self) -> None: + bs = BusState( + total_demand_w=5000.0, + total_supply_w=3000.0, + grid_power_w=1000.0, + ) + assert not bs.is_balanced() + + +class TestComponentRole: + def test_role_ordering(self) -> None: + roles = [ComponentRole.SLACK, ComponentRole.LOAD, ComponentRole.STORAGE, ComponentRole.SOURCE] + sorted_roles = sorted(roles, key=lambda r: r.value) + assert sorted_roles == [ + ComponentRole.LOAD, + ComponentRole.SOURCE, + ComponentRole.STORAGE, + ComponentRole.SLACK, + ] +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `.venv/bin/python -m pytest tests/test_energy/test_components.py -v` +Expected: All pass + +- [ ] **Step 5: Commit** + +```bash +git add src/span_panel_simulator/energy/ tests/test_energy/ +git commit -m "Add energy system types module with core dataclasses" +``` + +--- + +### Task 2: GridMeter and LoadGroup Components + +**Files:** +- Create: `src/span_panel_simulator/energy/components.py` +- Test: `tests/test_energy/test_components.py` + +- [ ] **Step 1: Write failing tests for GridMeter and LoadGroup** + +Append to `tests/test_energy/test_components.py`: + +```python +from span_panel_simulator.energy.components import GridMeter, LoadGroup + + +class TestLoadGroup: + def test_returns_demand(self) -> None: + load = LoadGroup(demand_w=5000.0) + contribution = load.resolve(BusState()) + assert contribution.demand_w == 5000.0 + assert contribution.supply_w == 0.0 + + def test_zero_demand(self) -> None: + load = LoadGroup(demand_w=0.0) + contribution = load.resolve(BusState()) + assert contribution.demand_w == 0.0 + + +class TestGridMeter: + def test_absorbs_deficit(self) -> None: + grid = GridMeter(connected=True) + bus = BusState(total_demand_w=5000.0, total_supply_w=3000.0) + contribution = grid.resolve(bus) + assert contribution.supply_w == 2000.0 + assert contribution.demand_w == 0.0 + + def test_absorbs_excess(self) -> None: + grid = GridMeter(connected=True) + bus = BusState(total_demand_w=2000.0, total_supply_w=5000.0) + contribution = grid.resolve(bus) + assert contribution.demand_w == 3000.0 + assert contribution.supply_w == 0.0 + + def test_zero_when_balanced(self) -> None: + grid = GridMeter(connected=True) + bus = BusState(total_demand_w=3000.0, total_supply_w=3000.0) + contribution = grid.resolve(bus) + assert contribution.demand_w == 0.0 + assert contribution.supply_w == 0.0 + + def test_zero_when_disconnected(self) -> None: + grid = GridMeter(connected=False) + bus = BusState(total_demand_w=5000.0, total_supply_w=0.0) + contribution = grid.resolve(bus) + assert contribution.demand_w == 0.0 + assert contribution.supply_w == 0.0 +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `.venv/bin/python -m pytest tests/test_energy/test_components.py::TestGridMeter -v` +Expected: FAIL with `ImportError` + +- [ ] **Step 3: Implement GridMeter and LoadGroup** + +Create `src/span_panel_simulator/energy/components.py`: + +```python +"""Physical energy components for the panel bus. + +Each component has a role (LOAD, SOURCE, STORAGE, SLACK) and implements +``resolve()`` which returns a ``PowerContribution`` given the current +``BusState``. All power values are non-negative magnitudes; direction +is expressed by which field (``demand_w`` vs ``supply_w``) is populated. +""" + +from __future__ import annotations + +from span_panel_simulator.energy.types import ( + BusState, + ComponentRole, + PowerContribution, +) + + +class Component: + """Base class for all bus components.""" + + role: ComponentRole + + def resolve(self, bus_state: BusState) -> PowerContribution: + raise NotImplementedError + + +class LoadGroup(Component): + """Consumer load — declares demand on the bus.""" + + role = ComponentRole.LOAD + + def __init__(self, demand_w: float = 0.0) -> None: + self.demand_w = demand_w + + def resolve(self, bus_state: BusState) -> PowerContribution: + return PowerContribution(demand_w=self.demand_w) + + +class GridMeter(Component): + """Utility grid connection — slack bus that absorbs residual.""" + + role = ComponentRole.SLACK + + def __init__(self, connected: bool = True) -> None: + self.connected = connected + + def resolve(self, bus_state: BusState) -> PowerContribution: + if not self.connected: + return PowerContribution() + deficit = bus_state.net_deficit_w + if deficit > 0: + return PowerContribution(supply_w=deficit) + elif deficit < 0: + return PowerContribution(demand_w=-deficit) + return PowerContribution() +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `.venv/bin/python -m pytest tests/test_energy/test_components.py -v` +Expected: All pass + +- [ ] **Step 5: Commit** + +```bash +git add src/span_panel_simulator/energy/components.py tests/test_energy/test_components.py +git commit -m "Add GridMeter and LoadGroup energy components" +``` + +--- + +### Task 3: PVSource Component + +**Files:** +- Modify: `src/span_panel_simulator/energy/components.py` +- Test: `tests/test_energy/test_components.py` + +- [ ] **Step 1: Write failing tests for PVSource** + +Append to `tests/test_energy/test_components.py`: + +```python +from span_panel_simulator.energy.components import GridMeter, LoadGroup, PVSource + + +class TestPVSource: + def test_returns_available_power_when_online(self) -> None: + pv = PVSource(available_power_w=4000.0, online=True) + contribution = pv.resolve(BusState()) + assert contribution.supply_w == 4000.0 + assert contribution.demand_w == 0.0 + + def test_returns_zero_when_offline(self) -> None: + pv = PVSource(available_power_w=4000.0, online=False) + contribution = pv.resolve(BusState()) + assert contribution.supply_w == 0.0 + assert contribution.demand_w == 0.0 + + def test_zero_production(self) -> None: + pv = PVSource(available_power_w=0.0, online=True) + contribution = pv.resolve(BusState()) + assert contribution.supply_w == 0.0 +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `.venv/bin/python -m pytest tests/test_energy/test_components.py::TestPVSource -v` +Expected: FAIL with `ImportError` + +- [ ] **Step 3: Implement PVSource** + +Add to `src/span_panel_simulator/energy/components.py`: + +```python +class PVSource(Component): + """Solar PV inverter — declares available production.""" + + role = ComponentRole.SOURCE + + def __init__(self, available_power_w: float = 0.0, online: bool = True) -> None: + self.available_power_w = available_power_w + self.online = online + + def resolve(self, bus_state: BusState) -> PowerContribution: + if not self.online: + return PowerContribution() + return PowerContribution(supply_w=self.available_power_w) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `.venv/bin/python -m pytest tests/test_energy/test_components.py -v` +Expected: All pass + +- [ ] **Step 5: Commit** + +```bash +git add src/span_panel_simulator/energy/components.py tests/test_energy/test_components.py +git commit -m "Add PVSource energy component" +``` + +--- + +### Task 4: BESSUnit Component — Discharge and GFE + +**Files:** +- Modify: `src/span_panel_simulator/energy/components.py` +- Test: `tests/test_energy/test_components.py` + +- [ ] **Step 1: Write failing tests for BESSUnit discharge behavior** + +Append to `tests/test_energy/test_components.py`: + +```python +from span_panel_simulator.energy.components import BESSUnit, GridMeter, LoadGroup, PVSource + + +class TestBESSUnitDischarge: + def _make_bess( + self, + *, + nameplate_kwh: float = 13.5, + max_discharge_w: float = 5000.0, + max_charge_w: float = 3500.0, + soe_kwh: float = 10.0, + backup_reserve_pct: float = 20.0, + hard_min_pct: float = 5.0, + scheduled_state: str = "discharging", + requested_power_w: float = 5000.0, + ) -> BESSUnit: + return BESSUnit( + nameplate_capacity_kwh=nameplate_kwh, + max_charge_w=max_charge_w, + max_discharge_w=max_discharge_w, + charge_efficiency=0.95, + discharge_efficiency=0.95, + backup_reserve_pct=backup_reserve_pct, + hard_min_pct=hard_min_pct, + hybrid=False, + pv_source=None, + soe_kwh=soe_kwh, + scheduled_state=scheduled_state, + requested_power_w=requested_power_w, + ) + + def test_discharge_throttled_to_deficit(self) -> None: + """GFE: only source what loads demand.""" + bess = self._make_bess(requested_power_w=5000.0) + bus = BusState(total_demand_w=2000.0, total_supply_w=0.0) + contribution = bess.resolve(bus) + assert contribution.supply_w == 2000.0 + assert bess.effective_state == "discharging" + + def test_discharge_idle_when_no_deficit(self) -> None: + """Solar covers all loads — no discharge needed.""" + bess = self._make_bess(requested_power_w=5000.0) + bus = BusState(total_demand_w=3000.0, total_supply_w=4000.0) + contribution = bess.resolve(bus) + assert contribution.supply_w == 0.0 + assert bess.effective_state == "idle" + assert bess.effective_power_w == 0.0 + + def test_discharge_limited_by_max_rate(self) -> None: + bess = self._make_bess(max_discharge_w=2000.0, requested_power_w=5000.0) + bus = BusState(total_demand_w=8000.0, total_supply_w=0.0) + contribution = bess.resolve(bus) + assert contribution.supply_w == 2000.0 + + def test_discharge_limited_by_requested(self) -> None: + bess = self._make_bess(requested_power_w=1500.0) + bus = BusState(total_demand_w=5000.0, total_supply_w=0.0) + contribution = bess.resolve(bus) + assert contribution.supply_w == 1500.0 + + def test_discharge_stops_at_backup_reserve(self) -> None: + """SOE at backup reserve — cannot discharge.""" + bess = self._make_bess( + nameplate_kwh=10.0, + soe_kwh=2.0, # exactly 20% of 10 kWh + backup_reserve_pct=20.0, + ) + bus = BusState(total_demand_w=5000.0, total_supply_w=0.0) + contribution = bess.resolve(bus) + assert contribution.supply_w == 0.0 + assert bess.effective_state == "idle" + + def test_idle_state_returns_zero(self) -> None: + bess = self._make_bess(scheduled_state="idle") + bus = BusState(total_demand_w=5000.0, total_supply_w=0.0) + contribution = bess.resolve(bus) + assert contribution.supply_w == 0.0 + assert contribution.demand_w == 0.0 +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `.venv/bin/python -m pytest tests/test_energy/test_components.py::TestBESSUnitDischarge -v` +Expected: FAIL with `ImportError` + +- [ ] **Step 3: Implement BESSUnit** + +Add to `src/span_panel_simulator/energy/components.py`: + +```python +_SOE_MAX_PCT = 100.0 +_MAX_INTEGRATION_DELTA_S = 300.0 + + +class BESSUnit(Component): + """Battery Energy Storage System — GFE-aware storage component. + + When discharging, the BESS only sources what the bus actually demands + (GFE constraint). Enforces SOE bounds, max charge/discharge rates, + and efficiency losses. Controls co-located PV online status when + configured as a hybrid inverter. + """ + + role = ComponentRole.STORAGE + + def __init__( + self, + *, + nameplate_capacity_kwh: float, + max_charge_w: float, + max_discharge_w: float, + charge_efficiency: float, + discharge_efficiency: float, + backup_reserve_pct: float, + hard_min_pct: float, + hybrid: bool, + pv_source: PVSource | None, + soe_kwh: float, + scheduled_state: str = "idle", + requested_power_w: float = 0.0, + ) -> None: + self.nameplate_capacity_kwh = nameplate_capacity_kwh + self.max_charge_w = max_charge_w + self.max_discharge_w = max_discharge_w + self.charge_efficiency = charge_efficiency + self.discharge_efficiency = discharge_efficiency + self.backup_reserve_pct = backup_reserve_pct + self.hard_min_pct = hard_min_pct + self.hybrid = hybrid + self.pv_source = pv_source + self.soe_kwh = soe_kwh + self.scheduled_state = scheduled_state + self.requested_power_w = requested_power_w + + # Output — set by resolve() + self.effective_power_w: float = 0.0 + self.effective_state: str = "idle" + + # Timestamp tracking for energy integration + self._last_ts: float | None = None + + @property + def soe_percentage(self) -> float: + if self.nameplate_capacity_kwh <= 0: + return 0.0 + return self.soe_kwh / self.nameplate_capacity_kwh * 100.0 + + def resolve(self, bus_state: BusState) -> PowerContribution: + if self.scheduled_state == "idle": + self.effective_power_w = 0.0 + self.effective_state = "idle" + return PowerContribution() + if self.scheduled_state == "discharging": + return self._resolve_discharge(bus_state) + if self.scheduled_state == "charging": + return self._resolve_charge(bus_state) + self.effective_power_w = 0.0 + self.effective_state = "idle" + return PowerContribution() + + def _resolve_discharge(self, bus_state: BusState) -> PowerContribution: + deficit = bus_state.net_deficit_w + if deficit <= 0: + self.effective_power_w = 0.0 + self.effective_state = "idle" + return PowerContribution() + max_for_soe = self._max_discharge_for_soe() + if max_for_soe <= 0: + self.effective_power_w = 0.0 + self.effective_state = "idle" + return PowerContribution() + power = min(self.requested_power_w, deficit, self.max_discharge_w, max_for_soe) + self.effective_power_w = power + self.effective_state = "discharging" + return PowerContribution(supply_w=power) + + def _resolve_charge(self, bus_state: BusState) -> PowerContribution: + max_for_soe = self._max_charge_for_soe() + if max_for_soe <= 0: + self.effective_power_w = 0.0 + self.effective_state = "idle" + return PowerContribution() + power = min(self.requested_power_w, self.max_charge_w, max_for_soe) + self.effective_power_w = power + self.effective_state = "charging" + return PowerContribution(demand_w=power) + + def _max_discharge_for_soe(self) -> float: + """Max discharge power before hitting SOE floor (backup reserve).""" + min_kwh = self.nameplate_capacity_kwh * self.backup_reserve_pct / 100.0 + available_kwh = self.soe_kwh - min_kwh + if available_kwh <= 0: + return 0.0 + # Convert to instantaneous watts (assuming 1-second resolution is + # conservative; actual integration uses real delta). + return available_kwh * 1000.0 * 3600.0 # effectively unlimited for a single tick + + def _max_charge_for_soe(self) -> float: + """Max charge power before hitting SOE ceiling (100%).""" + max_kwh = self.nameplate_capacity_kwh * _SOE_MAX_PCT / 100.0 + headroom_kwh = max_kwh - self.soe_kwh + if headroom_kwh <= 0: + return 0.0 + return headroom_kwh * 1000.0 * 3600.0 # effectively unlimited for a single tick + + def integrate_energy(self, ts: float) -> None: + """Integrate effective power over elapsed time to update SOE.""" + if self._last_ts is None: + self._last_ts = ts + return + + delta_s = ts - self._last_ts + self._last_ts = ts + if delta_s <= 0: + return + delta_s = min(delta_s, _MAX_INTEGRATION_DELTA_S) + delta_hours = delta_s / 3600.0 + + mag = abs(self.effective_power_w) + if self.effective_state == "charging" and mag > 0: + energy_kwh = (mag / 1000.0) * delta_hours * self.charge_efficiency + self.soe_kwh += energy_kwh + elif self.effective_state == "discharging" and mag > 0: + energy_kwh = (mag / 1000.0) * delta_hours / self.discharge_efficiency + self.soe_kwh -= energy_kwh + + max_kwh = self.nameplate_capacity_kwh * _SOE_MAX_PCT / 100.0 + min_kwh = self.nameplate_capacity_kwh * self.hard_min_pct / 100.0 + self.soe_kwh = max(min_kwh, min(max_kwh, self.soe_kwh)) + + def update_pv_online_status(self, grid_connected: bool) -> None: + """Control co-located PV based on hybrid inverter capability.""" + if self.pv_source is None: + return + if self.hybrid: + self.pv_source.online = True + elif not grid_connected: + self.pv_source.online = False + else: + self.pv_source.online = True +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `.venv/bin/python -m pytest tests/test_energy/test_components.py -v` +Expected: All pass + +- [ ] **Step 5: Commit** + +```bash +git add src/span_panel_simulator/energy/components.py tests/test_energy/test_components.py +git commit -m "Add BESSUnit component with GFE discharge throttling" +``` + +--- + +### Task 5: BESSUnit — Charging, SOE Integration, and Hybrid PV Control + +**Files:** +- Test: `tests/test_energy/test_components.py` + +- [ ] **Step 1: Write failing tests for charging, SOE, and hybrid PV** + +Append to `tests/test_energy/test_components.py`: + +```python +class TestBESSUnitCharge: + def _make_bess(self, **kwargs: object) -> BESSUnit: + defaults: dict[str, object] = dict( + nameplate_capacity_kwh=13.5, + max_charge_w=3500.0, + max_discharge_w=5000.0, + charge_efficiency=0.95, + discharge_efficiency=0.95, + backup_reserve_pct=20.0, + hard_min_pct=5.0, + hybrid=False, + pv_source=None, + soe_kwh=6.75, + scheduled_state="charging", + requested_power_w=3000.0, + ) + defaults.update(kwargs) + return BESSUnit(**defaults) + + def test_charge_at_requested_rate(self) -> None: + bess = self._make_bess(requested_power_w=2000.0) + bus = BusState(total_demand_w=1000.0, total_supply_w=5000.0) + contribution = bess.resolve(bus) + assert contribution.demand_w == 2000.0 + assert bess.effective_state == "charging" + + def test_charge_limited_by_max_rate(self) -> None: + bess = self._make_bess(max_charge_w=1500.0, requested_power_w=3000.0) + bus = BusState() + contribution = bess.resolve(bus) + assert contribution.demand_w == 1500.0 + + def test_charge_stops_at_full_soe(self) -> None: + bess = self._make_bess(nameplate_capacity_kwh=10.0, soe_kwh=10.0) + bus = BusState() + contribution = bess.resolve(bus) + assert contribution.demand_w == 0.0 + assert bess.effective_state == "idle" + + +class TestBESSUnitSOEIntegration: + def _make_bess(self, **kwargs: object) -> BESSUnit: + defaults: dict[str, object] = dict( + nameplate_capacity_kwh=10.0, + max_charge_w=3500.0, + max_discharge_w=5000.0, + charge_efficiency=0.95, + discharge_efficiency=0.95, + backup_reserve_pct=20.0, + hard_min_pct=5.0, + hybrid=False, + pv_source=None, + soe_kwh=5.0, + scheduled_state="idle", + requested_power_w=0.0, + ) + defaults.update(kwargs) + return BESSUnit(**defaults) + + def test_discharge_decreases_soe(self) -> None: + bess = self._make_bess( + soe_kwh=5.0, + scheduled_state="discharging", + requested_power_w=2000.0, + ) + bus = BusState(total_demand_w=5000.0, total_supply_w=0.0) + bess.resolve(bus) + # First tick establishes timestamp + bess.integrate_energy(1000.0) + # Second tick integrates over 1 hour + bess.integrate_energy(4600.0) # 3600 seconds later + # 2000W for 1 hour / 0.95 efficiency = ~2.105 kWh consumed + assert bess.soe_kwh < 5.0 + expected = 5.0 - (2.0 / 0.95) + assert abs(bess.soe_kwh - expected) < 0.01 + + def test_charge_increases_soe(self) -> None: + bess = self._make_bess( + soe_kwh=3.0, + scheduled_state="charging", + requested_power_w=2000.0, + ) + bus = BusState() + bess.resolve(bus) + bess.integrate_energy(1000.0) + bess.integrate_energy(4600.0) + # 2000W for 1 hour * 0.95 efficiency = 1.9 kWh stored + expected = 3.0 + (2.0 * 0.95) + assert abs(bess.soe_kwh - expected) < 0.01 + + def test_soe_clamped_to_bounds(self) -> None: + bess = self._make_bess(nameplate_capacity_kwh=10.0, soe_kwh=9.9) + bess.effective_state = "charging" + bess.effective_power_w = 50000.0 # absurdly high + bess.integrate_energy(0.0) + bess.integrate_energy(3600.0) + assert bess.soe_kwh <= 10.0 + + +class TestBESSUnitHybridPV: + def test_hybrid_keeps_pv_online_when_grid_disconnected(self) -> None: + pv = PVSource(available_power_w=4000.0, online=True) + bess = BESSUnit( + nameplate_capacity_kwh=13.5, + max_charge_w=3500.0, + max_discharge_w=5000.0, + charge_efficiency=0.95, + discharge_efficiency=0.95, + backup_reserve_pct=20.0, + hard_min_pct=5.0, + hybrid=True, + pv_source=pv, + soe_kwh=6.75, + ) + bess.update_pv_online_status(grid_connected=False) + assert pv.online is True + + def test_non_hybrid_sheds_pv_when_grid_disconnected(self) -> None: + pv = PVSource(available_power_w=4000.0, online=True) + bess = BESSUnit( + nameplate_capacity_kwh=13.5, + max_charge_w=3500.0, + max_discharge_w=5000.0, + charge_efficiency=0.95, + discharge_efficiency=0.95, + backup_reserve_pct=20.0, + hard_min_pct=5.0, + hybrid=False, + pv_source=pv, + soe_kwh=6.75, + ) + bess.update_pv_online_status(grid_connected=False) + assert pv.online is False + + def test_non_hybrid_pv_online_when_grid_connected(self) -> None: + pv = PVSource(available_power_w=4000.0, online=False) + bess = BESSUnit( + nameplate_capacity_kwh=13.5, + max_charge_w=3500.0, + max_discharge_w=5000.0, + charge_efficiency=0.95, + discharge_efficiency=0.95, + backup_reserve_pct=20.0, + hard_min_pct=5.0, + hybrid=False, + pv_source=pv, + soe_kwh=6.75, + ) + bess.update_pv_online_status(grid_connected=True) + assert pv.online is True +``` + +- [ ] **Step 2: Run tests to verify they pass** + +Run: `.venv/bin/python -m pytest tests/test_energy/test_components.py -v` +Expected: All pass (implementation already done in Task 4) + +- [ ] **Step 3: Commit** + +```bash +git add tests/test_energy/test_components.py +git commit -m "Add BESSUnit charge, SOE integration, and hybrid PV tests" +``` + +--- + +### Task 6: PanelBus + +**Files:** +- Create: `src/span_panel_simulator/energy/bus.py` +- Test: `tests/test_energy/test_bus.py` + +- [ ] **Step 1: Write failing bus integration tests** + +Create `tests/test_energy/test_bus.py`: + +```python +"""Layer 2: Bus integration tests — conservation enforcement.""" + +from __future__ import annotations + +from span_panel_simulator.energy.bus import PanelBus +from span_panel_simulator.energy.components import BESSUnit, GridMeter, LoadGroup, PVSource + + +def _make_bess(**kwargs: object) -> BESSUnit: + defaults: dict[str, object] = dict( + nameplate_capacity_kwh=13.5, + max_charge_w=3500.0, + max_discharge_w=5000.0, + charge_efficiency=0.95, + discharge_efficiency=0.95, + backup_reserve_pct=20.0, + hard_min_pct=5.0, + hybrid=False, + pv_source=None, + soe_kwh=10.0, + scheduled_state="idle", + requested_power_w=0.0, + ) + defaults.update(kwargs) + return BESSUnit(**defaults) + + +class TestBusConservation: + def test_load_only_grid_covers(self) -> None: + bus = PanelBus( + components=[LoadGroup(demand_w=5000.0), GridMeter(connected=True)] + ) + state = bus.resolve() + assert state.is_balanced() + assert state.grid_power_w == 5000.0 + + def test_load_and_pv_grid_covers_deficit(self) -> None: + bus = PanelBus( + components=[ + LoadGroup(demand_w=5000.0), + PVSource(available_power_w=3000.0, online=True), + GridMeter(connected=True), + ] + ) + state = bus.resolve() + assert state.is_balanced() + assert abs(state.grid_power_w - 2000.0) < 0.01 + + def test_pv_exceeds_load_grid_absorbs(self) -> None: + bus = PanelBus( + components=[ + LoadGroup(demand_w=2000.0), + PVSource(available_power_w=5000.0, online=True), + GridMeter(connected=True), + ] + ) + state = bus.resolve() + assert state.is_balanced() + # Grid absorbs 3kW excess (negative = exporting/absorbing) + assert abs(state.grid_power_w - (-3000.0)) < 0.01 + # Demand/supply only reflect non-SLACK components + assert abs(state.total_demand_w - 2000.0) < 0.01 + assert abs(state.total_supply_w - 5000.0) < 0.01 + + def test_bess_discharge_reduces_grid(self) -> None: + bess = _make_bess(scheduled_state="discharging", requested_power_w=3000.0) + bus = PanelBus( + components=[ + LoadGroup(demand_w=5000.0), + bess, + GridMeter(connected=True), + ] + ) + state = bus.resolve() + assert state.is_balanced() + assert abs(state.grid_power_w - 2000.0) < 0.01 + + def test_bess_charge_increases_grid(self) -> None: + bess = _make_bess(scheduled_state="charging", requested_power_w=3000.0, soe_kwh=5.0) + bus = PanelBus( + components=[ + LoadGroup(demand_w=2000.0), + bess, + GridMeter(connected=True), + ] + ) + state = bus.resolve() + assert state.is_balanced() + assert abs(state.grid_power_w - 5000.0) < 0.01 + + def test_grid_never_negative_from_bess_discharge(self) -> None: + bess = _make_bess(scheduled_state="discharging", requested_power_w=5000.0) + bus = PanelBus( + components=[ + LoadGroup(demand_w=2000.0), + PVSource(available_power_w=1000.0, online=True), + bess, + GridMeter(connected=True), + ] + ) + state = bus.resolve() + assert state.is_balanced() + assert state.grid_power_w >= -0.01 # never negative + + def test_conservation_grid_disconnected(self) -> None: + bess = _make_bess(scheduled_state="discharging", requested_power_w=5000.0) + bus = PanelBus( + components=[ + LoadGroup(demand_w=3000.0), + bess, + GridMeter(connected=False), + ] + ) + state = bus.resolve() + assert state.is_balanced() + assert abs(state.grid_power_w) < 0.01 + + def test_conservation_no_bess(self) -> None: + bus = PanelBus( + components=[ + LoadGroup(demand_w=5000.0), + PVSource(available_power_w=2000.0, online=True), + GridMeter(connected=True), + ] + ) + state = bus.resolve() + assert state.is_balanced() + + def test_conservation_no_pv(self) -> None: + bus = PanelBus( + components=[ + LoadGroup(demand_w=5000.0), + GridMeter(connected=True), + ] + ) + state = bus.resolve() + assert state.is_balanced() +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `.venv/bin/python -m pytest tests/test_energy/test_bus.py -v` +Expected: FAIL with `ImportError` + +- [ ] **Step 3: Implement PanelBus** + +Create `src/span_panel_simulator/energy/bus.py`: + +```python +"""PanelBus — role-ordered power flow resolution with conservation enforcement.""" + +from __future__ import annotations + +from span_panel_simulator.energy.components import Component +from span_panel_simulator.energy.types import BusState, ComponentRole + + +class PanelBus: + """Resolves components in role order and enforces conservation of energy. + + Components are evaluated in order: LOAD -> SOURCE -> STORAGE -> SLACK. + SLACK (grid) contributions are tracked in ``grid_power_w`` only — they + are NOT folded into ``total_demand_w``/``total_supply_w`` so that the + conservation check ``total_demand = total_supply + grid_power`` holds + without double-counting. + """ + + def __init__(self, components: list[Component]) -> None: + self._components = components + + def resolve(self) -> BusState: + state = BusState() + for role in (ComponentRole.LOAD, ComponentRole.SOURCE, ComponentRole.STORAGE, ComponentRole.SLACK): + for component in self._components: + if component.role != role: + continue + contribution = component.resolve(state) + if role == ComponentRole.SLACK: + # Grid tracked separately — not in demand/supply totals + state.grid_power_w += contribution.supply_w - contribution.demand_w + else: + state.total_demand_w += contribution.demand_w + state.total_supply_w += contribution.supply_w + if role == ComponentRole.STORAGE: + state.storage_contribution_w += contribution.supply_w - contribution.demand_w + return state +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `.venv/bin/python -m pytest tests/test_energy/test_bus.py -v` +Expected: All pass + +- [ ] **Step 5: Commit** + +```bash +git add src/span_panel_simulator/energy/bus.py tests/test_energy/test_bus.py +git commit -m "Add PanelBus with role-ordered resolution and conservation checks" +``` + +--- + +### Task 7: EnergySystem + +**Files:** +- Create: `src/span_panel_simulator/energy/system.py` +- Modify: `src/span_panel_simulator/energy/__init__.py` +- Test: `tests/test_energy/test_scenarios.py` + +- [ ] **Step 1: Write failing scenario tests** + +Create `tests/test_energy/test_scenarios.py`: + +```python +"""Layer 3: Topology and scenario tests covering all identified issues.""" + +from __future__ import annotations + +from span_panel_simulator.energy.system import EnergySystem +from span_panel_simulator.energy.types import ( + BESSConfig, + EnergySystemConfig, + GridConfig, + LoadConfig, + PowerInputs, + PVConfig, +) + + +def _grid_online() -> GridConfig: + return GridConfig(connected=True) + + +def _grid_offline() -> GridConfig: + return GridConfig(connected=False) + + +def _pv(nameplate_w: float = 6000.0, inverter_type: str = "ac_coupled") -> PVConfig: + return PVConfig(nameplate_w=nameplate_w, inverter_type=inverter_type) + + +def _bess( + *, + nameplate_kwh: float = 13.5, + max_discharge_w: float = 5000.0, + max_charge_w: float = 3500.0, + hybrid: bool = False, + backup_reserve_pct: float = 20.0, + initial_soe_kwh: float | None = None, +) -> BESSConfig: + return BESSConfig( + nameplate_kwh=nameplate_kwh, + max_discharge_w=max_discharge_w, + max_charge_w=max_charge_w, + hybrid=hybrid, + backup_reserve_pct=backup_reserve_pct, + initial_soe_kwh=initial_soe_kwh, + ) + + +# === GFE and Discharge Throttling (Issues 1, 2) === + + +class TestGFEThrottling: + def test_grid_never_negative_from_bess_discharge(self) -> None: + system = EnergySystem.from_config(EnergySystemConfig( + grid=_grid_online(), + pv=_pv(), + bess=_bess(), + loads=[LoadConfig(demand_w=3000.0)], + )) + state = system.tick(1000.0, PowerInputs( + pv_available_w=2000.0, + bess_scheduled_state="discharging", + bess_requested_w=5000.0, + load_demand_w=3000.0, + grid_connected=True, + )) + assert state.balanced + assert state.grid_power_w >= -0.01 + assert state.bess_power_w == 1000.0 + + def test_bess_covers_exact_deficit(self) -> None: + system = EnergySystem.from_config(EnergySystemConfig( + grid=_grid_online(), + bess=_bess(), + loads=[LoadConfig(demand_w=3000.0)], + )) + state = system.tick(1000.0, PowerInputs( + bess_scheduled_state="discharging", + bess_requested_w=5000.0, + load_demand_w=3000.0, + grid_connected=True, + )) + assert state.balanced + assert abs(state.grid_power_w) < 0.01 + assert abs(state.bess_power_w - 3000.0) < 0.01 + + def test_bess_idle_when_pv_exceeds_load(self) -> None: + system = EnergySystem.from_config(EnergySystemConfig( + grid=_grid_online(), + pv=_pv(), + bess=_bess(), + loads=[LoadConfig(demand_w=2000.0)], + )) + state = system.tick(1000.0, PowerInputs( + pv_available_w=5000.0, + bess_scheduled_state="discharging", + bess_requested_w=5000.0, + load_demand_w=2000.0, + grid_connected=True, + )) + assert state.balanced + assert state.bess_power_w == 0.0 + assert state.bess_state == "idle" + + +# === Hybrid vs Non-Hybrid Islanding (Issues 3, 5, 8) === + + +class TestIslanding: + def test_non_hybrid_pv_offline_bess_covers_all(self) -> None: + system = EnergySystem.from_config(EnergySystemConfig( + grid=_grid_offline(), + pv=_pv(inverter_type="ac_coupled"), + bess=_bess(hybrid=False), + loads=[LoadConfig(demand_w=3000.0)], + )) + state = system.tick(1000.0, PowerInputs( + pv_available_w=4000.0, + bess_scheduled_state="discharging", + bess_requested_w=5000.0, + load_demand_w=3000.0, + grid_connected=False, + )) + assert state.balanced + assert state.pv_power_w == 0.0 + assert state.bess_power_w == 3000.0 + assert state.grid_power_w == 0.0 + + def test_hybrid_pv_online_bess_covers_gap(self) -> None: + system = EnergySystem.from_config(EnergySystemConfig( + grid=_grid_offline(), + pv=_pv(inverter_type="hybrid"), + bess=_bess(hybrid=True), + loads=[LoadConfig(demand_w=5000.0)], + )) + state = system.tick(1000.0, PowerInputs( + pv_available_w=3000.0, + bess_scheduled_state="discharging", + bess_requested_w=5000.0, + load_demand_w=5000.0, + grid_connected=False, + )) + assert state.balanced + assert state.pv_power_w == 3000.0 + assert state.bess_power_w == 2000.0 + assert state.grid_power_w == 0.0 + + def test_hybrid_island_solar_excess_charges_bess(self) -> None: + system = EnergySystem.from_config(EnergySystemConfig( + grid=_grid_offline(), + pv=_pv(inverter_type="hybrid"), + bess=_bess(hybrid=True, max_charge_w=3000.0, initial_soe_kwh=5.0), + loads=[LoadConfig(demand_w=2000.0)], + )) + state = system.tick(1000.0, PowerInputs( + pv_available_w=5000.0, + bess_scheduled_state="charging", + bess_requested_w=3000.0, + load_demand_w=2000.0, + grid_connected=False, + )) + assert state.balanced + assert state.pv_power_w == 5000.0 + assert state.bess_state == "charging" + assert state.bess_power_w == 3000.0 + assert state.grid_power_w == 0.0 + + def test_non_hybrid_island_ignores_solar_excess(self) -> None: + system = EnergySystem.from_config(EnergySystemConfig( + grid=_grid_offline(), + pv=_pv(inverter_type="ac_coupled"), + bess=_bess(hybrid=False), + loads=[LoadConfig(demand_w=3000.0)], + )) + state = system.tick(1000.0, PowerInputs( + pv_available_w=5000.0, + bess_scheduled_state="charging", + bess_requested_w=3000.0, + load_demand_w=3000.0, + grid_connected=False, + )) + assert state.balanced + assert state.pv_power_w == 0.0 + # Must discharge to cover load since PV is offline + assert state.bess_power_w == 3000.0 + assert state.bess_state == "discharging" + + +# === Charging/Discharging Grid Impact (Issues 6, 7, 9) === + + +class TestGridImpact: + def test_charging_increases_grid_import(self) -> None: + system = EnergySystem.from_config(EnergySystemConfig( + grid=_grid_online(), + bess=_bess(max_charge_w=3000.0), + loads=[LoadConfig(demand_w=2000.0)], + )) + state = system.tick(1000.0, PowerInputs( + bess_scheduled_state="charging", + bess_requested_w=3000.0, + load_demand_w=2000.0, + grid_connected=True, + )) + assert state.balanced + assert abs(state.grid_power_w - 5000.0) < 0.01 + + def test_discharging_decreases_grid_import(self) -> None: + system = EnergySystem.from_config(EnergySystemConfig( + grid=_grid_online(), + bess=_bess(max_discharge_w=3000.0), + loads=[LoadConfig(demand_w=5000.0)], + )) + state = system.tick(1000.0, PowerInputs( + bess_scheduled_state="discharging", + bess_requested_w=3000.0, + load_demand_w=5000.0, + grid_connected=True, + )) + assert state.balanced + assert abs(state.grid_power_w - 2000.0) < 0.01 + + def test_add_bess_reduces_grid_in_modeling(self) -> None: + config_no_bess = EnergySystemConfig( + grid=_grid_online(), + loads=[LoadConfig(demand_w=5000.0)], + ) + config_with_bess = EnergySystemConfig( + grid=_grid_online(), + bess=_bess(max_discharge_w=3000.0), + loads=[LoadConfig(demand_w=5000.0)], + ) + sys_b = EnergySystem.from_config(config_no_bess) + sys_a = EnergySystem.from_config(config_with_bess) + + state_b = sys_b.tick(1000.0, PowerInputs(load_demand_w=5000.0)) + state_a = sys_a.tick(1000.0, PowerInputs( + load_demand_w=5000.0, + bess_scheduled_state="discharging", + bess_requested_w=3000.0, + )) + assert state_b.balanced and state_a.balanced + assert abs(state_b.grid_power_w - 5000.0) < 0.01 + assert abs(state_a.grid_power_w - 2000.0) < 0.01 + + +# === No Stale State (Issue 10) === + + +class TestIndependentInstances: + def test_two_instances_share_no_state(self) -> None: + config = EnergySystemConfig( + grid=_grid_online(), + bess=_bess(initial_soe_kwh=10.0), + loads=[LoadConfig(demand_w=3000.0)], + ) + sys1 = EnergySystem.from_config(config) + sys2 = EnergySystem.from_config(config) + + # Discharge sys1 for a tick + sys1.tick(1000.0, PowerInputs( + bess_scheduled_state="discharging", + bess_requested_w=3000.0, + load_demand_w=3000.0, + )) + sys1.tick(4600.0, PowerInputs( + bess_scheduled_state="discharging", + bess_requested_w=3000.0, + load_demand_w=3000.0, + )) + + # sys2 should still have initial SOE + state2 = sys2.tick(1000.0, PowerInputs( + bess_scheduled_state="discharging", + bess_requested_w=3000.0, + load_demand_w=3000.0, + )) + assert state2.soe_kwh == 10.0 + + +# === EVSE as Consumer === + + +class TestEVSE: + def test_evse_is_pure_load(self) -> None: + system = EnergySystem.from_config(EnergySystemConfig( + grid=_grid_online(), + loads=[ + LoadConfig(demand_w=3000.0), + LoadConfig(demand_w=7200.0), # EVSE at 30A + ], + )) + state = system.tick(1000.0, PowerInputs( + load_demand_w=10200.0, + grid_connected=True, + )) + assert state.balanced + assert abs(state.grid_power_w - 10200.0) < 0.01 + assert abs(state.load_power_w - 10200.0) < 0.01 + + +# === SOE Duration (Nameplate) === + + +class TestNameplateDuration: + def test_larger_nameplate_sustains_longer(self) -> None: + small_config = EnergySystemConfig( + grid=_grid_online(), + bess=_bess(nameplate_kwh=5.0, max_discharge_w=2500.0, initial_soe_kwh=2.5), + loads=[LoadConfig(demand_w=2500.0)], + ) + large_config = EnergySystemConfig( + grid=_grid_online(), + bess=_bess(nameplate_kwh=20.0, max_discharge_w=2500.0, initial_soe_kwh=10.0), + loads=[LoadConfig(demand_w=2500.0)], + ) + small = EnergySystem.from_config(small_config) + large = EnergySystem.from_config(large_config) + + inputs = PowerInputs( + bess_scheduled_state="discharging", + bess_requested_w=2500.0, + load_demand_w=2500.0, + grid_connected=True, + ) + + # Simulate 2 hours in 60-second ticks + ts = 0.0 + for _ in range(120): + ts += 60.0 + s_small = small.tick(ts, inputs) + s_large = large.tick(ts, inputs) + + # Small BESS (5kWh, started at 50%, reserve=20%=1kWh, usable=1.5kWh) + # should be depleted; large should still be going + assert s_small.bess_state == "idle" + assert s_large.bess_state == "discharging" +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `.venv/bin/python -m pytest tests/test_energy/test_scenarios.py -v` +Expected: FAIL with `ImportError` + +- [ ] **Step 3: Implement EnergySystem** + +Create `src/span_panel_simulator/energy/system.py`: + +```python +"""EnergySystem — the top-level energy balance resolver. + +Constructed from an ``EnergySystemConfig``, ticked with ``PowerInputs``, +and returns a ``SystemState``. Pure value object — no external dependencies, +no shared mutable state. +""" + +from __future__ import annotations + +from span_panel_simulator.energy.bus import PanelBus +from span_panel_simulator.energy.components import BESSUnit, GridMeter, LoadGroup, PVSource +from span_panel_simulator.energy.types import ( + BESSConfig, + EnergySystemConfig, + PowerInputs, + SystemState, +) + + +class EnergySystem: + """Component-based energy balance resolver. + + Instantiate via ``from_config()``. Call ``tick()`` each simulation + step to resolve power flows across the bus. + """ + + def __init__( + self, + bus: PanelBus, + grid: GridMeter, + pv: PVSource | None, + bess: BESSUnit | None, + load: LoadGroup, + ) -> None: + self.bus = bus + self.grid = grid + self.pv = pv + self.bess = bess + self.load = load + + @staticmethod + def from_config(config: EnergySystemConfig) -> EnergySystem: + grid = GridMeter(connected=config.grid.connected) + + pv: PVSource | None = None + if config.pv is not None: + pv = PVSource(available_power_w=0.0, online=True) + + bess: BESSUnit | None = None + if config.bess is not None: + bc: BESSConfig = config.bess + initial_soe = bc.initial_soe_kwh + if initial_soe is None: + initial_soe = bc.nameplate_kwh * 0.5 + bess = BESSUnit( + nameplate_capacity_kwh=bc.nameplate_kwh, + max_charge_w=bc.max_charge_w, + max_discharge_w=bc.max_discharge_w, + charge_efficiency=bc.charge_efficiency, + discharge_efficiency=bc.discharge_efficiency, + backup_reserve_pct=bc.backup_reserve_pct, + hard_min_pct=bc.hard_min_pct, + hybrid=bc.hybrid, + pv_source=pv, + soe_kwh=initial_soe, + ) + + total_demand = sum(lc.demand_w for lc in config.loads) + load = LoadGroup(demand_w=total_demand) + + components: list = [load] + if pv is not None: + components.append(pv) + if bess is not None: + components.append(bess) + components.append(grid) + + bus = PanelBus(components=components) + return EnergySystem(bus=bus, grid=grid, pv=pv, bess=bess, load=load) + + def tick(self, ts: float, inputs: PowerInputs) -> SystemState: + # 1. Apply topology + self.grid.connected = inputs.grid_connected + if self.bess is not None: + self.bess.update_pv_online_status(inputs.grid_connected) + elif self.pv is not None and not inputs.grid_connected: + self.pv.online = False + + # 2. Set component inputs + self.load.demand_w = inputs.load_demand_w + if self.pv is not None: + self.pv.available_power_w = inputs.pv_available_w + if self.bess is not None: + self.bess.scheduled_state = inputs.bess_scheduled_state + self.bess.requested_power_w = inputs.bess_requested_w + + # Non-hybrid islanding override: if grid disconnected and PV + # is offline, BESS must discharge regardless of schedule + if not inputs.grid_connected and not self.bess.hybrid: + self.bess.scheduled_state = "discharging" + self.bess.requested_power_w = self.bess.max_discharge_w + + # 3. Resolve bus + bus_state = self.bus.resolve() + + # 4. Integrate BESS energy + if self.bess is not None: + self.bess.integrate_energy(ts) + + # 5. Return resolved state + pv_power = 0.0 + if self.pv is not None and self.pv.online: + pv_power = self.pv.available_power_w + + bess_power = 0.0 + bess_state = "idle" + soe_kwh = 0.0 + soe_pct = 0.0 + if self.bess is not None: + bess_power = self.bess.effective_power_w + bess_state = self.bess.effective_state + soe_kwh = self.bess.soe_kwh + soe_pct = self.bess.soe_percentage + + return SystemState( + grid_power_w=bus_state.grid_power_w, + pv_power_w=pv_power, + bess_power_w=bess_power, + bess_state=bess_state, + load_power_w=inputs.load_demand_w, + soe_kwh=soe_kwh, + soe_percentage=soe_pct, + balanced=bus_state.is_balanced(), + ) +``` + +- [ ] **Step 4: Update `__init__.py` with public API exports** + +Update `src/span_panel_simulator/energy/__init__.py`: + +```python +"""Component-based energy system for the SPAN panel simulator. + +Public API: + EnergySystem — top-level resolver (construct via from_config) + SystemState — resolved output of a tick + PowerInputs — external inputs that drive each tick + EnergySystemConfig, GridConfig, PVConfig, BESSConfig, LoadConfig — configuration +""" + +from span_panel_simulator.energy.system import EnergySystem +from span_panel_simulator.energy.types import ( + BESSConfig, + EnergySystemConfig, + GridConfig, + LoadConfig, + PowerInputs, + PVConfig, + SystemState, +) + +__all__ = [ + "BESSConfig", + "EnergySystem", + "EnergySystemConfig", + "GridConfig", + "LoadConfig", + "PowerInputs", + "PVConfig", + "SystemState", +] +``` + +- [ ] **Step 5: Run all energy tests** + +Run: `.venv/bin/python -m pytest tests/test_energy/ -v` +Expected: All pass + +- [ ] **Step 6: Run full test suite to confirm nothing is broken** + +Run: `.venv/bin/python -m pytest tests/ -v` +Expected: All 193+ tests pass + +- [ ] **Step 7: Commit** + +```bash +git add src/span_panel_simulator/energy/ tests/test_energy/ +git commit -m "Add EnergySystem with from_config factory and full scenario tests" +``` + +--- + +## Phase 2: Wire Into Engine + +### Task 8: Add EnergySystem to Engine Initialization + +**Files:** +- Modify: `src/span_panel_simulator/engine.py` + +The engine needs to construct an `EnergySystem` from its configuration. This task adds the construction without yet using it for power flow — that comes in Task 9. + +- [ ] **Step 1: Add import and instance variable** + +In `src/span_panel_simulator/engine.py`, add import after existing imports (after line 30): + +```python +from span_panel_simulator.energy import ( + BESSConfig, + EnergySystem, + EnergySystemConfig, + GridConfig, + LoadConfig, + PowerInputs, + PVConfig, + SystemState, +) +``` + +In `DynamicSimulationEngine.__init__()` (after line 843 `self._bsee` assignment), add: + +```python +self._energy_system: EnergySystem | None = None +``` + +- [ ] **Step 2: Add `_build_energy_system()` helper** + +Add method to `DynamicSimulationEngine` (after `_create_bsee()`, after line 1839): + +```python +def _build_energy_system(self) -> EnergySystem | None: + """Construct an EnergySystem from current circuit configuration.""" + if not self._config: + return None + + # Grid config + grid_config = GridConfig(connected=not self._forced_grid_offline) + + # PV config — find producer circuits + pv_config: PVConfig | None = None + for circuit in self._circuits.values(): + if circuit.energy_mode == "producer": + nameplate = float(circuit.template.get("typical_power", 0)) + inverter_type = circuit.template.get("inverter_type", "ac_coupled") + pv_config = PVConfig(nameplate_w=abs(nameplate), inverter_type=inverter_type) + break + + # BESS config — from battery circuit template + bess_config: BESSConfig | None = None + battery_circuit = self._find_battery_circuit() + if battery_circuit is not None: + battery_cfg = battery_circuit.template.get("battery_behavior", {}) + if isinstance(battery_cfg, dict) and battery_cfg.get("enabled", False): + nameplate = float(battery_cfg.get("nameplate_capacity_kwh", 13.5)) + hybrid = battery_cfg.get("inverter_type") == "hybrid" + bess_config = BESSConfig( + nameplate_kwh=nameplate, + max_charge_w=abs(float(battery_cfg.get("max_charge_power", 3500.0))), + max_discharge_w=abs(float(battery_cfg.get("max_discharge_power", 3500.0))), + charge_efficiency=float(battery_cfg.get("charge_efficiency", 0.95)), + discharge_efficiency=float(battery_cfg.get("discharge_efficiency", 0.95)), + backup_reserve_pct=float(battery_cfg.get("backup_reserve_pct", 20.0)), + hybrid=hybrid, + initial_soe_kwh=self._bsee.soe_kwh if self._bsee is not None else None, + ) + + # Load configs — all consumer circuits + loads = [LoadConfig() for c in self._circuits.values() if c.energy_mode == "consumer"] + + config = EnergySystemConfig( + grid=grid_config, + pv=pv_config, + bess=bess_config, + loads=loads, + ) + return EnergySystem.from_config(config) +``` + +- [ ] **Step 3: Construct energy system during initialization** + +In `initialize_async()` (after `_create_bsee()` call, after line 906), add: + +```python +self._energy_system = self._build_energy_system() +``` + +- [ ] **Step 4: Run full test suite** + +Run: `.venv/bin/python -m pytest tests/ -v` +Expected: All tests pass (energy system constructed but not yet used for output) + +- [ ] **Step 5: Commit** + +```bash +git add src/span_panel_simulator/energy/ src/span_panel_simulator/engine.py +git commit -m "Wire EnergySystem construction into engine initialization" +``` + +--- + +### Task 9: Use EnergySystem in `get_snapshot()` + +**Files:** +- Modify: `src/span_panel_simulator/engine.py` + +Replace the inline energy aggregation (lines ~1250-1296) with `EnergySystem.tick()`. The engine still ticks circuits via the behavior engine, but reads grid/battery/pv power from `SystemState`. + +- [ ] **Step 1: Add `_collect_power_inputs()` helper** + +Add method to `DynamicSimulationEngine`: + +```python +def _collect_power_inputs(self) -> PowerInputs: + """Collect current circuit state into PowerInputs for the energy system.""" + pv_power = 0.0 + load_power = 0.0 + bess_power = 0.0 + bess_state = "idle" + + for circuit in self._circuits.values(): + power = circuit.instant_power_w + if circuit.energy_mode == "producer": + pv_power += power + elif circuit.energy_mode == "bidirectional": + bess_power = power + if self._bsee is not None: + bess_state = self._bsee.battery_state + elif self._behavior_engine is not None: + bess_state = self._behavior_engine.last_battery_direction + else: + load_power += power + + return PowerInputs( + pv_available_w=pv_power, + bess_requested_w=bess_power, + bess_scheduled_state=bess_state, + load_demand_w=load_power, + grid_connected=not self._forced_grid_offline, + ) +``` + +- [ ] **Step 2: Replace energy aggregation in `get_snapshot()`** + +In `get_snapshot()`, after the circuit ticking and global overrides (after line ~1232, before the old aggregation), replace the aggregation block (lines ~1249-1295) with: + +```python +# 5. Resolve energy balance via component system +system_state: SystemState | None = None +if self._energy_system is not None: + inputs = self._collect_power_inputs() + system_state = self._energy_system.tick(current_time, inputs) + + # Reflect effective battery power back to circuit + battery_circuit = self._find_battery_circuit() + if battery_circuit is not None and self._energy_system.bess is not None: + battery_circuit._instant_power_w = self._energy_system.bess.effective_power_w + + grid_power = system_state.grid_power_w + site_power = system_state.load_power_w - system_state.pv_power_w + battery_power_w = system_state.bess_power_w +``` + +Keep the old code path as a fallback for when `_energy_system` is None (during initialization before config is loaded). + +- [ ] **Step 3: Update BSEE interaction** + +After the energy system resolves, the old BSEE update call (line ~1286) should be skipped when the energy system is active (BESSUnit handles SOE tracking). Guard it: + +```python +if self._bsee is not None and self._energy_system is None: + self._bsee.update(current_time, battery_power_w, site_power_w=site_power) + # ... existing BSEE reflection code +``` + +- [ ] **Step 4: Run full test suite** + +Run: `.venv/bin/python -m pytest tests/ -v` +Expected: All tests pass + +- [ ] **Step 5: Commit** + +```bash +git add src/span_panel_simulator/engine.py +git commit -m "Use EnergySystem for power flow resolution in get_snapshot" +``` + +--- + +### Task 10: Use EnergySystem in `get_power_summary()` and `compute_modeling_data()` + +**Files:** +- Modify: `src/span_panel_simulator/engine.py` + +- [ ] **Step 1: Update `get_power_summary()` (lines ~1067-1154)** + +Replace the inline aggregation with a read from the energy system's last state. Add a `_last_system_state` field to cache the most recent tick result: + +In `__init__()` add: + +```python +self._last_system_state: SystemState | None = None +``` + +In `get_snapshot()` after `system_state = self._energy_system.tick(...)`: + +```python +self._last_system_state = system_state +``` + +Then simplify `get_power_summary()` to read from `_last_system_state` when available, falling back to the old logic only when the energy system isn't initialized. + +- [ ] **Step 2: Update `compute_modeling_data()` (lines ~1507-1682)** + +Replace the inline grid/battery calculation (lines ~1627-1650) with independent `EnergySystem` instances: + +In `compute_modeling_data()`, before the timestamp loop, construct two energy systems: + +```python +config_before = self._build_energy_config(baseline=True) +config_after = self._build_energy_config(baseline=False) +system_before = EnergySystem.from_config(config_before) if config_before else None +system_after = EnergySystem.from_config(config_after) if config_after else None +``` + +Add `_build_energy_config()` helper that creates `EnergySystemConfig` from the current or baseline configuration. + +Inside the timestamp loop, replace the manual grid calculation with: + +```python +if system_after is not None: + inputs_a = PowerInputs( + pv_available_w=prod_a, + bess_requested_w=raw_batt_a, + bess_scheduled_state=cloned_bsee_state, + load_demand_w=site_a + prod_a, # consumption component + grid_connected=True, + ) + state_a = system_after.tick(ts, inputs_a) + grid_after = state_a.grid_power_w + signed_battery_after = -state_a.bess_power_w if state_a.bess_state == "discharging" else state_a.bess_power_w +``` + +- [ ] **Step 3: Run full test suite** + +Run: `.venv/bin/python -m pytest tests/ -v` +Expected: All tests pass + +- [ ] **Step 4: Commit** + +```bash +git add src/span_panel_simulator/engine.py +git commit -m "Use EnergySystem in power summary and modeling data" +``` + +--- + +## Phase 3: Eliminate Old Paths + +### Task 11: Remove Old Energy Balance Code and BSEE + +**Files:** +- Modify: `src/span_panel_simulator/engine.py` +- Delete: `src/span_panel_simulator/bsee.py` +- Modify: `src/span_panel_simulator/circuit.py` + +- [ ] **Step 1: Remove old inline energy aggregation from `get_snapshot()`** + +Remove the fallback path that was guarded by `if self._energy_system is None`. At this point the energy system is always present. Remove the old aggregation code (the `total_consumption`, `total_production`, `battery_circuit_power` accumulation loop and the `site_power = total_consumption - total_production` / `grid_power = site_power - battery_circuit_power` formulas). + +- [ ] **Step 2: Remove old `get_power_summary()` inline aggregation** + +Remove the `battery_state` sign-flipping block (lines ~1093-1110) and the `max(0.0, total_consumption - total_production)` formula. Replace with reads from `_last_system_state`. + +- [ ] **Step 3: Remove old BSEE interaction from engine** + +Remove: +- `from span_panel_simulator.bsee import BatteryStorageEquipment` import (line 26) +- `self._bsee` field from `__init__()` (line 843) +- `_create_bsee()` method (lines 1809-1839) +- `_bsee` assignment in `initialize_async()` (line 906) +- The `self._bsee.update()` call and reflection block in `get_snapshot()` (lines ~1285-1295) +- `self._bsee.set_forced_offline()` in `set_grid_online()` (line 982) +- All reads of `self._bsee` properties (soe_percentage, battery_state, etc.) — replace with reads from `SystemState` or `self._energy_system.bess` +- The cloned BSEE construction in `compute_modeling_data()` (lines ~1579-1591) + +- [ ] **Step 4: Remove grid-offline force-discharge override from behavior engine** + +In `_apply_battery_behavior()` (line ~478), remove: + +```python +if self._grid_offline: + self._last_battery_direction = "discharging" + return self._get_discharge_power(battery_config, current_hour) +``` + +The `EnergySystem.tick()` now handles islanding behavior in its non-hybrid override logic. + +- [ ] **Step 5: Remove dashboard `battery = consumption - pv` shortcut** + +In `get_power_summary()` (line ~1110), remove: + +```python +if self.has_battery: + battery = total_consumption - pv +``` + +This is now handled by `SystemState`. + +- [ ] **Step 6: Update circuit.py battery direction** + +In `circuit.py` `_resolve_battery_direction()` (lines 297-320): this method should read from the energy system's `bess.effective_state` rather than querying the behavior engine's `last_battery_direction`. The engine should pass this through after each tick. + +- [ ] **Step 7: Delete `bsee.py`** + +```bash +git rm src/span_panel_simulator/bsee.py +``` + +- [ ] **Step 8: Run full test suite** + +Run: `.venv/bin/python -m pytest tests/ -v` +Expected: All tests pass + +- [ ] **Step 9: Run type checker** + +Run: `.venv/bin/python -m mypy src/span_panel_simulator/ --no-error-summary` +Expected: No errors + +- [ ] **Step 10: Commit** + +```bash +git add -A +git commit -m "Remove old energy balance paths and delete bsee.py" +``` + +--- + +## Phase 4: Dead Code Removal + +### Task 12: Remove All Dead Code + +**Files:** +- Modify: `src/span_panel_simulator/engine.py` +- Modify: `src/span_panel_simulator/circuit.py` +- Possibly modify: `src/span_panel_simulator/dashboard/routes.py` + +- [ ] **Step 1: Search for dead references** + +Search for all references to removed code: + +```bash +# References to old BSEE +.venv/bin/python -m pytest tests/ -v # ensure clean first +``` + +Then grep for orphaned references: + +- `BatteryStorageEquipment` — should have zero hits outside of git history +- `_bsee` — should have zero hits in engine.py +- `_create_bsee` — should be gone +- `battery_circuit_power` variable in get_snapshot — should be gone +- `signed_battery_before`, `signed_battery_after` — should be gone from modeling if replaced +- `cloned_bsee` — should be gone +- `_forced_grid_offline` checks that duplicate what EnergySystem handles +- `set_solar_excess` on behavior engine — if the energy system handles solar-excess through bus ordering, this may become dead +- `last_battery_direction` on behavior engine — check if still needed after circuit.py update + +- [ ] **Step 2: Remove orphaned imports and variables** + +Remove any imports, variables, or methods that are no longer referenced after the migration. Check: + +- `engine.py`: unused imports, orphaned helper methods, variables only written but never read +- `circuit.py`: `_resolve_battery_direction` if replaced by energy system state +- `behavior_mutable_state.py`: fields that were only used by cloned BSEE + +- [ ] **Step 3: Remove the GFE throttling we added to bsee.py earlier in this conversation** + +This was the `site_power_w` parameter we added to `BatteryStorageEquipment.update()`. Since `bsee.py` is deleted in Task 11, verify it is indeed gone. If any vestiges remain (e.g., the engine passing `site_power_w`), remove them. + +- [ ] **Step 4: Run full test suite** + +Run: `.venv/bin/python -m pytest tests/ -v` +Expected: All tests pass + +- [ ] **Step 5: Run type checker** + +Run: `.venv/bin/python -m mypy src/span_panel_simulator/ --no-error-summary` +Expected: No errors + +- [ ] **Step 6: Run linter** + +Run: `.venv/bin/python -m ruff check src/span_panel_simulator/` +Expected: No errors (or only pre-existing ones) + +- [ ] **Step 7: Commit** + +```bash +git add -A +git commit -m "Remove dead code from energy system migration" +``` + +--- + +## Summary + +| Phase | Tasks | What it achieves | +|-------|-------|-----------------| +| **Phase 1** | Tasks 1-7 | Energy system built and tested in isolation. Engine untouched. | +| **Phase 2** | Tasks 8-10 | Engine wired to EnergySystem for snapshot, dashboard, and modeling. | +| **Phase 3** | Task 11 | Old energy balance code and bsee.py removed. Single source of truth. | +| **Phase 4** | Task 12 | Dead code sweep. No orphaned references. Clean codebase. | diff --git a/docs/superpowers/plans/2026-03-28-tou-rate-integration.md b/docs/superpowers/plans/2026-03-28-tou-rate-integration.md new file mode 100644 index 0000000..dd1d983 --- /dev/null +++ b/docs/superpowers/plans/2026-03-28-tou-rate-integration.md @@ -0,0 +1,2229 @@ +# ToU Rate Integration Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Integrate OpenEI URDB rate plans into the simulator so users can select current and proposed ToU rates, and the modeling view displays cost alongside energy for Before/After comparisons. + +**Architecture:** A new `rates/` package handles all rate concerns: API client (`openei.py`), rate lookup (`resolver.py`), and cost calculation (`cost_engine.py`). Rate data is cached simulator-wide in `data/rates_cache.yaml`. The engine's `compute_modeling_data` delegates cost math to the cost engine. The modeling view gains a rate selection UI above the charts with an OpenEI configuration dialog. + +**Tech Stack:** Python 3.14, aiohttp (existing), dataclasses, pytest. No new dependencies. + +**Spec:** `docs/superpowers/specs/2026-03-28-tou-rate-integration-design.md` + +**AGENTS.md Rules:** The `rates/` package is a peer to `energy/` — it consumes the power arrays produced by `compute_modeling_data` but does not participate in energy dispatch. The engine passes power arrays to the cost engine; it never resolves rates or computes costs inline. + +--- + +## File Structure + +| File | Action | Responsibility | +|------|--------|---------------| +| `src/span_panel_simulator/rates/__init__.py` | Create | Public API re-exports | +| `src/span_panel_simulator/rates/types.py` | Create | URDBRecord, RateCacheEntry, AttributionMeta, CostLedger, OpenEIConfig, UtilitySummary, RatePlanSummary | +| `src/span_panel_simulator/rates/resolver.py` | Create | `resolve_rate(timestamp, tz, record)` -> (import, export) rate per kWh | +| `src/span_panel_simulator/rates/cost_engine.py` | Create | `compute_costs(timestamps, power_kw, record, tz)` -> CostLedger | +| `src/span_panel_simulator/rates/openei.py` | Create | URDB API client: fetch utilities, plans, detail | +| `src/span_panel_simulator/rates/cache.py` | Create | Rate cache manager: load/save/get/set rates_cache.yaml + openei config | +| `tests/test_rates/__init__.py` | Create | Test package init | +| `tests/test_rates/test_resolver.py` | Create | Rate resolution unit tests | +| `tests/test_rates/test_cost_engine.py` | Create | Cost calculation unit tests | +| `tests/test_rates/test_openei.py` | Create | API client tests with mocked HTTP | +| `tests/test_rates/test_cache.py` | Create | Cache read/write tests | +| `src/span_panel_simulator/dashboard/routes.py` | Modify | Add rate API endpoints | +| `src/span_panel_simulator/dashboard/templates/partials/modeling_view.html` | Modify | Rate selection UI, cost display, OpenEI dialog | +| `src/span_panel_simulator/engine.py` | Modify | Wire cost engine into `compute_modeling_data` response | + +--- + +## Phase 1: Rate Types and Resolution + +### Task 1: Types Module + +**Files:** +- Create: `src/span_panel_simulator/rates/__init__.py` +- Create: `src/span_panel_simulator/rates/types.py` +- Create: `tests/test_rates/__init__.py` + +- [ ] **Step 1: Create package structure** + +```bash +mkdir -p src/span_panel_simulator/rates tests/test_rates +``` + +- [ ] **Step 2: Write types module** + +Create `src/span_panel_simulator/rates/__init__.py`: + +```python +"""ToU rate integration — OpenEI URDB rate plans and cost calculation.""" +``` + +Create `tests/test_rates/__init__.py`: + +```python +``` + +Create `src/span_panel_simulator/rates/types.py`: + +```python +"""Core types for ToU rate integration. + +URDBRecord mirrors the OpenEI URDB API response schema. Records are +stored verbatim — never modified after fetch. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, TypedDict + + +# -- URDB schema types (read-only mirrors of API response) --------------- + +class URDBRateTier(TypedDict, total=False): + """Single tier within a rate period.""" + + rate: float # $/kWh + max: float # kWh ceiling for this tier (absent on last tier) + unit: str # e.g. "kWh" + adj: float # adjustment factor + + +class URDBRecord(TypedDict, total=False): + """Subset of the OpenEI URDB v3 rate record we consume. + + Stored verbatim from the API — all fields are optional because + different rate plans populate different subsets. + """ + + label: str + utility: str + name: str + uri: str + startdate: int # epoch seconds + enddate: int # epoch seconds + sector: str + description: str + source: str + + # Energy charges + energyratestructure: list[list[URDBRateTier]] + energyweekdayschedule: list[list[int]] # 12 x 24 (month x hour) + energyweekendschedule: list[list[int]] # 12 x 24 + + # Export / sell rates + sell: list[list[URDBRateTier]] # same shape as energyratestructure + usenetmetering: bool + + # Fixed charges + fixedmonthlycharge: float + minmonthlycharge: float + annualmincharge: float + + # Demand charges (flat) + flatdemandstructure: list[list[URDBRateTier]] + flatdemandmonths: list[list[int]] # 12-element, period per month + + # Demand charges (time-based) — stored but not used in v1 + demandratestructure: list[list[URDBRateTier]] + demandweekdayschedule: list[list[int]] + demandweekendschedule: list[list[int]] + + +# -- Metadata and cache types ------------------------------------------- + +@dataclass(frozen=True) +class AttributionMeta: + """Provenance metadata for a cached rate record.""" + + provider: str + url: str + license: str + api_version: int + + +@dataclass(frozen=True) +class RateCacheEntry: + """A cached URDB record with its metadata envelope.""" + + source: str + retrieved_at: str # ISO 8601 + attribution: AttributionMeta + record: dict[str, Any] # raw URDB JSON — typed access via URDBRecord + + +@dataclass(frozen=True) +class OpenEIConfig: + """User-configurable OpenEI API settings.""" + + api_url: str = "https://api.openei.org/utility_rates" + api_key: str = "" + + +# -- Cost calculation result -------------------------------------------- + +@dataclass(frozen=True) +class CostLedger: + """Result of applying a rate schedule to a power time-series.""" + + import_cost: float # $ over horizon + export_credit: float # $ over horizon + fixed_charges: float # $ over horizon (monthly fixed + flat demand) + net_cost: float # import_cost - export_credit + fixed_charges + + +# -- API response summaries --------------------------------------------- + +@dataclass(frozen=True) +class UtilitySummary: + """Minimal info for a utility returned by URDB search.""" + + utility_name: str + eia_id: str + + +@dataclass(frozen=True) +class RatePlanSummary: + """Minimal info for a rate plan in a utility's offerings.""" + + label: str + name: str + startdate: int # epoch seconds + enddate: int | None # epoch seconds or None if open-ended + description: str +``` + +- [ ] **Step 3: Commit** + +```bash +git add src/span_panel_simulator/rates/ tests/test_rates/ +git commit -m "Add rates package with core types for URDB integration" +``` + +--- + +### Task 2: Rate Resolver + +**Files:** +- Create: `src/span_panel_simulator/rates/resolver.py` +- Create: `tests/test_rates/test_resolver.py` + +- [ ] **Step 1: Write the failing tests** + +Create `tests/test_rates/test_resolver.py`: + +```python +"""Tests for ToU rate resolution from URDB schedule matrices.""" + +from __future__ import annotations + +import calendar +from datetime import datetime, timezone + +from zoneinfo import ZoneInfo + +from span_panel_simulator.rates.resolver import resolve_rate + + +def _epoch(year: int, month: int, day: int, hour: int, tz: str) -> int: + """Return epoch seconds for a local datetime.""" + dt = datetime(year, month, day, hour, tzinfo=ZoneInfo(tz)) + return int(dt.timestamp()) + + +# Minimal URDB record for testing: 2 periods (off-peak=0, peak=1) +# Peak: weekdays 16-21, all months +# Off-peak: everything else +_WEEKDAY_SCHEDULE = [[0] * 24 for _ in range(12)] +for _m in range(12): + for _h in range(16, 21): + _WEEKDAY_SCHEDULE[_m][_h] = 1 + +_WEEKEND_SCHEDULE = [[0] * 24 for _ in range(12)] + +_RECORD: dict = { + "energyratestructure": [ + [{"rate": 0.10}], # period 0: off-peak $0.10/kWh + [{"rate": 0.30}], # period 1: peak $0.30/kWh + ], + "energyweekdayschedule": _WEEKDAY_SCHEDULE, + "energyweekendschedule": _WEEKEND_SCHEDULE, + "sell": [ + [{"rate": 0.04}], # period 0: off-peak export $0.04/kWh + [{"rate": 0.08}], # period 1: peak export $0.08/kWh + ], +} + +TZ = "America/Los_Angeles" + + +class TestResolveRate: + """Rate resolution against URDB schedule matrices.""" + + def test_weekday_peak_hour(self) -> None: + # Wednesday 2026-07-15 at 17:00 local -> peak (period 1) + ts = _epoch(2026, 7, 15, 17, TZ) + imp, exp = resolve_rate(ts, TZ, _RECORD) + assert imp == 0.30 + assert exp == 0.08 + + def test_weekday_offpeak_hour(self) -> None: + # Wednesday 2026-07-15 at 10:00 local -> off-peak (period 0) + ts = _epoch(2026, 7, 15, 10, TZ) + imp, exp = resolve_rate(ts, TZ, _RECORD) + assert imp == 0.10 + assert exp == 0.04 + + def test_weekend_always_offpeak(self) -> None: + # Saturday 2026-07-18 at 17:00 local -> off-peak (weekend schedule) + ts = _epoch(2026, 7, 18, 17, TZ) + imp, exp = resolve_rate(ts, TZ, _RECORD) + assert imp == 0.10 + assert exp == 0.04 + + def test_no_sell_field_returns_zero_export(self) -> None: + record_no_sell = { + "energyratestructure": [[{"rate": 0.15}]], + "energyweekdayschedule": [[0] * 24 for _ in range(12)], + "energyweekendschedule": [[0] * 24 for _ in range(12)], + } + ts = _epoch(2026, 1, 5, 12, TZ) + imp, exp = resolve_rate(ts, TZ, record_no_sell) + assert imp == 0.15 + assert exp == 0.0 + + def test_winter_vs_summer_month(self) -> None: + # Same hour, different months — verify month index is used + ts_jan = _epoch(2026, 1, 7, 17, TZ) # Wednesday + ts_jul = _epoch(2026, 7, 15, 17, TZ) # Wednesday + imp_jan, _ = resolve_rate(ts_jan, TZ, _RECORD) + imp_jul, _ = resolve_rate(ts_jul, TZ, _RECORD) + # Both should be peak since our test schedule uses peak 16-21 all months + assert imp_jan == 0.30 + assert imp_jul == 0.30 + + def test_boundary_hour_15_is_offpeak(self) -> None: + # Hour 15 is off-peak (peak starts at 16) + ts = _epoch(2026, 7, 15, 15, TZ) + imp, _ = resolve_rate(ts, TZ, _RECORD) + assert imp == 0.10 + + def test_boundary_hour_16_is_peak(self) -> None: + # Hour 16 is peak + ts = _epoch(2026, 7, 15, 16, TZ) + imp, _ = resolve_rate(ts, TZ, _RECORD) + assert imp == 0.30 + + def test_boundary_hour_21_is_offpeak(self) -> None: + # Hour 21 is off-peak (peak ends at 20 inclusive) + ts = _epoch(2026, 7, 15, 21, TZ) + imp, _ = resolve_rate(ts, TZ, _RECORD) + assert imp == 0.10 +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `pytest tests/test_rates/test_resolver.py -v` +Expected: FAIL with `ModuleNotFoundError: No module named 'span_panel_simulator.rates.resolver'` + +- [ ] **Step 3: Write the resolver** + +Create `src/span_panel_simulator/rates/resolver.py`: + +```python +"""Resolve import/export rates from URDB schedule matrices. + +Given a UNIX timestamp and timezone, looks up the rate for that hour +using the URDB energyweekday/weekendschedule (12x24 month x hour +matrices) and energyratestructure (period -> tier list). + +Uses tier 1 (index 0) only in v1. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any + +from zoneinfo import ZoneInfo + + +def resolve_rate( + timestamp: int, + tz: str, + record: dict[str, Any], +) -> tuple[float, float]: + """Return (import_rate_per_kwh, export_rate_per_kwh) for a timestamp. + + Parameters + ---------- + timestamp: + UNIX epoch seconds. + tz: + IANA timezone string (e.g. "America/Los_Angeles"). + record: + URDB record dict containing energyratestructure and schedule matrices. + """ + dt = datetime.fromtimestamp(timestamp, tz=ZoneInfo(tz)) + month_idx = dt.month - 1 # 0-based (Jan=0) + hour_idx = dt.hour # 0-based (0-23) + + # Weekday: Mon=0 .. Sun=6; URDB weekday schedule covers Mon-Fri + is_weekend = dt.weekday() >= 5 + if is_weekend: + schedule = record.get("energyweekendschedule", []) + else: + schedule = record.get("energyweekdayschedule", []) + + # Look up period index from schedule matrix + if schedule and month_idx < len(schedule) and hour_idx < len(schedule[month_idx]): + period_idx = schedule[month_idx][hour_idx] + else: + period_idx = 0 + + # Import rate: tier 1 of the period + import_rate = 0.0 + rate_structure = record.get("energyratestructure", []) + if rate_structure and period_idx < len(rate_structure): + tiers = rate_structure[period_idx] + if tiers: + import_rate = tiers[0].get("rate", 0.0) + + # Export rate: tier 1 of the sell structure, same period + export_rate = 0.0 + sell_structure = record.get("sell", []) + if sell_structure and period_idx < len(sell_structure): + sell_tiers = sell_structure[period_idx] + if sell_tiers: + export_rate = sell_tiers[0].get("rate", 0.0) + + return (import_rate, export_rate) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `pytest tests/test_rates/test_resolver.py -v` +Expected: All 8 tests PASS + +- [ ] **Step 5: Commit** + +```bash +git add src/span_panel_simulator/rates/resolver.py tests/test_rates/test_resolver.py +git commit -m "Add rate resolver for URDB schedule matrix lookup" +``` + +--- + +### Task 3: Cost Engine + +**Files:** +- Create: `src/span_panel_simulator/rates/cost_engine.py` +- Create: `tests/test_rates/test_cost_engine.py` + +- [ ] **Step 1: Write the failing tests** + +Create `tests/test_rates/test_cost_engine.py`: + +```python +"""Tests for the cost engine — applies rates to power time-series.""" + +from __future__ import annotations + +from datetime import datetime + +import pytest +from zoneinfo import ZoneInfo + +from span_panel_simulator.rates.cost_engine import compute_costs + +TZ = "America/Los_Angeles" + + +def _epoch(year: int, month: int, day: int, hour: int) -> int: + dt = datetime(year, month, day, hour, tzinfo=ZoneInfo(TZ)) + return int(dt.timestamp()) + + +# Flat rate: single period, $0.20 import, $0.05 export, all hours +_FLAT_RECORD: dict = { + "energyratestructure": [[{"rate": 0.20}]], + "energyweekdayschedule": [[0] * 24 for _ in range(12)], + "energyweekendschedule": [[0] * 24 for _ in range(12)], + "sell": [[{"rate": 0.05}]], + "fixedmonthlycharge": 10.0, +} + + +class TestComputeCosts: + """Cost engine applies rates to hourly power arrays.""" + + def test_pure_import(self) -> None: + # 3 hours at 1.0 kW import = 3 kWh * $0.20 = $0.60 + ts = [_epoch(2026, 3, 15, h) for h in range(10, 13)] + power = [1.0, 1.0, 1.0] + result = compute_costs(ts, power, _FLAT_RECORD, TZ) + assert result.import_cost == pytest.approx(0.60) + assert result.export_credit == pytest.approx(0.0) + + def test_pure_export(self) -> None: + # 2 hours at -2.0 kW (export) = 4 kWh * $0.05 = $0.20 + ts = [_epoch(2026, 3, 15, h) for h in range(10, 12)] + power = [-2.0, -2.0] + result = compute_costs(ts, power, _FLAT_RECORD, TZ) + assert result.import_cost == pytest.approx(0.0) + assert result.export_credit == pytest.approx(0.20) + + def test_mixed_import_export(self) -> None: + # Hour 1: 1.0 kW import = $0.20, Hour 2: -1.0 kW export = $0.05 + ts = [_epoch(2026, 3, 15, 10), _epoch(2026, 3, 15, 11)] + power = [1.0, -1.0] + result = compute_costs(ts, power, _FLAT_RECORD, TZ) + assert result.import_cost == pytest.approx(0.20) + assert result.export_credit == pytest.approx(0.05) + + def test_net_cost_includes_fixed(self) -> None: + # 1 hour in March -> 1 month -> $10 fixed + ts = [_epoch(2026, 3, 15, 10)] + power = [1.0] # $0.20 import + result = compute_costs(ts, power, _FLAT_RECORD, TZ) + assert result.fixed_charges == pytest.approx(10.0) + assert result.net_cost == pytest.approx(0.20 - 0.0 + 10.0) + + def test_multi_month_fixed_charges(self) -> None: + # Hours spanning Jan and Feb -> 2 months -> $20 fixed + ts = [_epoch(2026, 1, 31, 23), _epoch(2026, 2, 1, 0)] + power = [0.0, 0.0] + result = compute_costs(ts, power, _FLAT_RECORD, TZ) + assert result.fixed_charges == pytest.approx(20.0) + + def test_no_fixed_charge_field(self) -> None: + record_no_fixed: dict = { + "energyratestructure": [[{"rate": 0.10}]], + "energyweekdayschedule": [[0] * 24 for _ in range(12)], + "energyweekendschedule": [[0] * 24 for _ in range(12)], + } + ts = [_epoch(2026, 3, 15, 10)] + power = [1.0] + result = compute_costs(ts, power, record_no_fixed, TZ) + assert result.fixed_charges == pytest.approx(0.0) + + def test_zero_power(self) -> None: + ts = [_epoch(2026, 3, 15, h) for h in range(10, 14)] + power = [0.0, 0.0, 0.0, 0.0] + result = compute_costs(ts, power, _FLAT_RECORD, TZ) + assert result.import_cost == pytest.approx(0.0) + assert result.export_credit == pytest.approx(0.0) + + def test_empty_arrays(self) -> None: + result = compute_costs([], [], _FLAT_RECORD, TZ) + assert result.import_cost == pytest.approx(0.0) + assert result.export_credit == pytest.approx(0.0) + assert result.fixed_charges == pytest.approx(0.0) + assert result.net_cost == pytest.approx(0.0) + + def test_flat_demand_included_in_fixed(self) -> None: + record_with_demand: dict = { + "energyratestructure": [[{"rate": 0.10}]], + "energyweekdayschedule": [[0] * 24 for _ in range(12)], + "energyweekendschedule": [[0] * 24 for _ in range(12)], + "fixedmonthlycharge": 5.0, + "flatdemandstructure": [[{"rate": 8.0}]], + "flatdemandmonths": [[0] * 12], + } + # 1 month -> $5 fixed + $8 flat demand = $13 + ts = [_epoch(2026, 3, 15, 10)] + power = [0.0] + result = compute_costs(ts, power, record_with_demand, TZ) + assert result.fixed_charges == pytest.approx(13.0) +``` + +Note: Add `import pytest` at the top of the test file (for `pytest.approx`). + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `pytest tests/test_rates/test_cost_engine.py -v` +Expected: FAIL with `ModuleNotFoundError: No module named 'span_panel_simulator.rates.cost_engine'` + +- [ ] **Step 3: Write the cost engine** + +Create `src/span_panel_simulator/rates/cost_engine.py`: + +```python +"""Batch cost calculation over a modeling horizon. + +Applies a URDB rate record to hourly power arrays, producing a +CostLedger with import cost, export credit, fixed charges, and net cost. +""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any + +from zoneinfo import ZoneInfo + +from span_panel_simulator.rates.resolver import resolve_rate +from span_panel_simulator.rates.types import CostLedger + + +def compute_costs( + timestamps: list[int], + power_kw: list[float], + record: dict[str, Any], + tz: str, + resolution_s: int = 3600, +) -> CostLedger: + """Compute total costs over a horizon. + + Parameters + ---------- + timestamps: + UNIX epoch seconds, one per interval (hourly). + power_kw: + Grid power per interval in kW. Positive = import, negative = export. + record: + URDB record dict. + tz: + IANA timezone string. + resolution_s: + Interval length in seconds (default 3600 = 1 hour). + """ + import_cost = 0.0 + export_credit = 0.0 + months_seen: set[tuple[int, int]] = set() # (year, month) + + for ts, pwr in zip(timestamps, power_kw): + import_rate, export_rate = resolve_rate(ts, tz, record) + energy_kwh = pwr * resolution_s / 3600 + + if energy_kwh > 0: + import_cost += energy_kwh * import_rate + elif energy_kwh < 0: + export_credit += abs(energy_kwh) * export_rate + + dt = datetime.fromtimestamp(ts, tz=ZoneInfo(tz)) + months_seen.add((dt.year, dt.month)) + + num_months = len(months_seen) + fixed_monthly = record.get("fixedmonthlycharge", 0.0) or 0.0 + flat_demand = _flat_demand_per_month(record) + fixed_charges = num_months * (fixed_monthly + flat_demand) + + net_cost = import_cost - export_credit + fixed_charges + + return CostLedger( + import_cost=import_cost, + export_credit=export_credit, + fixed_charges=fixed_charges, + net_cost=net_cost, + ) + + +def _flat_demand_per_month(record: dict[str, Any]) -> float: + """Extract flat demand charge per month from URDB record. + + Uses tier 1 of period 0 from flatdemandstructure. + """ + structure = record.get("flatdemandstructure", []) + if not structure: + return 0.0 + period_tiers = structure[0] + if not period_tiers: + return 0.0 + return period_tiers[0].get("rate", 0.0) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `pytest tests/test_rates/test_cost_engine.py -v` +Expected: All 9 tests PASS + +- [ ] **Step 5: Commit** + +```bash +git add src/span_panel_simulator/rates/cost_engine.py tests/test_rates/test_cost_engine.py +git commit -m "Add cost engine for hourly power-to-cost calculation" +``` + +--- + +## Phase 2: Rate Cache and OpenEI Client + +### Task 4: Rate Cache Manager + +**Files:** +- Create: `src/span_panel_simulator/rates/cache.py` +- Create: `tests/test_rates/test_cache.py` + +- [ ] **Step 1: Write the failing tests** + +Create `tests/test_rates/test_cache.py`: + +```python +"""Tests for the rate cache manager.""" + +from __future__ import annotations + +from pathlib import Path + +from span_panel_simulator.rates.cache import RateCache +from span_panel_simulator.rates.types import OpenEIConfig + + +SAMPLE_URDB_RECORD: dict = { + "label": "abc123", + "utility": "Pacific Gas & Electric Co", + "name": "E-TOU-C", + "energyratestructure": [[{"rate": 0.25}]], + "energyweekdayschedule": [[0] * 24 for _ in range(12)], + "energyweekendschedule": [[0] * 24 for _ in range(12)], +} + + +class TestRateCache: + """Rate cache load/save/get/set operations.""" + + def test_empty_cache_returns_none(self, tmp_path: Path) -> None: + cache = RateCache(tmp_path / "rates_cache.yaml") + assert cache.get_cached_rate("nonexistent") is None + + def test_cache_and_retrieve(self, tmp_path: Path) -> None: + cache = RateCache(tmp_path / "rates_cache.yaml") + cache.cache_rate("abc123", SAMPLE_URDB_RECORD) + entry = cache.get_cached_rate("abc123") + assert entry is not None + assert entry.record["label"] == "abc123" + assert entry.record["utility"] == "Pacific Gas & Electric Co" + assert entry.source == "openei_urdb" + assert entry.attribution.license == "CC0" + + def test_persistence_across_instances(self, tmp_path: Path) -> None: + path = tmp_path / "rates_cache.yaml" + cache1 = RateCache(path) + cache1.cache_rate("abc123", SAMPLE_URDB_RECORD) + + cache2 = RateCache(path) + entry = cache2.get_cached_rate("abc123") + assert entry is not None + assert entry.record["name"] == "E-TOU-C" + + def test_current_rate_label(self, tmp_path: Path) -> None: + cache = RateCache(tmp_path / "rates_cache.yaml") + assert cache.get_current_rate_label() is None + cache.set_current_rate_label("abc123") + assert cache.get_current_rate_label() == "abc123" + + def test_current_rate_label_persists(self, tmp_path: Path) -> None: + path = tmp_path / "rates_cache.yaml" + cache1 = RateCache(path) + cache1.set_current_rate_label("abc123") + + cache2 = RateCache(path) + assert cache2.get_current_rate_label() == "abc123" + + def test_list_cached_rates(self, tmp_path: Path) -> None: + cache = RateCache(tmp_path / "rates_cache.yaml") + cache.cache_rate("abc123", SAMPLE_URDB_RECORD) + cache.cache_rate("def456", { + "label": "def456", + "utility": "SoCal Edison", + "name": "TOU-D-PRIME", + }) + summaries = cache.list_cached_rates() + assert len(summaries) == 2 + labels = {s["label"] for s in summaries} + assert labels == {"abc123", "def456"} + + def test_openei_config_defaults(self, tmp_path: Path) -> None: + cache = RateCache(tmp_path / "rates_cache.yaml") + config = cache.get_openei_config() + assert config.api_url == "https://api.openei.org/utility_rates" + assert config.api_key == "" + + def test_openei_config_set_and_get(self, tmp_path: Path) -> None: + cache = RateCache(tmp_path / "rates_cache.yaml") + cache.set_openei_config("https://custom.api/rates", "my-key-123") + config = cache.get_openei_config() + assert config.api_url == "https://custom.api/rates" + assert config.api_key == "my-key-123" + + def test_openei_config_persists(self, tmp_path: Path) -> None: + path = tmp_path / "rates_cache.yaml" + cache1 = RateCache(path) + cache1.set_openei_config("https://custom.api/rates", "my-key-123") + + cache2 = RateCache(path) + config = cache2.get_openei_config() + assert config.api_url == "https://custom.api/rates" + assert config.api_key == "my-key-123" + + def test_delete_cached_rate(self, tmp_path: Path) -> None: + cache = RateCache(tmp_path / "rates_cache.yaml") + cache.cache_rate("abc123", SAMPLE_URDB_RECORD) + assert cache.get_cached_rate("abc123") is not None + cache.delete_cached_rate("abc123") + assert cache.get_cached_rate("abc123") is None +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `pytest tests/test_rates/test_cache.py -v` +Expected: FAIL with `ModuleNotFoundError: No module named 'span_panel_simulator.rates.cache'` + +- [ ] **Step 3: Write the cache manager** + +Create `src/span_panel_simulator/rates/cache.py`: + +```python +"""Simulator-wide rate cache backed by a YAML file. + +Stores URDB records verbatim, keyed by their label. Also manages +the current rate selection and OpenEI API configuration. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +import yaml + +from span_panel_simulator.rates.types import ( + AttributionMeta, + OpenEIConfig, + RateCacheEntry, +) + +_DEFAULT_ATTRIBUTION = AttributionMeta( + provider="OpenEI Utility Rate Database", + url="https://openei.org/wiki/Utility_Rate_Database", + license="CC0", + api_version=3, +) + + +class RateCache: + """Manages the simulator-wide rate cache YAML file.""" + + def __init__(self, path: Path) -> None: + self._path = path + self._data = self._load() + + # -- Cache operations ------------------------------------------------ + + def get_cached_rate(self, label: str) -> RateCacheEntry | None: + """Return a cached rate entry by URDB label, or None.""" + rates = self._data.get("rates", {}) + entry = rates.get(label) + if entry is None: + return None + attr_data = entry.get("attribution", {}) + return RateCacheEntry( + source=entry.get("source", "openei_urdb"), + retrieved_at=entry.get("retrieved_at", ""), + attribution=AttributionMeta( + provider=attr_data.get("provider", _DEFAULT_ATTRIBUTION.provider), + url=attr_data.get("url", _DEFAULT_ATTRIBUTION.url), + license=attr_data.get("license", _DEFAULT_ATTRIBUTION.license), + api_version=attr_data.get("api_version", _DEFAULT_ATTRIBUTION.api_version), + ), + record=entry.get("record", {}), + ) + + def cache_rate(self, label: str, urdb_record: dict[str, Any]) -> None: + """Store a URDB record in the cache.""" + if "rates" not in self._data: + self._data["rates"] = {} + self._data["rates"][label] = { + "source": "openei_urdb", + "retrieved_at": datetime.now(timezone.utc).isoformat(), + "attribution": { + "provider": _DEFAULT_ATTRIBUTION.provider, + "url": _DEFAULT_ATTRIBUTION.url, + "license": _DEFAULT_ATTRIBUTION.license, + "api_version": _DEFAULT_ATTRIBUTION.api_version, + }, + "record": urdb_record, + } + self._save() + + def delete_cached_rate(self, label: str) -> None: + """Remove a rate from the cache.""" + rates = self._data.get("rates", {}) + rates.pop(label, None) + self._save() + + def list_cached_rates(self) -> list[dict[str, Any]]: + """Return summary dicts for all cached rates.""" + rates = self._data.get("rates", {}) + summaries = [] + for label, entry in rates.items(): + record = entry.get("record", {}) + summaries.append({ + "label": label, + "utility": record.get("utility", ""), + "name": record.get("name", ""), + "retrieved_at": entry.get("retrieved_at", ""), + }) + return summaries + + # -- Current rate selection ------------------------------------------ + + def get_current_rate_label(self) -> str | None: + """Return the simulator-wide current rate label, or None.""" + label = self._data.get("current_rate_label") + return label if label else None + + def set_current_rate_label(self, label: str) -> None: + """Set the simulator-wide current rate selection.""" + self._data["current_rate_label"] = label + self._save() + + # -- OpenEI configuration ------------------------------------------- + + def get_openei_config(self) -> OpenEIConfig: + """Return the stored OpenEI API settings.""" + cfg = self._data.get("openei", {}) + return OpenEIConfig( + api_url=cfg.get("api_url", OpenEIConfig.api_url), + api_key=cfg.get("api_key", OpenEIConfig.api_key), + ) + + def set_openei_config(self, api_url: str, api_key: str) -> None: + """Update the OpenEI API settings.""" + self._data["openei"] = { + "api_url": api_url, + "api_key": api_key, + } + self._save() + + # -- Persistence ----------------------------------------------------- + + def _load(self) -> dict[str, Any]: + if self._path.exists(): + with open(self._path) as f: + data = yaml.safe_load(f) + return data if isinstance(data, dict) else {} + return {} + + def _save(self) -> None: + self._path.parent.mkdir(parents=True, exist_ok=True) + with open(self._path, "w") as f: + yaml.dump( + self._data, + f, + default_flow_style=False, + sort_keys=False, + allow_unicode=True, + ) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `pytest tests/test_rates/test_cache.py -v` +Expected: All 10 tests PASS + +- [ ] **Step 5: Commit** + +```bash +git add src/span_panel_simulator/rates/cache.py tests/test_rates/test_cache.py +git commit -m "Add rate cache manager for URDB record persistence" +``` + +--- + +### Task 5: OpenEI API Client + +**Files:** +- Create: `src/span_panel_simulator/rates/openei.py` +- Create: `tests/test_rates/test_openei.py` + +- [ ] **Step 1: Write the failing tests** + +Create `tests/test_rates/test_openei.py`: + +```python +"""Tests for the OpenEI URDB API client (mocked HTTP).""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, patch + +import pytest + +from span_panel_simulator.rates.openei import ( + OpenEIError, + fetch_rate_detail, + fetch_rate_plans, + fetch_utilities, +) + +API_URL = "https://api.openei.org/utility_rates" +API_KEY = "test-key" + + +def _mock_response(json_data: dict, status: int = 200) -> AsyncMock: + """Create a mock aiohttp response.""" + resp = AsyncMock() + resp.status = status + resp.json = AsyncMock(return_value=json_data) + resp.text = AsyncMock(return_value=str(json_data)) + resp.__aenter__ = AsyncMock(return_value=resp) + resp.__aexit__ = AsyncMock(return_value=False) + return resp + + +class TestFetchUtilities: + """Fetch utilities by lat/lon.""" + + @pytest.mark.asyncio + async def test_returns_utility_summaries(self) -> None: + response_data = { + "items": [ + {"utility_name": "Pacific Gas & Electric Co", "eia": "14328"}, + {"utility_name": "City of Palo Alto", "eia": "14328"}, + ] + } + with patch("span_panel_simulator.rates.openei._get_json", return_value=response_data): + result = await fetch_utilities(37.7, -122.4, API_URL, API_KEY) + assert len(result) >= 1 + assert result[0].utility_name == "Pacific Gas & Electric Co" + + @pytest.mark.asyncio + async def test_empty_result(self) -> None: + with patch("span_panel_simulator.rates.openei._get_json", return_value={"items": []}): + result = await fetch_utilities(0.0, 0.0, API_URL, API_KEY) + assert result == [] + + +class TestFetchRatePlans: + """Fetch rate plans for a utility.""" + + @pytest.mark.asyncio + async def test_returns_plan_summaries(self) -> None: + response_data = { + "items": [ + { + "label": "abc123", + "name": "E-TOU-C", + "startdate": 1672531200, + "enddate": None, + "description": "Time of use residential", + }, + { + "label": "def456", + "name": "E-TOU-D", + "startdate": 1672531200, + "enddate": 1704067200, + "description": "Legacy TOU", + }, + ] + } + with patch("span_panel_simulator.rates.openei._get_json", return_value=response_data): + result = await fetch_rate_plans("Pacific Gas & Electric Co", API_URL, API_KEY) + assert len(result) == 2 + assert result[0].label == "abc123" + assert result[0].name == "E-TOU-C" + assert result[1].enddate == 1704067200 + + +class TestFetchRateDetail: + """Fetch full rate detail by label.""" + + @pytest.mark.asyncio + async def test_returns_full_record(self) -> None: + response_data = { + "items": [ + { + "label": "abc123", + "utility": "PG&E", + "name": "E-TOU-C", + "energyratestructure": [[{"rate": 0.25}]], + "energyweekdayschedule": [[0] * 24 for _ in range(12)], + "energyweekendschedule": [[0] * 24 for _ in range(12)], + } + ] + } + with patch("span_panel_simulator.rates.openei._get_json", return_value=response_data): + result = await fetch_rate_detail("abc123", API_URL, API_KEY) + assert result["label"] == "abc123" + assert result["energyratestructure"] == [[{"rate": 0.25}]] + + @pytest.mark.asyncio + async def test_label_not_found_raises(self) -> None: + with patch("span_panel_simulator.rates.openei._get_json", return_value={"items": []}): + with pytest.raises(OpenEIError, match="not found"): + await fetch_rate_detail("nonexistent", API_URL, API_KEY) + + @pytest.mark.asyncio + async def test_api_error_raises(self) -> None: + with patch( + "span_panel_simulator.rates.openei._get_json", + side_effect=OpenEIError("HTTP 401: Unauthorized"), + ): + with pytest.raises(OpenEIError, match="401"): + await fetch_rate_detail("abc123", API_URL, API_KEY) +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `pytest tests/test_rates/test_openei.py -v` +Expected: FAIL with `ModuleNotFoundError: No module named 'span_panel_simulator.rates.openei'` + +- [ ] **Step 3: Write the OpenEI client** + +Create `src/span_panel_simulator/rates/openei.py`: + +```python +"""OpenEI URDB API client. + +Fetches utility and rate plan data from the OpenEI Utility Rate +Database. All functions accept api_url and api_key so the base URL +and credentials are caller-configurable. +""" + +from __future__ import annotations + +import logging +from typing import Any + +import aiohttp + +from span_panel_simulator.rates.types import RatePlanSummary, UtilitySummary + +_LOG = logging.getLogger(__name__) + + +class OpenEIError(Exception): + """Raised when the URDB API returns an error or unexpected response.""" + + +async def _get_json(url: str, params: dict[str, str]) -> dict[str, Any]: + """Issue a GET request and return the parsed JSON response.""" + async with aiohttp.ClientSession() as session: + async with session.get(url, params=params) as resp: + if resp.status != 200: + body = await resp.text() + raise OpenEIError(f"HTTP {resp.status}: {body[:200]}") + data: dict[str, Any] = await resp.json(content_type=None) + return data + + +async def fetch_utilities( + lat: float, + lon: float, + api_url: str, + api_key: str, +) -> list[UtilitySummary]: + """Fetch utilities near a lat/lon from URDB. + + Returns de-duplicated utilities sorted by name. + """ + params = { + "version": "3", + "format": "json", + "api_key": api_key, + "lat": str(lat), + "lon": str(lon), + "sector": "Residential", + "detail": "minimal", + } + data = await _get_json(api_url, params) + items = data.get("items", []) + + seen: set[str] = set() + utilities: list[UtilitySummary] = [] + for item in items: + name = item.get("utility_name", item.get("utility", "")) + if not name or name in seen: + continue + seen.add(name) + utilities.append(UtilitySummary( + utility_name=name, + eia_id=str(item.get("eia", "")), + )) + utilities.sort(key=lambda u: u.utility_name) + return utilities + + +async def fetch_rate_plans( + utility: str, + api_url: str, + api_key: str, + sector: str = "Residential", +) -> list[RatePlanSummary]: + """Fetch available rate plans for a utility.""" + params = { + "version": "3", + "format": "json", + "api_key": api_key, + "ratesforutility": utility, + "sector": sector, + "detail": "minimal", + } + data = await _get_json(api_url, params) + items = data.get("items", []) + + plans: list[RatePlanSummary] = [] + for item in items: + plans.append(RatePlanSummary( + label=item.get("label", ""), + name=item.get("name", ""), + startdate=item.get("startdate", 0), + enddate=item.get("enddate"), + description=item.get("description", ""), + )) + return plans + + +async def fetch_rate_detail( + label: str, + api_url: str, + api_key: str, +) -> dict[str, Any]: + """Fetch the full rate record for a URDB label. + + Returns the raw URDB record dict (to be stored verbatim). + Raises OpenEIError if the label is not found. + """ + params = { + "version": "3", + "format": "json", + "api_key": api_key, + "getpage": label, + "detail": "full", + } + data = await _get_json(api_url, params) + items = data.get("items", []) + if not items: + raise OpenEIError(f"Rate plan '{label}' not found in URDB") + record: dict[str, Any] = items[0] + return record +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `pytest tests/test_rates/test_openei.py -v` +Expected: All 5 tests PASS + +- [ ] **Step 5: Commit** + +```bash +git add src/span_panel_simulator/rates/openei.py tests/test_rates/test_openei.py +git commit -m "Add OpenEI URDB API client for rate plan discovery" +``` + +--- + +## Phase 3: API Endpoints + +### Task 6: Rate API Routes + +**Files:** +- Modify: `src/span_panel_simulator/dashboard/routes.py` + +This task adds all rate-related HTTP endpoints. The routes follow the existing pattern: thin handlers that parse the request, call the rates package, and return JSON. + +- [ ] **Step 1: Add RateCache initialization** + +The `RateCache` instance needs to be accessible from route handlers. Add it to the dashboard app keys. + +Modify `src/span_panel_simulator/dashboard/keys.py` — add a new app key: + +```python +APP_KEY_RATE_CACHE = web.AppKey("rate_cache", default=None) +``` + +Note: Read the file first to see the exact pattern for existing keys, then add the new key following the same pattern. + +- [ ] **Step 2: Initialize RateCache in the app factory** + +Find where the dashboard app is created (likely `dashboard/app.py` or similar) and add: + +```python +from span_panel_simulator.rates.cache import RateCache + +# In the app setup function, after config_dir is available: +rate_cache = RateCache(config_dir / "rates_cache.yaml") +app[APP_KEY_RATE_CACHE] = rate_cache +``` + +Note: Read the app factory file to find the exact location and follow the existing pattern for setting app keys. + +- [ ] **Step 3: Add rate route handlers to routes.py** + +Add to `src/span_panel_simulator/dashboard/routes.py`: + +```python +from span_panel_simulator.rates.cache import RateCache +from span_panel_simulator.rates.openei import OpenEIError, fetch_rate_detail, fetch_rate_plans, fetch_utilities + +def _rate_cache(request: web.Request) -> RateCache: + return request.app[APP_KEY_RATE_CACHE] + + +# -- Rate endpoints -- + + +async def handle_get_openei_config(request: web.Request) -> web.Response: + """GET /rates/openei-config — return current API URL and key.""" + config = _rate_cache(request).get_openei_config() + return web.json_response({ + "api_url": config.api_url, + "api_key": config.api_key, + }) + + +async def handle_put_openei_config(request: web.Request) -> web.Response: + """PUT /rates/openei-config — update API URL and key.""" + body = await request.json() + api_url = body.get("api_url", "").strip() + api_key = body.get("api_key", "").strip() + if not api_url or not api_key: + return web.json_response({"error": "api_url and api_key are required"}, status=400) + _rate_cache(request).set_openei_config(api_url, api_key) + return web.json_response({"ok": True}) + + +async def handle_get_utilities(request: web.Request) -> web.Response: + """GET /rates/utilities?lat=&lon= — utilities near location.""" + lat = request.query.get("lat") + lon = request.query.get("lon") + if lat is None or lon is None: + return web.json_response({"error": "lat and lon are required"}, status=400) + config = _rate_cache(request).get_openei_config() + if not config.api_key: + return web.json_response({"error": "OpenEI API key not configured"}, status=400) + try: + results = await fetch_utilities(float(lat), float(lon), config.api_url, config.api_key) + except OpenEIError as e: + return web.json_response({"error": str(e)}, status=502) + return web.json_response([ + {"utility_name": u.utility_name, "eia_id": u.eia_id} + for u in results + ]) + + +async def handle_get_rate_plans(request: web.Request) -> web.Response: + """GET /rates/plans?utility=§or= — rate plans for a utility.""" + utility = request.query.get("utility") + if not utility: + return web.json_response({"error": "utility is required"}, status=400) + sector = request.query.get("sector", "Residential") + config = _rate_cache(request).get_openei_config() + if not config.api_key: + return web.json_response({"error": "OpenEI API key not configured"}, status=400) + try: + plans = await fetch_rate_plans(utility, config.api_url, config.api_key, sector) + except OpenEIError as e: + return web.json_response({"error": str(e)}, status=502) + return web.json_response([ + { + "label": p.label, + "name": p.name, + "startdate": p.startdate, + "enddate": p.enddate, + "description": p.description, + } + for p in plans + ]) + + +async def handle_fetch_rate(request: web.Request) -> web.Response: + """POST /rates/fetch {label} — fetch from URDB and cache.""" + body = await request.json() + label = body.get("label", "").strip() + if not label: + return web.json_response({"error": "label is required"}, status=400) + config = _rate_cache(request).get_openei_config() + if not config.api_key: + return web.json_response({"error": "OpenEI API key not configured"}, status=400) + try: + record = await fetch_rate_detail(label, config.api_url, config.api_key) + except OpenEIError as e: + return web.json_response({"error": str(e)}, status=502) + cache = _rate_cache(request) + cache.cache_rate(label, record) + return web.json_response({ + "label": label, + "utility": record.get("utility", ""), + "name": record.get("name", ""), + }) + + +async def handle_refresh_rate(request: web.Request) -> web.Response: + """POST /rates/refresh {label} — re-fetch a cached rate.""" + body = await request.json() + label = body.get("label", "").strip() + if not label: + return web.json_response({"error": "label is required"}, status=400) + config = _rate_cache(request).get_openei_config() + if not config.api_key: + return web.json_response({"error": "OpenEI API key not configured"}, status=400) + try: + record = await fetch_rate_detail(label, config.api_url, config.api_key) + except OpenEIError as e: + return web.json_response({"error": str(e)}, status=502) + _rate_cache(request).cache_rate(label, record) + return web.json_response({"ok": True, "label": label}) + + +async def handle_get_rates_cache(request: web.Request) -> web.Response: + """GET /rates/cache — list all cached rate summaries.""" + return web.json_response(_rate_cache(request).list_cached_rates()) + + +async def handle_get_current_rate(request: web.Request) -> web.Response: + """GET /rates/current — current rate label and summary.""" + cache = _rate_cache(request) + label = cache.get_current_rate_label() + if label is None: + return web.json_response({"label": None}) + entry = cache.get_cached_rate(label) + if entry is None: + return web.json_response({"label": label, "error": "cached record missing"}) + return web.json_response({ + "label": label, + "utility": entry.record.get("utility", ""), + "name": entry.record.get("name", ""), + "retrieved_at": entry.retrieved_at, + }) + + +async def handle_put_current_rate(request: web.Request) -> web.Response: + """PUT /rates/current {label} — set simulator-wide current rate.""" + body = await request.json() + label = body.get("label", "").strip() + if not label: + return web.json_response({"error": "label is required"}, status=400) + _rate_cache(request).set_current_rate_label(label) + return web.json_response({"ok": True}) + + +async def handle_get_rate_detail(request: web.Request) -> web.Response: + """GET /rates/detail/{label} — full cached record.""" + label = request.match_info["label"] + entry = _rate_cache(request).get_cached_rate(label) + if entry is None: + return web.json_response({"error": "not found"}, status=404) + return web.json_response(entry.record) + + +async def handle_get_rate_attribution(request: web.Request) -> web.Response: + """GET /rates/attribution/{label} — attribution metadata.""" + label = request.match_info["label"] + entry = _rate_cache(request).get_cached_rate(label) + if entry is None: + return web.json_response({"error": "not found"}, status=404) + return web.json_response({ + "provider": entry.attribution.provider, + "url": entry.attribution.url, + "license": entry.attribution.license, + "api_version": entry.attribution.api_version, + "retrieved_at": entry.retrieved_at, + }) +``` + +- [ ] **Step 4: Register the routes in setup_routes** + +Add to the `setup_routes` function in `routes.py`, after the existing modeling-data route: + +```python + # Rate plan management + app.router.add_get("/rates/openei-config", handle_get_openei_config) + app.router.add_put("/rates/openei-config", handle_put_openei_config) + app.router.add_get("/rates/utilities", handle_get_utilities) + app.router.add_get("/rates/plans", handle_get_rate_plans) + app.router.add_post("/rates/fetch", handle_fetch_rate) + app.router.add_post("/rates/refresh", handle_refresh_rate) + app.router.add_get("/rates/cache", handle_get_rates_cache) + app.router.add_get("/rates/current", handle_get_current_rate) + app.router.add_put("/rates/current", handle_put_current_rate) + app.router.add_get("/rates/detail/{label}", handle_get_rate_detail) + app.router.add_get("/rates/attribution/{label}", handle_get_rate_attribution) +``` + +- [ ] **Step 5: Run the existing test suite to verify no regressions** + +Run: `pytest tests/ -v --timeout=30` +Expected: All existing tests PASS + +- [ ] **Step 6: Commit** + +```bash +git add src/span_panel_simulator/dashboard/routes.py src/span_panel_simulator/dashboard/keys.py +git commit -m "Add rate API endpoints for URDB discovery, caching, and selection" +``` + +Note: Also `git add` the app factory file if it was modified in step 2. + +--- + +## Phase 4: Engine Integration + +### Task 7: Wire Cost Engine into compute_modeling_data + +**Files:** +- Modify: `src/span_panel_simulator/engine.py` +- Modify: `src/span_panel_simulator/dashboard/routes.py` (pass rate cache to engine) + +The cost engine is called **after** the Before/After power arrays are computed. The engine does not resolve rates or compute costs — it delegates to `compute_costs`. + +- [ ] **Step 1: Add cost calculation to compute_modeling_data** + +Read `engine.py` and find the return dict at the end of `compute_modeling_data` (around line 1562). Add cost calculation just before the return: + +```python +from span_panel_simulator.rates.cache import RateCache +from span_panel_simulator.rates.cost_engine import compute_costs +``` + +Add a `rate_cache` parameter to `compute_modeling_data` and an optional `proposed_rate_label`: + +```python +async def compute_modeling_data( + self, + horizon_hours: int, + rate_cache: RateCache | None = None, + proposed_rate_label: str | None = None, +) -> dict[str, Any]: +``` + +Just before the `return` statement, add: + +```python + # -- Cost calculation (optional, requires rate cache) -------- + result: dict[str, Any] = { + "horizon_start": int(horizon_start), + "horizon_end": int(horizon_end), + "resolution_s": 3600, + "time_zone": tz_str, + "timestamps": [int(t) for t in timestamps], + "site_power": site_power_arr, + "grid_power": grid_power_arr, + "pv_power_before": pv_before_arr, + "pv_power_after": pv_after_arr, + "pv_power": pv_after_arr, + "battery_power": battery_power_arr, + "battery_power_before": battery_before_arr, + "circuits": circuits_response, + } + + if rate_cache is not None: + current_label = rate_cache.get_current_rate_label() + if current_label is not None: + current_entry = rate_cache.get_cached_rate(current_label) + if current_entry is not None: + ts_list = result["timestamps"] + before_costs = compute_costs( + ts_list, site_power_arr, current_entry.record, tz_str, + ) + result["before_costs"] = { + "import_cost": round(before_costs.import_cost, 2), + "export_credit": round(before_costs.export_credit, 2), + "fixed_charges": round(before_costs.fixed_charges, 2), + "net_cost": round(before_costs.net_cost, 2), + } + + # After: use proposed rate if set, otherwise current + after_record = current_entry.record + if proposed_rate_label: + proposed_entry = rate_cache.get_cached_rate(proposed_rate_label) + if proposed_entry is not None: + after_record = proposed_entry.record + after_costs = compute_costs( + ts_list, grid_power_arr, after_record, tz_str, + ) + result["after_costs"] = { + "import_cost": round(after_costs.import_cost, 2), + "export_credit": round(after_costs.export_credit, 2), + "fixed_charges": round(after_costs.fixed_charges, 2), + "net_cost": round(after_costs.net_cost, 2), + } + + return result +``` + +- [ ] **Step 2: Update the DashboardContext to pass rate_cache** + +Modify `src/span_panel_simulator/dashboard/context.py` to extend the `get_modeling_data` callable signature, or alternatively update `handle_modeling_data` in `routes.py` to call the cost engine itself after receiving the result. + +The cleaner approach: keep the engine method signature unchanged, and compute costs in the route handler. This avoids changing the DashboardContext interface. Update `handle_modeling_data`: + +```python +async def handle_modeling_data(request: web.Request) -> web.Response: + """Return time-series for Before/After energy comparison.""" + ctx = _ctx(request) + horizon_key = request.query.get("horizon", "1mo") + horizon_hours = _HORIZON_MAP.get(horizon_key, 730) + + config_file = resolve_modeling_config_filename(ctx, request.query.get("config")) + result = await ctx.get_modeling_data(horizon_hours, config_file) + if result is None: + return web.json_response({"error": "No running simulation"}, status=503) + if "error" in result: + return web.json_response(result, status=400) + + # Attach cost data if rate cache is available + cache = _rate_cache(request) + if cache is not None: + proposed_label = request.query.get("proposed_rate_label") + _attach_costs(result, cache, proposed_label) + + return web.json_response(result) +``` + +Add the helper function: + +```python +def _attach_costs( + result: dict[str, Any], + cache: RateCache, + proposed_rate_label: str | None, +) -> None: + """Add before_costs and after_costs to a modeling result dict.""" + current_label = cache.get_current_rate_label() + if current_label is None: + return + current_entry = cache.get_cached_rate(current_label) + if current_entry is None: + return + + tz_str = result["time_zone"] + ts_list = result["timestamps"] + + before_costs = compute_costs(ts_list, result["site_power"], current_entry.record, tz_str) + result["before_costs"] = { + "import_cost": round(before_costs.import_cost, 2), + "export_credit": round(before_costs.export_credit, 2), + "fixed_charges": round(before_costs.fixed_charges, 2), + "net_cost": round(before_costs.net_cost, 2), + } + + after_record = current_entry.record + if proposed_rate_label: + proposed_entry = cache.get_cached_rate(proposed_rate_label) + if proposed_entry is not None: + after_record = proposed_entry.record + after_costs = compute_costs(ts_list, result["grid_power"], after_record, tz_str) + result["after_costs"] = { + "import_cost": round(after_costs.import_cost, 2), + "export_credit": round(after_costs.export_credit, 2), + "fixed_charges": round(after_costs.fixed_charges, 2), + "net_cost": round(after_costs.net_cost, 2), + } +``` + +This approach is better because: +- `engine.py` stays focused on energy calculation (respects AGENTS.md boundary) +- No changes to `DashboardContext` interface +- Cost calculation is a route-level concern — decorating the response + +- [ ] **Step 3: Run full test suite** + +Run: `pytest tests/ -v --timeout=30` +Expected: All tests PASS (no engine signature changes) + +- [ ] **Step 4: Commit** + +```bash +git add src/span_panel_simulator/dashboard/routes.py +git commit -m "Wire cost engine into modeling-data response via route handler" +``` + +--- + +## Phase 5: Modeling View UI + +### Task 8: Rate Plan Selection UI + +**Files:** +- Modify: `src/span_panel_simulator/dashboard/templates/partials/modeling_view.html` + +This task adds the rate plan selection section and OpenEI dialog to the modeling view HTML. The JavaScript interacts with the rate API endpoints from Task 6. + +- [ ] **Step 1: Add rate plan section HTML** + +Insert after the `` div (line 34) and before `` (line 36) in `modeling_view.html`: + +```html + + + + + + + + +``` + +- [ ] **Step 2: Add rate plan JavaScript** + +Add to the ` diff --git a/src/span_panel_simulator/dashboard/templates/partials/entity_edit.html b/src/span_panel_simulator/dashboard/templates/partials/entity_edit.html index 1eb3968..1bbf607 100644 --- a/src/span_panel_simulator/dashboard/templates/partials/entity_edit.html +++ b/src/span_panel_simulator/dashboard/templates/partials/entity_edit.html @@ -21,7 +21,7 @@

Editing: {{ e.name }}

{% endif %} - {% if e.entity_type != 'pv' %} + {% if e.entity_type not in ('pv', 'battery') %} {% endif %} + {% if e.entity_type != 'battery' %} + {% endif %} {% if e.entity_type != 'battery' %} + {% elif e.entity_type == 'battery' %} + {# Battery: Energy Profile is not user-facing — inverter rates are in Battery Behavior below #} {% else %}
Energy Profile diff --git a/src/span_panel_simulator/dashboard/templates/partials/modeling_view.html b/src/span_panel_simulator/dashboard/templates/partials/modeling_view.html index edc1b49..1a1acb6 100644 --- a/src/span_panel_simulator/dashboard/templates/partials/modeling_view.html +++ b/src/span_panel_simulator/dashboard/templates/partials/modeling_view.html @@ -33,27 +33,65 @@

Modeling —

Loading modeling data... - +