mirror of
https://github.com/Akkudoktor-EOS/EOS.git
synced 2026-10-09 16:06:40 +00:00
Andreas
38eb4ad450
Defer inactive EV deadline fields to the optimizer port, reject nonfinite export rates, and document the hourly Optimize boundary. Verify converter IDs, rates and LCOS, separate GENETIC0 interpolation, physical boundary flows and independent GENETIC repricing.
188 lines
7.0 KiB
Python
188 lines
7.0 KiB
Python
import json
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import patch
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
from pypdf import PdfReader
|
|
|
|
from akkudoktoreos.config.config import ConfigEOS
|
|
from akkudoktoreos.core.cache import CacheEnergyManagementStore
|
|
from akkudoktoreos.core.coreabc import get_ems
|
|
from akkudoktoreos.optimization.genetic.genetic import GeneticOptimization
|
|
from akkudoktoreos.optimization.genetic.geneticparams import (
|
|
GeneticOptimizationParameters,
|
|
)
|
|
from akkudoktoreos.optimization.genetic.geneticsolution import GeneticSolution
|
|
from akkudoktoreos.optimization.genetic.geneticvisualize import (
|
|
genetic_prepare_visualize,
|
|
)
|
|
from akkudoktoreos.utils.datetimeutil import to_datetime
|
|
|
|
ems_eos = get_ems(init=True) # init once
|
|
|
|
DIR_TESTDATA = Path(__file__).parent / "testdata" / "genetic"
|
|
|
|
|
|
def compare_dict(actual: dict[str, Any], expected: dict[str, Any]):
|
|
assert set(actual) == set(expected)
|
|
|
|
for key, value in expected.items():
|
|
if isinstance(value, dict):
|
|
assert isinstance(actual[key], dict)
|
|
compare_dict(actual[key], value)
|
|
elif isinstance(value, list):
|
|
assert isinstance(actual[key], list)
|
|
assert actual[key] == pytest.approx(value)
|
|
else:
|
|
assert actual[key] == pytest.approx(value)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"fn_in, fn_out, ngen, break_even",
|
|
[
|
|
("optimize_input_1.json", "optimize_result_1.json", 3, 0),
|
|
("optimize_input_2.json", "optimize_result_2.json", 3, 0),
|
|
("optimize_input_2.json", "optimize_result_2_full.json", 400, 0),
|
|
("optimize_input_1.json", "optimize_result_1_be.json", 3, 1),
|
|
("optimize_input_2.json", "optimize_result_2_be.json", 3, 1),
|
|
],
|
|
)
|
|
async def test_optimize(
|
|
fn_in: str,
|
|
fn_out: str,
|
|
ngen: int,
|
|
break_even: int,
|
|
config_eos: ConfigEOS,
|
|
is_finalize: bool,
|
|
):
|
|
"""Test optimize_ems."""
|
|
# Test parameters
|
|
fixed_start_hour = 10
|
|
fixed_seed = 42
|
|
|
|
# Assure configuration holds the correct values
|
|
config_eos.merge_settings_from_dict(
|
|
{
|
|
"prediction": {
|
|
"hours": 48
|
|
},
|
|
"optimization": {
|
|
"algorithm": "GENETIC",
|
|
"genetic": {
|
|
"horizon_hours": 48,
|
|
"individuals": 300,
|
|
"generations": 10,
|
|
"penalties": {
|
|
"ev_soc_miss": 10,
|
|
"ac_charge_break_even": break_even,
|
|
}
|
|
}
|
|
},
|
|
"devices": {
|
|
"max_electric_vehicles": 1,
|
|
"electric_vehicles": { "ev1":
|
|
{
|
|
"charge_rates": [0.0, 0.375, 0.5, 0.625, 0.75, 0.875, 1.0],
|
|
}
|
|
},
|
|
}
|
|
}
|
|
)
|
|
|
|
# Load input and output data
|
|
parameter_file = DIR_TESTDATA / fn_in
|
|
with parameter_file.open("r") as f_in:
|
|
input_data = GeneticOptimizationParameters(**json.load(f_in))
|
|
|
|
# Fake energy management run start datetime
|
|
ems_eos.set_start_datetime(to_datetime().set(hour=fixed_start_hour))
|
|
|
|
# Throw away any cached results of the last energy management run.
|
|
CacheEnergyManagementStore().clear()
|
|
|
|
genetic_optimization = GeneticOptimization(fixed_seed=fixed_seed)
|
|
|
|
# Activate with pytest --finalize
|
|
if ngen > 10 and not is_finalize:
|
|
pytest.skip()
|
|
|
|
# Call the optimization function
|
|
genetic_solution = genetic_optimization.optimize_ems(
|
|
parameters=input_data, start_hour=fixed_start_hour, ngen=ngen
|
|
)
|
|
|
|
# Write test output to file, so we can take it as new data on intended change
|
|
TESTDATA_FILE = DIR_TESTDATA / f"new_{fn_out}"
|
|
with TESTDATA_FILE.open("w", encoding="utf-8", newline="\n") as f_out:
|
|
f_out.write(genetic_solution.model_dump_json(indent=4, exclude_unset=True))
|
|
|
|
solution_file = DIR_TESTDATA / fn_out
|
|
# In case a new test case is added, we don't want to fail here, so the new output is written
|
|
# to disk before
|
|
try:
|
|
with solution_file.open("r") as f_out:
|
|
expected_data = json.load(f_out)
|
|
expected_result = GeneticSolution(**expected_data)
|
|
except ValidationError:
|
|
# Expected genetic solution data does not fit to GeneticSolution data schema
|
|
# Possibly the GeneticSolution class changed.
|
|
pytest.fail(
|
|
f"ValidationError: Can not load expected solution from {solution_file}\n"
|
|
f"cp {TESTDATA_FILE} {solution_file}\n"
|
|
)
|
|
except FileNotFoundError:
|
|
# Should not happen
|
|
pytest.fail(
|
|
f"FileNotFoundError: Can not load expected solution from {solution_file}\n"
|
|
f"cp {TESTDATA_FILE} {solution_file}\n"
|
|
)
|
|
|
|
# Keep the output contract, but do not demand an identical stochastic
|
|
# schedule or monetary golden from the previous direct-consumption model.
|
|
assert set(genetic_solution.model_dump()) == set(expected_result.model_dump())
|
|
result = genetic_solution.result
|
|
expected_slots = len(input_data.ems.pv_forecast_wh) - fixed_start_hour
|
|
assert len(result.grid_consumption_wh_per_hour) == expected_slots
|
|
assert len(result.grid_feed_in_wh_per_hour) == expected_slots
|
|
prices = np.asarray(genetic_solution.parameters.ems.electricity_price_per_wh)[fixed_start_hour:]
|
|
tariffs = genetic_solution.parameters.ems.feed_in_tariff_per_wh
|
|
if isinstance(tariffs, list):
|
|
tariffs = np.asarray(tariffs)[fixed_start_hour:]
|
|
expected_costs = np.asarray(result.grid_consumption_wh_per_hour) * prices
|
|
expected_revenues = np.asarray(result.grid_feed_in_wh_per_hour) * tariffs
|
|
np.testing.assert_allclose(result.costs_per_hour, expected_costs)
|
|
np.testing.assert_allclose(result.revenue_per_hour, expected_revenues)
|
|
assert result.total_costs == pytest.approx(sum(expected_costs))
|
|
assert result.total_revenue == pytest.approx(sum(expected_revenues))
|
|
assert result.total_balance == pytest.approx(sum(expected_costs) - sum(expected_revenues))
|
|
assert result.total_losses == pytest.approx(sum(result.losses_per_hour))
|
|
assert all(value >= 0 for value in result.grid_consumption_wh_per_hour)
|
|
assert all(value >= 0 for value in result.grid_feed_in_wh_per_hour)
|
|
assert all(0 <= value <= 100 for value in result.battery_soc_per_hour)
|
|
assert all(0 <= value <= 100 for value in result.ev_soc_per_hour)
|
|
|
|
# Check the correct generic optimization solution is created
|
|
optimization_solution = await genetic_solution.optimization_solution()
|
|
# @TODO
|
|
|
|
# Check the correct generic energy management plan is created
|
|
plan = genetic_solution.energy_management_plan()
|
|
# @TODO
|
|
|
|
# Check visualization works
|
|
pdf = genetic_prepare_visualize(
|
|
solution=genetic_solution,
|
|
)
|
|
assert pdf.startswith(b"%PDF-")
|
|
|
|
reader = PdfReader(BytesIO(pdf))
|
|
assert len(reader.pages) == 6
|
|
|
|
|
|
# Everything passed, remove generated files
|
|
TESTDATA_FILE.unlink()
|