mirror of
https://github.com/Akkudoktor-EOS/EOS.git
synced 2026-11-13 09:13:30 +01:00
a06473b6d6
LFP batteries can only recalibrate their state of charge close to full charge, so many systems expect a full charge every few weeks and some start one on their own at an arbitrary, possibly expensive time. EOS can now plan it instead (GENETIC only, off by default). New battery settings: full_charge_interval_days (unset = off), full_charge_soc_percentage, full_charge_detect_percentage, full_charge_allow_grid and the informational cell_chemistry. The last full charge is read from the measured SoC history; nothing is stored. When the next one is due within the horizon (or overdue), the maximum SoC is raised to the target. With full_charge_allow_grid the optimizer must reach the target once before the deadline, which is at least 24 h away (penalty battery_full_charge_miss on the best SoC reached), so it picks the cheapest time itself. Without it, PV surplus may fill the battery but the target is not enforced. Signed-off-by: SchaafAlexander <86607129+SchaafAlexander@users.noreply.github.com> Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
433 lines
18 KiB
Python
433 lines
18 KiB
Python
"""Periodic full charge of a stationary battery (e.g. LFP recalibration)."""
|
|
|
|
from datetime import timedelta
|
|
from types import SimpleNamespace
|
|
from typing import Any, Optional
|
|
from unittest.mock import AsyncMock, MagicMock, Mock
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from akkudoktoreos.config.config import ConfigEOS
|
|
from akkudoktoreos.core.cache import CacheEnergyManagementStore
|
|
from akkudoktoreos.core.coreabc import get_ems
|
|
from akkudoktoreos.devices.genetic.battery import Battery, SolarPanelBatteryParameters
|
|
from akkudoktoreos.devices.settings.batterysettings import BatteriesCommonSettings
|
|
from akkudoktoreos.optimization.genetic import configrequest
|
|
from akkudoktoreos.optimization.genetic.configrequest import ConfigOptimizationRequest
|
|
from akkudoktoreos.optimization.genetic.fullcharge import (
|
|
FULL_CHARGE_MIN_LEAD_HOURS,
|
|
enforced_full_charge_deadline,
|
|
full_charge_deadline,
|
|
)
|
|
from akkudoktoreos.optimization.genetic.genetic import GeneticOptimization
|
|
from akkudoktoreos.optimization.genetic.geneticparams import (
|
|
GeneticOptimizationParameters,
|
|
)
|
|
from akkudoktoreos.utils.datetimeutil import DateTime, to_datetime
|
|
|
|
ems_eos = get_ems(init=True) # init once
|
|
|
|
NOW = to_datetime("2026-11-20T12:00:00+01:00")
|
|
|
|
|
|
def _daily(days: int, value: float, now: DateTime = NOW) -> list[tuple[DateTime, Any]]:
|
|
"""One SoC sample per day for the last ``days`` days, oldest first."""
|
|
return [(now.subtract(days=day), value) for day in range(days, 0, -1)]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Deadline from the measured history
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_deadline_counts_from_last_full_charge():
|
|
samples = _daily(21, 0.4) + [(NOW.subtract(days=3), 0.995)]
|
|
assert full_charge_deadline(samples, NOW, 21, 99) == NOW.subtract(days=3).add(days=21)
|
|
|
|
|
|
def test_deadline_uses_the_latest_of_several_full_charges():
|
|
samples = [(NOW.subtract(days=10), 1.0), (NOW.subtract(days=4), 1.0)] + _daily(21, 0.4)
|
|
assert full_charge_deadline(samples, NOW, 21, 99) == NOW.subtract(days=4).add(days=21)
|
|
|
|
|
|
def test_detect_threshold_decides_what_counts_as_full():
|
|
samples = _daily(21, 0.4) + [(NOW.subtract(days=2), 0.98)]
|
|
# 98 % is not full with the default threshold: the whole window is history
|
|
# without a full charge, so the interval counts from the oldest sample.
|
|
assert full_charge_deadline(samples, NOW, 21, 99) == NOW
|
|
assert full_charge_deadline(samples, NOW, 21, 98) == NOW.subtract(days=2).add(days=21)
|
|
|
|
|
|
def test_overdue_deadline_stays_in_the_past():
|
|
"""The real deadline decides whether a full charge is due."""
|
|
samples = [(NOW.subtract(days=30), 1.0)] + _daily(29, 0.4)
|
|
# The interval is elapsed time: 21 x 24 h, across the DST change in October.
|
|
expected = NOW.subtract(days=30) + timedelta(days=21)
|
|
assert full_charge_deadline(samples, NOW, 21, 99) == expected
|
|
assert expected.timestamp() < NOW.timestamp()
|
|
|
|
|
|
def test_enforced_deadline_leaves_lead_time():
|
|
lead = NOW.add(hours=FULL_CHARGE_MIN_LEAD_HOURS)
|
|
assert enforced_full_charge_deadline(NOW.subtract(days=9), NOW) == lead
|
|
assert enforced_full_charge_deadline(NOW.add(hours=5), NOW) == lead
|
|
assert enforced_full_charge_deadline(NOW.add(days=3), NOW) == NOW.add(days=3)
|
|
|
|
|
|
def test_short_history_without_full_charge_counts_from_oldest_sample():
|
|
"""A fresh installation is not forced into a full charge on day one."""
|
|
samples = _daily(3, 0.4)
|
|
assert full_charge_deadline(samples, NOW, 21, 99) == NOW.subtract(days=3).add(days=21)
|
|
|
|
|
|
@pytest.mark.parametrize("value", [None, np.nan, np.inf, "1.0"])
|
|
def test_invalid_and_future_samples_are_ignored(value):
|
|
samples = [(NOW.subtract(days=2), value), (NOW.add(minutes=5), 1.0)] + _daily(5, 0.4)
|
|
assert full_charge_deadline(samples, NOW, 21, 99) == NOW.subtract(days=5).add(days=21)
|
|
|
|
|
|
def test_no_history_gives_no_deadline():
|
|
assert full_charge_deadline([], NOW, 21, 99) is None
|
|
assert full_charge_deadline([(NOW.subtract(days=1), None)], NOW, 21, 99) is None
|
|
|
|
|
|
def test_fractional_interval():
|
|
samples = [(NOW.subtract(days=1), 1.0)]
|
|
assert full_charge_deadline(samples, NOW, 2.5, 99) == NOW.subtract(days=1).add(hours=60)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Settings and parameters
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_settings_defaults_disable_the_feature():
|
|
settings = BatteriesCommonSettings(device_id="battery1")
|
|
assert settings.full_charge_interval_days is None
|
|
assert settings.cell_chemistry is None
|
|
assert settings.full_charge_soc_percentage == 100
|
|
assert settings.full_charge_detect_percentage == 99
|
|
assert settings.full_charge_allow_grid is False
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"extra",
|
|
[
|
|
{"full_charge_interval_days": 0},
|
|
{"full_charge_interval_days": -1},
|
|
{"full_charge_soc_percentage": 0},
|
|
{"full_charge_soc_percentage": 101},
|
|
{"full_charge_soc_percentage": 95, "full_charge_detect_percentage": 99},
|
|
{"cell_chemistry": "lithium"},
|
|
],
|
|
)
|
|
def test_invalid_settings_are_rejected(extra):
|
|
with pytest.raises(ValidationError):
|
|
BatteriesCommonSettings(device_id="battery1", **extra)
|
|
|
|
|
|
def test_parameters_reject_target_above_max_soc():
|
|
with pytest.raises(ValidationError):
|
|
SolarPanelBatteryParameters(
|
|
device_id="battery1",
|
|
capacity_wh=10000,
|
|
max_soc_percentage=90,
|
|
full_charge_soc_percentage=100,
|
|
)
|
|
|
|
|
|
def test_parameters_parse_naive_deadline_as_local_time():
|
|
parameters = SolarPanelBatteryParameters.model_validate(
|
|
{
|
|
"device_id": "battery1",
|
|
"capacity_wh": 10000,
|
|
"full_charge_soc_percentage": 100,
|
|
"full_charge_deadline_datetime": "2026-11-21T07:00:00",
|
|
}
|
|
)
|
|
assert parameters.full_charge_deadline_datetime == to_datetime("2026-11-21T07:00:00")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Resolving the configuration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def full_charge_request(config_eos, monkeypatch):
|
|
"""A configured battery with a 30 h control horizon from 10:15 UTC."""
|
|
config_eos.merge_settings_from_dict(
|
|
{
|
|
"prediction": {"hours": 48},
|
|
"optimization": {
|
|
"algorithm": "GENETIC",
|
|
"genetic": {
|
|
"interval_sec": 900,
|
|
"horizon_hours": 30,
|
|
"tail_horizon_hours": 0,
|
|
"terminal_value_mode": "FIXED",
|
|
"terminal_value_euro_per_kwh": 0.0,
|
|
"individuals": 10,
|
|
"generations": 10,
|
|
"seed": 42,
|
|
},
|
|
},
|
|
"devices": {
|
|
"max_batteries": 1,
|
|
"max_electric_vehicles": 0,
|
|
"max_inverters": 1,
|
|
"max_home_appliances": 0,
|
|
"batteries": {
|
|
"storage": {
|
|
"capacity_wh": 10000,
|
|
"max_charge_power_w": 5000,
|
|
"max_soc_percentage": 90,
|
|
"full_charge_interval_days": 21,
|
|
}
|
|
},
|
|
"electric_vehicles": {},
|
|
"inverters": {"inverter": {"battery_id": "storage", "max_power_w": 5000}},
|
|
"home_appliances": {},
|
|
},
|
|
}
|
|
)
|
|
start = to_datetime("2026-09-12T10:15:00Z", in_timezone="UTC")
|
|
ems = SimpleNamespace(
|
|
start_datetime=start, observation_datetime=start, genetic_solution=lambda: None
|
|
)
|
|
monkeypatch.setattr(configrequest, "get_ems", lambda: ems)
|
|
measurement = Mock(key_to_lists=AsyncMock(return_value=([], [])))
|
|
monkeypatch.setattr(ConfigOptimizationRequest, "measurement", measurement)
|
|
data = {
|
|
"soc": {"storage": 40},
|
|
"forecasts": {
|
|
"pv_forecast_wh": [0.0] * 192,
|
|
"total_load": [200.0] * 192,
|
|
"electricity_price_per_wh": [0.0003] * 192,
|
|
"feed_in_tariff_per_wh": [0.00008] * 192,
|
|
},
|
|
}
|
|
return config_eos, start, measurement, data
|
|
|
|
|
|
def _history(start: DateTime, full_days_ago: Optional[float]) -> tuple[list[DateTime], list[float]]:
|
|
"""Hourly SoC history over 30 days, full once ``full_days_ago`` (if given)."""
|
|
dates = [start.subtract(hours=hour) for hour in range(30 * 24, 0, -1)]
|
|
values = [0.4] * len(dates)
|
|
if full_days_ago is not None:
|
|
index = len(dates) - int(full_days_ago * 24)
|
|
values[index] = 1.0
|
|
return dates, values
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_feature_disabled_reads_no_history(full_charge_request):
|
|
config, _, measurement, data = full_charge_request
|
|
config.devices.batteries["storage"].full_charge_interval_days = None
|
|
parameters = await ConfigOptimizationRequest.model_validate(data).resolve()
|
|
assert parameters.pv_battery is not None
|
|
assert parameters.pv_battery.max_soc_percentage == 90
|
|
assert parameters.pv_battery.full_charge_soc_percentage is None
|
|
measurement.key_to_lists.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_not_due_within_horizon_leaves_battery_unchanged(full_charge_request):
|
|
_, start, measurement, data = full_charge_request
|
|
measurement.key_to_lists.return_value = _history(start, full_days_ago=5)
|
|
parameters = await ConfigOptimizationRequest.model_validate(data).resolve()
|
|
assert parameters.pv_battery is not None
|
|
assert parameters.pv_battery.max_soc_percentage == 90
|
|
assert parameters.pv_battery.full_charge_soc_percentage is None
|
|
assert parameters.pv_battery.full_charge_deadline_datetime is None
|
|
# The history window covers the whole interval.
|
|
kwargs = measurement.key_to_lists.call_args.kwargs
|
|
assert kwargs["key"] == "storage-soc-factor"
|
|
assert kwargs["start_datetime"] == start.subtract(days=21)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_due_without_grid_only_raises_the_maximum(full_charge_request):
|
|
_, start, measurement, data = full_charge_request
|
|
measurement.key_to_lists.return_value = _history(start, full_days_ago=20.5)
|
|
parameters = await ConfigOptimizationRequest.model_validate(data).resolve()
|
|
assert parameters.pv_battery is not None
|
|
assert parameters.pv_battery.max_soc_percentage == 100
|
|
assert parameters.pv_battery.full_charge_soc_percentage is None
|
|
assert parameters.pv_battery.full_charge_deadline_datetime is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_due_with_grid_enforces_target_and_deadline(full_charge_request):
|
|
config, start, measurement, data = full_charge_request
|
|
config.devices.batteries["storage"].full_charge_allow_grid = True
|
|
measurement.key_to_lists.return_value = _history(start, full_days_ago=None)
|
|
parameters = await ConfigOptimizationRequest.model_validate(data).resolve()
|
|
assert parameters.pv_battery is not None
|
|
assert parameters.pv_battery.max_soc_percentage == 100
|
|
assert parameters.pv_battery.full_charge_soc_percentage == 100
|
|
# Overdue: the optimizer gets the minimum lead time.
|
|
assert parameters.pv_battery.full_charge_deadline_datetime == start.add(
|
|
hours=FULL_CHARGE_MIN_LEAD_HOURS
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_overdue_full_charge_is_due_with_a_horizon_shorter_than_the_lead_time(
|
|
full_charge_request,
|
|
):
|
|
"""Regression: the lead time must not push an overdue full charge out of the horizon."""
|
|
config, start, measurement, data = full_charge_request
|
|
config.merge_settings_from_dict({"optimization": {"genetic": {"horizon_hours": 12}}})
|
|
config.devices.batteries["storage"].full_charge_allow_grid = True
|
|
measurement.key_to_lists.return_value = _history(start, full_days_ago=None)
|
|
parameters = await ConfigOptimizationRequest.model_validate(data).resolve()
|
|
assert parameters.pv_battery is not None
|
|
assert parameters.pv_battery.max_soc_percentage == 100
|
|
assert parameters.pv_battery.full_charge_soc_percentage == 100
|
|
assert parameters.pv_battery.full_charge_deadline_datetime == start.add(
|
|
hours=FULL_CHARGE_MIN_LEAD_HOURS
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_history_does_not_schedule_a_full_charge(full_charge_request):
|
|
config, _, measurement, data = full_charge_request
|
|
config.devices.batteries["storage"].full_charge_allow_grid = True
|
|
measurement.key_to_lists.side_effect = KeyError("storage-soc-factor")
|
|
parameters = await ConfigOptimizationRequest.model_validate(data).resolve()
|
|
assert parameters.pv_battery is not None
|
|
assert parameters.pv_battery.max_soc_percentage == 90
|
|
assert parameters.pv_battery.full_charge_soc_percentage is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Optimizer
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _optimizer_config(config_eos: ConfigEOS, hours: int) -> None:
|
|
config_eos.merge_settings_from_dict(
|
|
{
|
|
"prediction": {"hours": hours},
|
|
"optimization": {
|
|
"genetic": {
|
|
"individuals": 100,
|
|
"generations": 40,
|
|
"tail_horizon_hours": 0,
|
|
"horizon_hours": hours,
|
|
"interval_sec": 3600,
|
|
"terminal_value_mode": "FIXED",
|
|
"terminal_value_euro_per_kwh": 0.0,
|
|
}
|
|
},
|
|
}
|
|
)
|
|
|
|
|
|
def _full_charge_parameters(hours: int, **battery_extra) -> GeneticOptimizationParameters:
|
|
"""No PV and expensive energy: without a target nothing would be charged."""
|
|
return GeneticOptimizationParameters.model_validate(
|
|
dict(
|
|
ems={
|
|
"pv_prognose_wh": [0.0] * hours,
|
|
"strompreis_euro_pro_wh": [0.0004] * hours,
|
|
"einspeiseverguetung_euro_pro_wh": [0.00007] * hours,
|
|
"preis_euro_pro_wh_akku": 0.0,
|
|
"gesamtlast": [200.0] * hours,
|
|
},
|
|
pv_battery={
|
|
"device_id": "battery1",
|
|
"capacity_wh": 10000,
|
|
"initial_soc_percentage": 30,
|
|
"min_soc_percentage": 0,
|
|
"max_soc_percentage": 100,
|
|
"charging_efficiency": 1.0,
|
|
"discharging_efficiency": 1.0,
|
|
"max_charge_power_w": 5000,
|
|
**battery_extra,
|
|
},
|
|
inverter={
|
|
"device_id": "inverter1",
|
|
"max_power_wh": 10000,
|
|
"battery_id": "battery1",
|
|
"ac_to_dc_efficiency": 1.0,
|
|
"dc_to_ac_efficiency": 1.0,
|
|
"max_ac_charge_power_w": 5000,
|
|
},
|
|
ev=None,
|
|
)
|
|
)
|
|
|
|
|
|
def test_full_charge_slot_resolution(config_eos: ConfigEOS):
|
|
_optimizer_config(config_eos, 24)
|
|
ems_eos.set_start_datetime(to_datetime().set(hour=0, minute=0))
|
|
optimization = GeneticOptimization(fixed_seed=1)
|
|
optimization._slot0_datetime = optimization.ems.start_datetime
|
|
slot0 = optimization._slot0_datetime
|
|
|
|
assert optimization._full_charge_slot(_full_charge_parameters(24)) is None
|
|
parameters = _full_charge_parameters(24, full_charge_soc_percentage=100)
|
|
assert optimization._full_charge_slot(parameters) == optimization.control_end_slot
|
|
parameters = _full_charge_parameters(
|
|
24, full_charge_soc_percentage=100, full_charge_deadline_datetime=slot0.add(hours=10)
|
|
)
|
|
assert optimization._full_charge_slot(parameters) == 10
|
|
parameters = _full_charge_parameters(
|
|
24, full_charge_soc_percentage=100, full_charge_deadline_datetime=slot0.add(hours=100)
|
|
)
|
|
assert optimization._full_charge_slot(parameters) == optimization.control_end_slot
|
|
|
|
|
|
def test_maximum_soc_before_deadline_counts(config_eos: ConfigEOS):
|
|
_optimizer_config(config_eos, 24)
|
|
ems_eos.set_start_datetime(to_datetime().set(hour=0, minute=0))
|
|
optimization = GeneticOptimization(fixed_seed=1)
|
|
optimization.simulation.battery = MagicMock(spec=Battery)
|
|
optimization.simulation.battery.current_soc_percentage.return_value = 40.0
|
|
result = {"akku_soc_pro_stunde": [30.0, 100.0, 60.0, 50.0]}
|
|
|
|
optimization._full_charge_deadline_slot = None
|
|
assert optimization._battery_max_soc_until_full_charge_deadline(result, 10) is None
|
|
|
|
# Slot 11 starts after the charge in slot 10 finished.
|
|
optimization._full_charge_deadline_slot = 11
|
|
assert optimization._battery_max_soc_until_full_charge_deadline(result, 10) == 100.0
|
|
optimization._full_charge_deadline_slot = 10
|
|
assert optimization._battery_max_soc_until_full_charge_deadline(result, 10) == 30.0
|
|
|
|
# Deadline at the end also counts the SoC after the last slot.
|
|
result = {"akku_soc_pro_stunde": [30.0, 20.0]}
|
|
optimization.simulation.battery.current_soc_percentage.return_value = 95.0
|
|
optimization._full_charge_deadline_slot = 99
|
|
assert optimization._battery_max_soc_until_full_charge_deadline(result, 10) == 95.0
|
|
|
|
|
|
def _run(config_eos: ConfigEOS, **battery_extra) -> list[float]:
|
|
hours = 24
|
|
_optimizer_config(config_eos, hours)
|
|
ems_eos.set_start_datetime(to_datetime().set(hour=0, minute=0))
|
|
CacheEnergyManagementStore().clear()
|
|
solution = GeneticOptimization(fixed_seed=42).optimize_ems(
|
|
parameters=_full_charge_parameters(hours, **battery_extra), start_hour=0, ngen=40
|
|
)
|
|
return list(solution.result.battery_soc_per_hour)
|
|
|
|
|
|
def test_due_full_charge_is_reached_before_deadline(config_eos: ConfigEOS):
|
|
without = _run(config_eos)
|
|
assert max(without) < 99.0
|
|
|
|
slot0 = to_datetime().set(hour=0, minute=0, second=0, microsecond=0)
|
|
soc = _run(
|
|
config_eos,
|
|
full_charge_soc_percentage=100,
|
|
full_charge_deadline_datetime=slot0.add(hours=12),
|
|
)
|
|
# Slot 12 starts at the deadline; its start-of-slot SoC includes every
|
|
# charge that finished in time.
|
|
assert max(soc[:13]) >= 99.0
|