mirror of
https://github.com/Akkudoktor-EOS/EOS.git
synced 2026-10-09 16:06:40 +00:00
313 lines
13 KiB
Python
313 lines
13 KiB
Python
"""Offline regression tests for per-run optimization results and atomic publication."""
|
|
|
|
from asyncio import Lock
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, Mock
|
|
|
|
import pytest
|
|
|
|
from akkudoktoreos.core import ems as ems_module
|
|
from akkudoktoreos.core.emsettings import EnergyManagementMode
|
|
from akkudoktoreos.optimization.genetic0.genetic0params import (
|
|
Genetic0OptimizationParameters,
|
|
)
|
|
from akkudoktoreos.optimization.optimization import OptimizationAlgorithm
|
|
from akkudoktoreos.utils.datetimeutil import to_datetime
|
|
|
|
|
|
@pytest.fixture
|
|
def offline_ems(monkeypatch):
|
|
"""Run the actual EMS orchestration with no real adapters or prediction IO."""
|
|
cls = ems_module.EnergyManagement
|
|
for name in (
|
|
"_start_datetime",
|
|
"_last_run_datetime",
|
|
"_plan",
|
|
"_optimization_solution",
|
|
"_genetic_solution",
|
|
"_genetic0_solution",
|
|
):
|
|
monkeypatch.setattr(cls, name, None)
|
|
monkeypatch.setattr(cls, "_stage", ems_module.EnergyManagementStage.IDLE)
|
|
monkeypatch.setattr(cls, "_run_lock", Lock())
|
|
monkeypatch.setattr(ems_module, "CacheEnergyManagementStore", Mock())
|
|
return SimpleNamespace(
|
|
config=SimpleNamespace(
|
|
ems=SimpleNamespace(mode=EnergyManagementMode.OPTIMIZATION),
|
|
optimization=SimpleNamespace(
|
|
algorithm=OptimizationAlgorithm.GENETIC,
|
|
genetic=SimpleNamespace(generations=3, seed=17),
|
|
genetic0=SimpleNamespace(generations=5, seed=29),
|
|
),
|
|
server=SimpleNamespace(verbose=False),
|
|
),
|
|
prediction=SimpleNamespace(update_data=AsyncMock()),
|
|
adapter=SimpleNamespace(update_data=AsyncMock()),
|
|
set_start_datetime=cls.set_start_datetime,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("algorithm", list(OptimizationAlgorithm))
|
|
@pytest.mark.parametrize("selection", ["configured", "explicit"])
|
|
@pytest.mark.parametrize("supplied", [False, True])
|
|
async def test_optimization_routes_only_selected_algorithm(
|
|
monkeypatch, offline_ems, algorithm, selection, supplied
|
|
):
|
|
"""Configuration selection and explicit overrides retain isolated async paths."""
|
|
selected_name = "Genetic" if algorithm == OptimizationAlgorithm.GENETIC else "Genetic0"
|
|
suffix = "genetic" if algorithm == OptimizationAlgorithm.GENETIC else "genetic0"
|
|
other_suffix = "genetic0" if suffix == "genetic" else "genetic"
|
|
sentinel_parameters = object()
|
|
sentinel_result, sentinel_plan = object(), object()
|
|
solution = SimpleNamespace(
|
|
optimization_solution=AsyncMock(return_value=sentinel_result),
|
|
energy_management_plan=Mock(return_value=sentinel_plan),
|
|
)
|
|
constructors = {}
|
|
preparers = {}
|
|
for prefix in ("Genetic", "Genetic0"):
|
|
constructor = Mock()
|
|
constructor.return_value.optimize_ems.return_value = solution
|
|
constructors[prefix] = constructor
|
|
monkeypatch.setattr(ems_module, prefix + "Optimization", constructor)
|
|
prepare = AsyncMock(return_value=sentinel_parameters)
|
|
preparers[prefix] = prepare
|
|
monkeypatch.setattr(
|
|
getattr(ems_module, prefix + "OptimizationParameters"), "prepare", prepare
|
|
)
|
|
kwargs = {"start_datetime": to_datetime("2026-09-16T10:00:00+02:00")}
|
|
if selection == "configured":
|
|
offline_ems.config.optimization.algorithm = algorithm
|
|
else:
|
|
offline_ems.config.optimization.algorithm = OptimizationAlgorithm(other_suffix.upper())
|
|
kwargs["algorithm"] = algorithm
|
|
if supplied:
|
|
kwargs[suffix + "_parameters"] = sentinel_parameters
|
|
kwargs[suffix + "_generations"] = 7
|
|
kwargs[suffix + "_seed"] = 43
|
|
run_result = await ems_module.EnergyManagement.run(offline_ems, **kwargs)
|
|
assert run_result is solution
|
|
selected = constructors[selected_name]
|
|
expected_config = getattr(offline_ems.config.optimization, suffix)
|
|
selected.assert_called_once_with(
|
|
verbose=False, fixed_seed=43 if supplied else expected_config.seed
|
|
)
|
|
selected.return_value.optimize_ems.assert_called_once_with(
|
|
start_hour=10,
|
|
parameters=sentinel_parameters,
|
|
ngen=7 if supplied else expected_config.generations,
|
|
)
|
|
other_name = "Genetic0" if selected_name == "Genetic" else "Genetic"
|
|
constructors[other_name].assert_not_called()
|
|
preparers[other_name].assert_not_awaited()
|
|
if supplied:
|
|
preparers[selected_name].assert_not_awaited()
|
|
else:
|
|
preparers[selected_name].assert_awaited_once_with()
|
|
solution.optimization_solution.assert_awaited_once_with()
|
|
solution.energy_management_plan.assert_called_once_with()
|
|
cls = ems_module.EnergyManagement
|
|
assert getattr(cls, "_" + suffix + "_solution") is solution
|
|
assert getattr(cls, "_" + other_suffix + "_solution") is None
|
|
assert cls.optimization_solution() is sentinel_result
|
|
assert cls.plan() is sentinel_plan
|
|
assert cls.stage() == ems_module.EnergyManagementStage.IDLE
|
|
offline_ems.prediction.update_data.assert_awaited_once_with(
|
|
force_enable=False, force_update=False
|
|
)
|
|
assert offline_ems.adapter.update_data.await_count == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("mode", [EnergyManagementMode.DISABLED, EnergyManagementMode.PREDICTION])
|
|
async def test_non_optimization_modes_never_optimize(monkeypatch, offline_ems, mode):
|
|
constructors = [Mock(), Mock()]
|
|
monkeypatch.setattr(ems_module, "GeneticOptimization", constructors[0])
|
|
monkeypatch.setattr(ems_module, "Genetic0Optimization", constructors[1])
|
|
offline_ems.config.ems.mode = mode
|
|
await ems_module.EnergyManagement.run(offline_ems)
|
|
for constructor in constructors:
|
|
constructor.assert_not_called()
|
|
assert offline_ems.prediction.update_data.await_count == (
|
|
mode == EnergyManagementMode.PREDICTION
|
|
)
|
|
assert offline_ems.adapter.update_data.await_count == (mode == EnergyManagementMode.PREDICTION)
|
|
assert ems_module.EnergyManagement.plan() is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("prefix", ["Genetic", "Genetic0"])
|
|
async def test_missing_preparation_does_not_dispatch_controls(monkeypatch, offline_ems, prefix):
|
|
constructor = Mock()
|
|
monkeypatch.setattr(ems_module, prefix + "Optimization", constructor)
|
|
prepare = AsyncMock(return_value=None)
|
|
monkeypatch.setattr(getattr(ems_module, prefix + "OptimizationParameters"), "prepare", prepare)
|
|
await ems_module.EnergyManagement.run(
|
|
offline_ems, algorithm=OptimizationAlgorithm(prefix.upper())
|
|
)
|
|
prepare.assert_awaited_once_with()
|
|
constructor.assert_not_called()
|
|
assert offline_ems.adapter.update_data.await_count == 1 # acquisition only
|
|
assert ems_module.EnergyManagement.plan() is None
|
|
assert ems_module.EnergyManagement.stage() == ems_module.EnergyManagementStage.IDLE
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("start_hour", [None, 11])
|
|
async def test_legacy_optimize_explicitly_uses_genetic0(monkeypatch, start_hour):
|
|
"""Even with GENETIC configured, /optimize must never become the new optimizer."""
|
|
from akkudoktoreos.server import eos
|
|
from akkudoktoreos.server.rest.error import EOSProblem
|
|
|
|
fake = SimpleNamespace(
|
|
run=AsyncMock(return_value=None), genetic0_solution=Mock(return_value=None)
|
|
)
|
|
monkeypatch.setattr(eos, "get_ems", lambda: fake)
|
|
parameters = Genetic0OptimizationParameters.model_validate(
|
|
{
|
|
"ems": {
|
|
"pv_prognose_wh": [0.0, 100.0],
|
|
"gesamtlast": [100.0, 100.0],
|
|
"strompreis_euro_pro_wh": [0.0003, 0.0003],
|
|
"einspeiseverguetung_euro_pro_wh": 0.00008,
|
|
"preis_euro_pro_wh_akku": 0.0,
|
|
},
|
|
"pv_akku": None,
|
|
"eauto": None,
|
|
"inverter": None,
|
|
}
|
|
)
|
|
with pytest.raises(EOSProblem):
|
|
await eos.fastapi_optimize(parameters=parameters, start_hour=start_hour, ngen=2)
|
|
kwargs = fake.run.await_args.kwargs
|
|
assert kwargs["mode"] == EnergyManagementMode.OPTIMIZATION
|
|
assert kwargs["algorithm"] == OptimizationAlgorithm.GENETIC0
|
|
assert kwargs["genetic0_parameters"] is parameters
|
|
assert kwargs["genetic0_generations"] == 2
|
|
assert "genetic_parameters" not in kwargs
|
|
assert (
|
|
kwargs["start_datetime"] is None
|
|
if start_hour is None
|
|
else kwargs["start_datetime"].hour == 11
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("phase", ["optimizer", "conversion", "plan"])
|
|
async def test_failed_legacy_http_run_never_reports_previous_solution(
|
|
monkeypatch, offline_ems, phase
|
|
):
|
|
import json
|
|
from pathlib import Path
|
|
from types import MethodType
|
|
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from akkudoktoreos.optimization.genetic0.genetic0solution import Genetic0Solution
|
|
from akkudoktoreos.server import eos
|
|
|
|
cls = ems_module.EnergyManagement
|
|
data = json.loads(
|
|
(Path(__file__).parent / "testdata/genetic0/optimize_result_1.json").read_text()
|
|
)
|
|
previous = Genetic0Solution.model_validate(data)
|
|
monkeypatch.setattr(cls, "_genetic0_solution", previous)
|
|
constructor = Mock()
|
|
error = RuntimeError("synthetic " + phase + " failure")
|
|
native = SimpleNamespace(
|
|
optimization_solution=AsyncMock(return_value=object()),
|
|
energy_management_plan=Mock(return_value=object()),
|
|
)
|
|
constructor.return_value.optimize_ems.return_value = native
|
|
if phase == "optimizer":
|
|
constructor.return_value.optimize_ems.side_effect = error
|
|
elif phase == "conversion":
|
|
native.optimization_solution.side_effect = error
|
|
else:
|
|
native.energy_management_plan.side_effect = error
|
|
monkeypatch.setattr(ems_module, "Genetic0Optimization", constructor)
|
|
offline_ems.run = MethodType(cls.run, offline_ems)
|
|
offline_ems.genetic0_solution = cls.genetic0_solution
|
|
monkeypatch.setattr(eos, "get_ems", lambda: offline_ems)
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=eos.app, raise_app_exceptions=False), base_url="http://test"
|
|
) as client:
|
|
response = await client.post("/optimize?ngen=1", json=data["parameters"])
|
|
constructor.return_value.optimize_ems.assert_called_once()
|
|
# This diagnostic proves a failure is the stale-success bug, not invalid input.
|
|
if response.status_code == 200:
|
|
assert response.json()["start_solution"] == previous.start_solution
|
|
assert response.json()["result"]["total_balance"] == previous.result.total_balance
|
|
assert response.status_code >= 400, (
|
|
"The failing optimizer returned HTTP 200 with the previous solution"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("prefix", ["Genetic", "Genetic0"])
|
|
@pytest.mark.parametrize("phase", ["conversion", "plan"])
|
|
async def test_conversion_failure_preserves_consistent_previous_results(
|
|
monkeypatch, offline_ems, prefix, phase
|
|
):
|
|
cls = ems_module.EnergyManagement
|
|
previous_specific, previous_generic, previous_plan = object(), object(), object()
|
|
suffix = prefix.lower()
|
|
monkeypatch.setattr(cls, "_" + suffix + "_solution", previous_specific)
|
|
monkeypatch.setattr(cls, "_optimization_solution", previous_generic)
|
|
monkeypatch.setattr(cls, "_plan", previous_plan)
|
|
conversion = AsyncMock(return_value=object())
|
|
solution = SimpleNamespace(optimization_solution=conversion, energy_management_plan=Mock())
|
|
if phase == "conversion":
|
|
conversion.side_effect = RuntimeError("synthetic conversion failure")
|
|
else:
|
|
solution.energy_management_plan.side_effect = RuntimeError("synthetic plan failure")
|
|
constructor = Mock()
|
|
constructor.return_value.optimize_ems.return_value = solution
|
|
monkeypatch.setattr(ems_module, prefix + "Optimization", constructor)
|
|
result = await cls.run(
|
|
offline_ems,
|
|
algorithm=OptimizationAlgorithm(prefix.upper()),
|
|
**{suffix + "_parameters": object()},
|
|
)
|
|
assert result is None
|
|
conversion.assert_awaited_once_with()
|
|
assert solution.energy_management_plan.call_count == (phase == "plan")
|
|
assert offline_ems.adapter.update_data.await_count == 1
|
|
assert (
|
|
getattr(cls, "_" + suffix + "_solution"),
|
|
cls.optimization_solution(),
|
|
cls.plan(),
|
|
cls.stage(),
|
|
) == (
|
|
previous_specific,
|
|
previous_generic,
|
|
previous_plan,
|
|
ems_module.EnergyManagementStage.IDLE,
|
|
), (
|
|
"Failed conversion published a new algorithm result beside the old plan and left EMS in OPTIMIZATION"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_legacy_endpoint_uses_run_return_value_not_last_cached_solution(monkeypatch):
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from akkudoktoreos.optimization.genetic0.genetic0solution import Genetic0Solution
|
|
from akkudoktoreos.server import eos
|
|
|
|
data = json.loads(
|
|
(Path(__file__).parent / "testdata/genetic0/optimize_result_1.json").read_text()
|
|
)
|
|
produced = Genetic0Solution.model_validate(data)
|
|
fake = SimpleNamespace(
|
|
run=AsyncMock(return_value=produced),
|
|
genetic0_solution=Mock(side_effect=AssertionError("must use this run's result")),
|
|
)
|
|
monkeypatch.setattr(eos, "get_ems", lambda: fake)
|
|
result = await eos.fastapi_optimize(parameters=produced.parameters, ngen=1)
|
|
assert result.start_solution == produced.start_solution
|
|
assert result.result.total_balance == produced.result.total_balance
|
|
fake.genetic0_solution.assert_not_called()
|