183 lines
7.5 KiB
Python
183 lines
7.5 KiB
Python
"""Smoke tests for the battery state engine and oracle."""
|
||
from __future__ import annotations
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
import pytest
|
||
|
||
from pluginbattery.sim import (
|
||
Battery,
|
||
apply_nl_tariff,
|
||
oracle_daily_schedule,
|
||
simulate,
|
||
synthesize_pv,
|
||
)
|
||
|
||
|
||
def make_df(prices: list[float], demands_kwh: list[float]) -> pd.DataFrame:
|
||
idx = pd.date_range("2024-01-01", periods=len(prices), freq="h", tz="UTC")
|
||
return pd.DataFrame(
|
||
{
|
||
"eur_per_kwh": prices,
|
||
"power_w": np.array(demands_kwh) * 1000.0,
|
||
"irradiance_w_m2": 0.0,
|
||
"demand_kwh": demands_kwh,
|
||
},
|
||
index=idx,
|
||
)
|
||
|
||
|
||
def test_simulate_clamps_charge_to_capacity():
|
||
df = make_df([0.1] * 6, [0.5] * 6)
|
||
bat = Battery(capacity_kwh=1.0, max_charge_kw=0.8, max_discharge_kw=0.8,
|
||
round_trip_eff=1.0)
|
||
schedule = np.array([[0.8, 0.0]] * 6) # ask for full charge every hour
|
||
out = simulate(df, bat, schedule)
|
||
assert out["soc_kwh"].max() <= 1.0 + 1e-9
|
||
assert out["soc_kwh"].min() >= 0.0
|
||
|
||
|
||
def test_plugin_never_exports():
|
||
df = make_df([0.5, 0.5, 0.5], [0.3, 0.3, 0.3])
|
||
bat = Battery(capacity_kwh=2.0, max_charge_kw=0.8, max_discharge_kw=0.8,
|
||
round_trip_eff=1.0, allows_export=False, initial_soc_kwh=2.0)
|
||
schedule = np.array([[0.0, 0.8]] * 3) # try to dump at full power
|
||
out = simulate(df, bat, schedule)
|
||
assert (out["discharge_kwh"] <= out["demand_kwh"] + 1e-9).all()
|
||
assert (out["grid_kwh_with_battery"] >= -1e-9).all()
|
||
|
||
|
||
def test_greedy_does_not_grid_arbitrage_without_pv():
|
||
"""Greedy dispatch is self-consumption only — it won't import cheap and
|
||
discharge expensive without a surplus source. (LP would; greedy by
|
||
design doesn't, matching real plug-in firmware.)"""
|
||
prices = [0.05] * 12 + [0.50] * 12
|
||
demands = [1.0] * 24
|
||
df = make_df(prices, demands)
|
||
bat = Battery(capacity_kwh=2.0, max_charge_kw=0.8, max_discharge_kw=0.8,
|
||
round_trip_eff=0.9, allows_export=False)
|
||
schedule = oracle_daily_schedule(df, bat)
|
||
out = simulate(df, bat, schedule)
|
||
assert out["charge_kwh"].sum() == 0
|
||
assert out["discharge_kwh"].sum() == 0
|
||
|
||
|
||
def test_greedy_fills_from_surplus_then_overflows():
|
||
"""With PV surplus, greedy fills the battery as fast as power allows
|
||
until capacity is reached, then lets the rest export."""
|
||
prices = [0.20] * 24
|
||
demand = [0.1] * 24
|
||
pv = [0.0] * 6 + [3.0] * 6 + [0.0] * 12 # 6 sunny hours, 3 kWh/h surplus
|
||
df = make_df(prices, demand)
|
||
df["pv_kwh"] = pv
|
||
bat = Battery(capacity_kwh=2.0, max_charge_kw=0.8, max_discharge_kw=0.8,
|
||
round_trip_eff=1.0, allows_export=False)
|
||
schedule = oracle_daily_schedule(df, bat)
|
||
out = simulate(df, bat, schedule)
|
||
# Charging happens during the first surplus hours, capped at 0.8 kW
|
||
assert out["charge_kwh"].iloc[6:9].sum() == pytest.approx(2.0, abs=1e-6)
|
||
# Once full, no more charging even though surplus continues
|
||
assert out["charge_kwh"].iloc[9:12].sum() == 0
|
||
# Battery discharges into evening demand
|
||
assert out["discharge_kwh"].iloc[12:].sum() > 0
|
||
|
||
|
||
def test_apply_nl_tariff_matches_user_formula():
|
||
df = make_df([0.0, 0.10, -0.05], [1.0, 1.0, 1.0])
|
||
out = apply_nl_tariff(df)
|
||
expected = [0.0 * 1.21 + 0.136, 0.10 * 1.21 + 0.136, -0.05 * 1.21 + 0.136]
|
||
assert np.allclose(out["eur_per_kwh"].to_numpy(), expected)
|
||
assert np.allclose(out["epex_eur_per_kwh"].to_numpy(), [0.0, 0.10, -0.05])
|
||
|
||
|
||
def test_fixed_tax_reduces_optimal_cycle_count():
|
||
"""A flat per-kWh charge makes round-trip losses more expensive, so the
|
||
oracle should run fewer cycles when the same EPEX series is consumer-priced."""
|
||
prices = [0.05] * 12 + [0.20] * 12
|
||
df_raw = make_df(prices, [1.0] * 24)
|
||
df_consumer = apply_nl_tariff(df_raw)
|
||
bat = Battery(capacity_kwh=2.0, max_charge_kw=0.8, max_discharge_kw=0.8,
|
||
round_trip_eff=0.9, allows_export=False)
|
||
sched_raw = oracle_daily_schedule(df_raw, bat)
|
||
sched_cons = oracle_daily_schedule(df_consumer, bat)
|
||
# On consumer prices the opportunity cost of efficiency loss is higher,
|
||
# so total charge_kwh should not increase (often decreases or stays equal).
|
||
assert sched_cons[:, 0].sum() <= sched_raw[:, 0].sum() + 1e-6
|
||
|
||
|
||
def test_oracle_skips_arbitrage_when_eff_kills_it():
|
||
# Spread 0.20 → 0.21 against η_rt=0.5 means every cycle loses money.
|
||
df = make_df([0.20] * 12 + [0.21] * 12, [1.0] * 24)
|
||
bat = Battery(capacity_kwh=2.0, max_charge_kw=0.8, max_discharge_kw=0.8,
|
||
round_trip_eff=0.5, allows_export=False)
|
||
schedule = oracle_daily_schedule(df, bat)
|
||
out = simulate(df, bat, schedule)
|
||
assert out["savings"].sum() < 1e-6
|
||
|
||
|
||
def test_synthesize_pv_hits_target_annual():
|
||
"""synthesize_pv should calibrate so annual output ≈ target × kWp."""
|
||
n = 24 * 30 # 30 days
|
||
idx = pd.date_range("2024-06-01", periods=n, freq="h", tz="UTC")
|
||
# Simple square wave: 600 W/m² for 8 daylight hours, zero otherwise.
|
||
irr = np.zeros(n)
|
||
for d in range(30):
|
||
irr[d * 24 + 8 : d * 24 + 16] = 600.0
|
||
df = pd.DataFrame({
|
||
"eur_per_kwh": 0.20, "power_w": 0.0,
|
||
"irradiance_w_m2": irr, "demand_kwh": 0.0,
|
||
}, index=idx)
|
||
out = synthesize_pv(df, kwp=3.0, target_kwh_per_kwp_per_year=900.0)
|
||
annual_pv = out["pv_kwh"].sum() * (8766 / n)
|
||
assert abs(annual_pv - 3.0 * 900.0) < 1.0
|
||
|
||
|
||
def test_plugin_with_pv_does_not_push_to_grid():
|
||
"""Plug-in battery + surplus solar: discharge must be 0 in surplus hours."""
|
||
df = make_df([0.30] * 24, [0.5] * 24)
|
||
df["pv_kwh"] = [3.0] * 12 + [0.0] * 12 # huge midday surplus
|
||
bat = Battery(capacity_kwh=2.0, max_charge_kw=0.8, max_discharge_kw=0.8,
|
||
round_trip_eff=0.9, allows_export=False, initial_soc_kwh=2.0)
|
||
schedule = np.array([[0.0, 0.8]] * 24) # try to dump every hour
|
||
out = simulate(df, bat, schedule)
|
||
surplus_hours = out["pv_kwh"] > out["demand_kwh"]
|
||
assert (out.loc[surplus_hours, "discharge_kwh"] == 0).all()
|
||
|
||
|
||
def test_no_saldering_increases_battery_savings():
|
||
"""Removing saldering should make a battery on a PV system more valuable.
|
||
|
||
Reason: surplus solar that previously credited at consumer price now only
|
||
earns raw EPEX. Storing it for later self-consumption is now strictly
|
||
better than the previous opportunity cost.
|
||
"""
|
||
# Day with cheap morning EPEX, noon surplus solar, expensive evening.
|
||
prices_consumer = [0.20] * 6 + [0.15] * 6 + [0.40] * 12
|
||
prices_epex = [0.05] * 6 + [0.02] * 6 + [0.20] * 12 # before VAT/tax
|
||
demands = [0.5] * 24
|
||
pv = [0.0] * 8 + [3.0] * 6 + [0.0] * 10
|
||
n = 24
|
||
idx = pd.date_range("2024-01-01", periods=n, freq="h", tz="UTC")
|
||
base = pd.DataFrame({
|
||
"eur_per_kwh": prices_consumer,
|
||
"power_w": np.array(demands) * 1000.0,
|
||
"irradiance_w_m2": 0.0,
|
||
"demand_kwh": demands,
|
||
"pv_kwh": pv,
|
||
"epex_eur_per_kwh": prices_epex,
|
||
}, index=idx)
|
||
bat = Battery(capacity_kwh=5.0, max_charge_kw=2.5, max_discharge_kw=2.5,
|
||
round_trip_eff=0.9, allows_export=False)
|
||
|
||
# With full saldering (export = import).
|
||
df_sald = base.copy()
|
||
out_sald = simulate(df_sald, bat, oracle_daily_schedule(df_sald, bat))
|
||
savings_sald = out_sald["savings"].sum()
|
||
|
||
# Without saldering (export = raw EPEX).
|
||
df_no = base.copy()
|
||
df_no["export_eur_per_kwh"] = df_no["epex_eur_per_kwh"]
|
||
out_no = simulate(df_no, bat, oracle_daily_schedule(df_no, bat))
|
||
savings_no = out_no["savings"].sum()
|
||
|
||
assert savings_no > savings_sald
|