mattandClaude Sonnet 5
298724a9d3
CI / test (push) Failing after 45m4s
Implements the package from PLAN.md: config/data loading, the LP-first MILP-fallback rolling-horizon optimiser, battery degradation tracking, independent schedule validation, metrics, plots and the CLI, plus the matching test suite. Adds a Gitea Actions workflow (lint + tests) that posts a pass/fail notification to ntfy on every run, and a .gitignore for build/cache artefacts that had been tracked by mistake. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
130 lines
4.6 KiB
Python
130 lines
4.6 KiB
Python
"""The validator must catch each violation type when it is injected."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
import pytest
|
|
from conftest import prices_frame
|
|
|
|
from battery_dispatch.config import RunConfig
|
|
from battery_dispatch.optimiser import run_rolling_horizon
|
|
from battery_dispatch.validation import validate_schedule
|
|
|
|
|
|
@pytest.fixture
|
|
def schedule(spec) -> pd.DataFrame:
|
|
"""A genuine, valid two-day schedule to mutate in each test."""
|
|
prices = prices_frame(
|
|
[0.0] * 20 + [90.0] * 8 + [10.0] * 20 + [5.0] * 16 + [120.0] * 8 + [40.0] * 24,
|
|
[30.0] * 48,
|
|
)
|
|
result, _, _ = run_rolling_horizon(prices, spec, RunConfig())
|
|
return result
|
|
|
|
|
|
def test_a_valid_schedule_passes_every_check(schedule, spec):
|
|
report = validate_schedule(schedule, spec)
|
|
assert report.ok, report.summary()
|
|
assert len(report.checks) == 7
|
|
assert "all 7 validation checks passed" in report.summary()
|
|
|
|
|
|
def test_catches_charging_above_the_power_limit(schedule, spec):
|
|
schedule.loc[schedule.index[0], "charge_market_1_mw"] = 5.0
|
|
report = validate_schedule(schedule, spec)
|
|
|
|
assert not report.ok
|
|
assert report.checks["charge_within_limit"] is False
|
|
assert any("combined charge" in failure for failure in report.failures)
|
|
|
|
|
|
def test_catches_discharging_above_the_power_limit(schedule, spec):
|
|
schedule.loc[schedule.index[30], "discharge_market_2_mw"] = 3.5
|
|
report = validate_schedule(schedule, spec)
|
|
|
|
assert not report.ok
|
|
assert report.checks["discharge_within_limit"] is False
|
|
|
|
|
|
def test_catches_simultaneous_charge_and_discharge(schedule, spec):
|
|
row = schedule.index[0]
|
|
schedule.loc[row, "charge_market_1_mw"] = 1.0
|
|
schedule.loc[row, "discharge_market_1_mw"] = 1.0
|
|
report = validate_schedule(schedule, spec)
|
|
|
|
assert not report.ok
|
|
assert report.checks["no_simultaneous_charge_discharge"] is False
|
|
assert report.details["simultaneous_half_hours"] == 1.0
|
|
|
|
|
|
def test_catches_market_2_power_varying_within_an_hour(schedule, spec):
|
|
# Break only the second half-hour of hour zero.
|
|
schedule.loc[schedule.index[1], "discharge_market_2_mw"] = 1.0
|
|
schedule.loc[schedule.index[0], "discharge_market_2_mw"] = 0.0
|
|
report = validate_schedule(schedule, spec)
|
|
|
|
assert not report.ok
|
|
assert report.checks["hourly_commitment_constant"] is False
|
|
|
|
|
|
def test_catches_a_state_of_charge_that_does_not_follow_the_power_flows(
|
|
schedule, spec
|
|
):
|
|
schedule.loc[schedule.index[10], "soc_mwh"] += 1.5
|
|
report = validate_schedule(schedule, spec)
|
|
|
|
assert not report.ok
|
|
assert report.checks["soc_matches_power_flows"] is False
|
|
|
|
|
|
def test_catches_a_state_of_charge_above_capacity(schedule, spec):
|
|
"""Energy appearing from nowhere shows up as an out-of-bounds recompute."""
|
|
# Charge hard enough to overfill, and keep the reported SoC consistent so
|
|
# only the bounds check can fail.
|
|
schedule["charge_market_1_mw"] = 2.0
|
|
schedule["discharge_market_1_mw"] = 0.0
|
|
schedule["charge_market_2_mw"] = 0.0
|
|
schedule["discharge_market_2_mw"] = 0.0
|
|
delta = 0.5 * spec.charge_efficiency * schedule["charge_market_1_mw"]
|
|
schedule["soc_mwh"] = delta.cumsum()
|
|
for name in ("market_1", "market_2"):
|
|
schedule[f"revenue_{name}_gbp"] = (
|
|
0.5
|
|
* schedule[f"{name}_price"]
|
|
* (schedule[f"discharge_{name}_mw"] - schedule[f"charge_{name}_mw"])
|
|
)
|
|
|
|
report = validate_schedule(schedule, spec)
|
|
assert not report.ok
|
|
assert report.checks["soc_within_bounds"] is False
|
|
assert report.checks["soc_matches_power_flows"] is True
|
|
|
|
|
|
def test_catches_revenue_that_does_not_match_prices_and_powers(schedule, spec):
|
|
schedule.loc[schedule.index[5], "revenue_market_1_gbp"] += 25.0
|
|
report = validate_schedule(schedule, spec)
|
|
|
|
assert not report.ok
|
|
assert report.checks["revenue_matches_prices_and_powers"] is False
|
|
assert report.details["max_revenue_error_gbp"] == pytest.approx(25.0)
|
|
|
|
|
|
def test_catches_a_sign_error_in_the_revenue_convention(schedule, spec):
|
|
"""Paying for exports instead of being paid is the classic sign slip."""
|
|
schedule["revenue_market_1_gbp"] *= -1
|
|
report = validate_schedule(schedule, spec)
|
|
|
|
assert not report.ok
|
|
assert report.checks["revenue_matches_prices_and_powers"] is False
|
|
|
|
|
|
def test_reports_are_falsy_only_when_something_failed(schedule, spec):
|
|
good = validate_schedule(schedule, spec)
|
|
assert good.ok and not good.failures
|
|
|
|
schedule.loc[schedule.index[0], "charge_market_1_mw"] = np.float64(9.0)
|
|
bad = validate_schedule(schedule, spec)
|
|
assert not bad.ok and bad.failures
|
|
assert "FAILED" in bad.summary()
|