Mirrors/EOS
1
0
Fork 0
mirror of https://github.com/Akkudoktor-EOS/EOS.git synced 2026-10-29 09:11:57 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

test: consolidate suites and remove redundant cases (#1459)

Browse Source 
Regression tests were scattered across feature- and bug-specific files, and some suites repeated the same checks. Group tests by responsibility, isolate feature fixtures in test classes, and remove duplicate or obsolete cases.

The review covered all 127 collected test modules on main. This reduces them and removes or combines definitions:.

 No application code changes.
This commit is contained in:
Normann
2026-10-28 20:15:25 +01:00
committed by GitHub
parent 2863373b17
commit 5d7198d090
31 changed files with 1719 additions and 2159 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CONTRIBUTING.md
+12
View File
@@ -112,6 +112,18 @@ Use `pytest` to run tests locally:
uv run python -m pytest -vs --cov src --cov-report term-missing tests/
```
Group tests by the module or cohesive subsystem they exercise. Add regression tests to the
existing suite and use descriptive test classes to separate features, edge cases, and integration
scenarios. Scope feature-specific fixtures to their test class, especially autouse fixtures that
could change unrelated tests. Create another test file only for a distinct responsibility or
when the existing suite would become unwieldy; avoid one file per bug or pull request.
For example, keep time-window model tests in `tests/test_configabc.py`, measurement channels
and persistence in `tests/test_measurement.py`, and GENETIC0 parameter preparation in
`tests/test_genetic0params.py`. Keep provider-specific suites and unit/system tests separate.
When moving tests, verify that parametrized cases and fixture behavior are preserved with
`pytest --collect-only` and an execution of the affected suites.
### Commit message style
Our commit message checks use
docs/development/eos-consolidation.md
-4
View File
@@ -155,9 +155,7 @@ python -m pytest `
tests/test_config.py `
tests/test_configabc.py `
tests/test_configmigrate.py `
tests/test_configfile.py `
tests/test_pydantic.py `
tests/test_consolidation_config.py `
tests/test_genetichomeappliance.py `
tests/test_genetic0battery.py `
tests/test_genetic0inverterefficiency.py `
@@ -170,11 +168,9 @@ python -m pytest `
tests/test_terminalvalue.py `
tests/test_tailvalue_physics.py `
tests/test_interpolator.py `
tests/test_measurement_channels.py `
tests/test_measurement_energy.py `
tests/test_measurement_household.py `
tests/test_battery_capacity.py `
tests/test_measurement_file_restore.py `
tests/test_measurement.py `
tests/test_genetic0optimize.py `
tests/test_geneticoptimize.py `
tests/test_battery.py
+73 -1
View File
@@ -1,9 +1,15 @@
import unittest
import numpy as np
import pytest
from pydantic import ValidationError
from akkudoktoreos.devices.devices import BatteriesCommonSettings
from akkudoktoreos.devices.genetic.battery import Battery, SolarPanelBatteryParameters
from akkudoktoreos.devices.genetic.battery import (
Battery,
ElectricVehicleParameters,
SolarPanelBatteryParameters,
)
@pytest.fixture
@@ -428,3 +434,69 @@ def test_full_target_charges_to_capacity_without_overcharge():
assert battery.current_soc_percentage() <= 100
assert battery.current_soc_percentage() == pytest.approx(100)
assert battery.charge_energy(wh=1000, hour=3) == (0, 0)
_EV_CHARGE_RATES = [0.0] + [amps / 16 for amps in range(6, 17)]
def _minimum_current_battery(kind=ElectricVehicleParameters, soc=99, maximum=100):
parameters = kind(
device_id="ev-test", capacity_wh=60000,
initial_soc_percentage=int(soc), charging_efficiency=0.95,
discharging_efficiency=1.0, max_charge_power_w=11040,
max_soc_percentage=maximum, charge_rates=_EV_CHARGE_RATES,
)
value = Battery(parameters, prediction_hours=4, slot_duration_h=0.25)
value.reset()
value.soc_wh = soc / 100 * value.capacity_wh
value.charge_array = np.full(4, 0.375)
return value
class TestEVMinimumCurrent(unittest.TestCase):
def test_ev_finishes_within_final_slot_at_unchanged_minimum_current(self):
for soc in [98.5, 99.0, 99.9]:
with self.subTest(soc=soc):
value = _minimum_current_battery(soc=soc)
missing = 60000 - value.soc_wh
stored, losses = value.charge_energy(None, 0, charge_factor=0.375)
self.assertAlmostEqual(stored, missing)
self.assertAlmostEqual(value.soc_wh, 60000)
self.assertAlmostEqual(losses, missing / 0.95 - missing)
self.assertEqual(value.charge_array[0], 0.375)
self.assertLessEqual(stored + losses, 4140 * 0.25)
def test_full_ev_has_no_charge_command_or_energy(self):
value = _minimum_current_battery(soc=100)
self.assertEqual(value.charge_energy(None, 0, charge_factor=0.375), (0.0, 0.0))
self.assertEqual(value.charge_array[0], 0)
def test_complete_minimum_current_slot_preserves_watt_hours(self):
value = _minimum_current_battery(soc=80)
stored, losses = value.charge_energy(None, 0, charge_factor=0.375)
self.assertAlmostEqual(stored + losses, 4140 * 0.25)
self.assertAlmostEqual(stored, 4140 * 0.25 * 0.95)
def test_ev_respects_a_lower_configured_maximum(self):
value = _minimum_current_battery(soc=95.9, maximum=96)
value.charge_energy(None, 0, charge_factor=0.375)
self.assertAlmostEqual(value.soc_wh, 60000 * 0.96)
def test_stationary_battery_keeps_discrete_rate_behavior(self):
value = _minimum_current_battery(SolarPanelBatteryParameters, soc=99)
self.assertEqual(value.charge_energy(None, 0, charge_factor=0.375), (0.0, 0.0))
self.assertEqual(value.soc_wh, 59400)
self.assertEqual(value.charge_array[0], 0)
def test_final_slot_preserves_each_configured_current(self):
for amps in range(6, 17):
with self.subTest(amps=amps):
value = _minimum_current_battery(soc=99.9)
factor = amps / 16
value.charge_array[0] = factor
missing = value.max_soc_wh - value.soc_wh
stored, losses = value.charge_energy(None, 0, charge_factor=factor)
self.assertAlmostEqual(stored, missing)
self.assertAlmostEqual(value.soc_wh, value.max_soc_wh)
self.assertAlmostEqual(losses, missing / 0.95 - missing)
self.assertEqual(value.charge_array[0], factor)
tests/test_config.py
+190 -102
View File
@@ -1,16 +1,15 @@
import json
import tempfile
from pathlib import Path
from typing import Any, Optional, Union
from unittest.mock import patch
from typing import Any
from unittest.mock import Mock, patch
import numpy as np
import pytest
from loguru import logger
from pydantic import IPvAnyAddress, ValidationError
from pydantic import ValidationError
from akkudoktoreos.config.config import ConfigEOS, GeneralSettings
from akkudoktoreos.devices.settings.batterysettings import BATTERY_DEFAULT_CHARGE_RATES
from akkudoktoreos.devices.devices import DevicesCommonSettings
def assert_values_equal(actual, expected):
@@ -24,13 +23,43 @@ def assert_values_equal(actual, expected):
assert actual == expected, f"Expected {expected}, but got {actual}"
# overwrite config_mixin fixture from conftest
@pytest.fixture(autouse=True)
def config_mixin():
# ------------------------------------
# Runtime settings priority (issue #1303)
# ------------------------------------
@pytest.fixture
def config_eos_file(config_eos_factory) -> ConfigEOS:
"""ConfigEOS with the EOS configuration file as an active settings source."""
return config_eos_factory(
init={
"with_init_settings": True,
"with_env_settings": True,
"with_dotenv_settings": False,
"with_file_settings": True,
"with_file_secret_settings": False,
}
)
def write_config_file(config_eos: ConfigEOS, settings: dict[str, Any]) -> None:
"""Write settings to the EOS configuration file and load them."""
settings = {"general": {"version": config_eos.general.version}, **settings}
config_file_path = config_eos.general.config_file_path
assert config_file_path is not None
config_file_path.write_text(json.dumps(settings), encoding="utf-8")
config_eos.reset_settings()
class TestConfigEOS:
"""Configuration loading and runtime overrides without the ConfigMixin patch."""
@pytest.fixture(autouse=True)
def config_mixin(self):
"""Exercise real configuration lookup within this class."""
pass
def test_fixture_new_config_file(config_default_dirs):
def test_fixture_new_config_file(self, config_default_dirs):
"""Assure fixture stash_config_file is working."""
config_default_dir_user, config_default_dir_cwd, _, _ = config_default_dirs
@@ -40,8 +69,7 @@ def test_fixture_new_config_file(config_default_dirs):
assert not config_file_user.exists()
assert not config_file_cwd.exists()
def test_config_constants(config_eos):
def test_config_constants(self, config_eos):
"""Test config constants are the way expected by the tests."""
assert config_eos.APP_NAME == "net.akkudoktor.eos"
assert config_eos.APP_AUTHOR == "akkudoktor"
@@ -49,8 +77,7 @@ def test_config_constants(config_eos):
assert config_eos.ENCODING == "UTF-8"
assert config_eos.CONFIG_FILE_NAME == "EOS.config.json"
def test_computed_paths(config_eos, monkeypatch):
def test_computed_paths(self, config_eos, monkeypatch):
"""Test computed paths for output and cache."""
# Ensure we do not have the environment patched by conftest.py
@@ -83,8 +110,7 @@ def test_computed_paths(config_eos, monkeypatch):
# reset settings so the config_eos fixture can verify the default paths
config_eos.reset_settings()
def test_config_from_env(monkeypatch, config_eos):
def test_config_from_env(self, monkeypatch, config_eos):
"""Test configuration from env."""
assert config_eos.server.port == 8503
assert config_eos.server.eosdash_port == 8504
@@ -97,26 +123,32 @@ def test_config_from_env(monkeypatch, config_eos):
assert config_eos.server.port == 8553
assert config_eos.server.eosdash_port == 8555
def test_config_ipaddress(monkeypatch, config_eos):
"""Test configuration for IP adresses."""
def test_config_ipaddress(self, monkeypatch, config_eos):
"""Validate configured IPs and hostnames without external DNS."""
resolver = Mock(side_effect={
"mail.akkudoktor.net": "192.0.2.1",
"localhost": "127.0.0.1",
}.__getitem__)
monkeypatch.setattr("akkudoktoreos.server.server.socket.gethostbyname", resolver)
assert config_eos.server.host == "127.0.0.1"
monkeypatch.setenv("EOS_SERVER__HOST", "0.0.0.0")
config_eos.reset_settings()
assert config_eos.server.host == "0.0.0.0"
resolver.assert_not_called()
monkeypatch.setenv("EOS_SERVER__HOST", "mail.akkudoktor.net")
config_eos.reset_settings()
assert config_eos.server.host == "mail.akkudoktor.net"
resolver.assert_called_with("mail.akkudoktor.net")
# keep last
monkeypatch.setenv("EOS_SERVER__HOST", "localhost")
config_eos.reset_settings()
assert config_eos.server.host == "localhost"
resolver.assert_called_with("localhost")
def test_singleton_behavior(config_eos, config_default_dirs, monkeypatch):
def test_singleton_behavior(self, config_eos, config_default_dirs, monkeypatch):
"""Test that ConfigEOS behaves as a singleton."""
config_eos.reset_instance()
@@ -129,8 +161,7 @@ def test_singleton_behavior(config_eos, config_default_dirs, monkeypatch):
assert instance1 is instance2
assert instance1._config_file_path == instance2._config_file_path
def test_config_file_priority(config_default_dirs):
def test_config_file_priority(self, config_default_dirs):
"""Test config file priority.
Priority is:
@@ -160,9 +191,8 @@ def test_config_file_priority(config_default_dirs):
config_eos.update()
assert config_eos.general.config_file_path == config_file_user
@patch("akkudoktoreos.config.config.user_config_dir")
def test_get_config_file_path(user_config_dir_patch, config_eos, config_default_dirs, monkeypatch):
@patch("akkudoktoreos.config.config.user_config_dir")
def test_get_config_file_path(self, user_config_dir_patch, config_eos, config_default_dirs, monkeypatch):
"""Test that _get_config_file_path identifies the correct config file."""
config_default_dir_user, _, _, _ = config_default_dirs
user_config_dir_patch.return_value = str(config_default_dir_user)
@@ -209,8 +239,7 @@ def test_get_config_file_path(user_config_dir_patch, config_eos, config_default_
False,
)
def test_config_copy(config_eos, monkeypatch):
def test_config_copy(self, config_eos, monkeypatch):
"""Test if the config is copied to the provided path."""
with tempfile.TemporaryDirectory() as temp_dir:
temp_folder_path = Path(temp_dir)
@@ -222,16 +251,15 @@ def test_config_copy(config_eos, monkeypatch):
config_eos.update()
assert temp_config_file_path.exists()
@pytest.mark.parametrize(
@pytest.mark.parametrize(
"latitude, longitude, expected_timezone",
[
(40.7128, -74.0060, "America/New_York"), # Valid latitude/longitude
(None, None, None), # No location
(51.5074, -0.1278, "Europe/London"), # Another valid location
],
)
def test_config_common_settings_valid(latitude, longitude, expected_timezone):
)
def test_config_common_settings_valid(self, latitude, longitude, expected_timezone):
"""Test valid settings for GeneralSettings."""
general_settings = GeneralSettings(
latitude=latitude,
@@ -241,8 +269,7 @@ def test_config_common_settings_valid(latitude, longitude, expected_timezone):
assert general_settings.longitude == longitude
assert general_settings.timezone == expected_timezone
@pytest.mark.parametrize(
@pytest.mark.parametrize(
"field_name, invalid_value, expected_error",
[
("latitude", -91.0, "Input should be greater than or equal to -90"),
@@ -250,8 +277,8 @@ def test_config_common_settings_valid(latitude, longitude, expected_timezone):
("longitude", -181.0, "Input should be greater than or equal to -180"),
("longitude", 181.0, "Input should be less than or equal to 180"),
],
)
def test_config_common_settings_invalid(field_name, invalid_value, expected_error):
)
def test_config_common_settings_invalid(self, field_name, invalid_value, expected_error):
"""Test invalid settings for PredictionCommonSettings."""
valid_data = {
"latitude": 40.7128,
@@ -263,20 +290,17 @@ def test_config_common_settings_invalid(field_name, invalid_value, expected_erro
with pytest.raises(ValidationError, match=expected_error):
GeneralSettings.model_validate(valid_data)
def test_config_common_settings_no_location():
def test_config_common_settings_no_location(self):
"""Test that timezone is None when latitude and longitude are not provided."""
settings = GeneralSettings(latitude=None, longitude=None)
assert settings.timezone is None
def test_config_common_settings_with_location():
def test_config_common_settings_with_location(self):
"""Test that timezone is correctly computed when latitude and longitude are provided."""
settings = GeneralSettings(latitude=34.0522, longitude=-118.2437)
assert settings.timezone == "America/Los_Angeles"
def test_config_common_settings_timezone_none_when_coordinates_missing():
def test_config_common_settings_timezone_none_when_coordinates_missing(self):
"""Test that timezone is None when latitude or longitude is missing."""
config_no_latitude = GeneralSettings(latitude=None, longitude=-74.0060)
config_no_longitude = GeneralSettings(latitude=40.7128, longitude=None)
@@ -286,9 +310,7 @@ def test_config_common_settings_timezone_none_when_coordinates_missing():
assert config_no_longitude.timezone is None
assert config_no_coords.timezone is None
# Test partial assignments and possible side effects
@pytest.mark.parametrize(
@pytest.mark.parametrize(
"path, value, expected, exception",
[
# Correct value assignment
@@ -465,8 +487,8 @@ def test_config_common_settings_timezone_none_when_coordinates_missing():
TypeError,
),
],
)
def test_set_nested_key(path, value, expected, exception, config_eos):
)
def test_set_nested_key(self, path, value, expected, exception, config_eos):
if not exception:
config_eos.set_nested_value(path, value)
for expected_path, expected_value in expected:
@@ -486,36 +508,33 @@ def test_set_nested_key(path, value, expected, exception, config_eos):
f"Expected exception {exception}, but got {type(e)}: {e}"
)
def test_set_nested_key_read_only_computed_field(config_eos):
def test_set_nested_key_read_only_computed_field(self, config_eos):
"""Setting a computed field is reported as read-only, not as missing."""
path = "adapter/homeassistant/eos_device_instruction_entity_ids"
with pytest.raises(TypeError, match="read-only computed field"):
config_eos.set_nested_value(path, ["sensor.eos_battery1"])
def test_set_nested_key_unknown_field_still_reports_missing(config_eos):
def test_set_nested_key_unknown_field_still_reports_missing(self, config_eos):
"""A field that does not exist keeps the original 'does not exist' message."""
with pytest.raises(TypeError, match="does not exist in model"):
config_eos.set_nested_value("adapter/homeassistant/no_such_field", 1)
@pytest.mark.parametrize(
@pytest.mark.parametrize(
"path, expected_value, exception",
[
("general/latitude", 52.52, None),
("general/latitude/", 52.52, None),
("general/latitude/test", None, KeyError),
],
)
def test_get_nested_key(path, expected_value, exception, config_eos):
)
def test_get_nested_key(self, path, expected_value, exception, config_eos):
if not exception:
assert config_eos.get_nested_value(path) == expected_value
else:
with pytest.raises(exception):
config_eos.get_nested_value(path)
def test_merge_settings_from_dict_invalid(config_eos):
def test_merge_settings_from_dict_invalid(self, config_eos):
"""Test merging invalid data."""
invalid_settings = {
"general": {
@@ -526,8 +545,7 @@ def test_merge_settings_from_dict_invalid(config_eos):
with pytest.raises(Exception): # Pydantic ValidationError expected
config_eos.merge_settings_from_dict(invalid_settings)
def test_merge_settings_partial(config_eos):
def test_merge_settings_partial(self, config_eos):
"""Test merging only a subset of settings."""
partial_settings: dict[str, Any] = {
@@ -599,8 +617,7 @@ def test_merge_settings_partial(config_eos):
assert len(config_eos.devices.electric_vehicles) == 1
assert_values_equal(config_eos.devices.electric_vehicles["ev1"].charge_rates, [0.0, 0.375, 0.5, 0.625, 0.75, 0.875, 1.0])
def test_merge_settings_empty(config_eos):
def test_merge_settings_empty(self, config_eos):
"""Test merging an empty dictionary does not change settings."""
original_latitude = config_eos.general.latitude
@@ -608,36 +625,7 @@ def test_merge_settings_empty(config_eos):
assert config_eos.general.latitude == original_latitude # Should remain unchanged
# ------------------------------------
# Runtime settings priority (issue #1303)
# ------------------------------------
@pytest.fixture
def config_eos_file(config_eos_factory) -> ConfigEOS:
"""ConfigEOS with the EOS configuration file as an active settings source."""
return config_eos_factory(
init={
"with_init_settings": True,
"with_env_settings": True,
"with_dotenv_settings": False,
"with_file_settings": True,
"with_file_secret_settings": False,
}
)
def write_config_file(config_eos: ConfigEOS, settings: dict[str, Any]) -> None:
"""Write settings to the EOS configuration file and load them."""
settings = {"general": {"version": config_eos.general.version}, **settings}
config_file_path = config_eos.general.config_file_path
assert config_file_path is not None
config_file_path.write_text(json.dumps(settings), encoding="utf-8")
config_eos.reset_settings()
def test_merge_settings_overrides_config_file(config_eos_file):
def test_merge_settings_overrides_config_file(self, config_eos_file):
"""Runtime settings take precedence over the EOS configuration file."""
write_config_file(
config_eos_file,
@@ -666,8 +654,7 @@ def test_merge_settings_overrides_config_file(config_eos_file):
assert config_eos_file.optimization.genetic.individuals == 300
assert len(config_eos_file.pvforecast.planes) == 1
def test_merge_settings_overrides_env(config_eos_file, monkeypatch):
def test_merge_settings_overrides_env(self, config_eos_file, monkeypatch):
"""Runtime settings take precedence over environment variables."""
monkeypatch.setenv("EOS_OPTIMIZATION__GENETIC__INDIVIDUALS", "150")
config_eos_file.reset_settings()
@@ -677,8 +664,7 @@ def test_merge_settings_overrides_env(config_eos_file, monkeypatch):
assert config_eos_file.optimization.genetic.individuals == 300
def test_env_overrides_config_file_after_merge(config_eos_file, monkeypatch):
def test_env_overrides_config_file_after_merge(self, config_eos_file, monkeypatch):
"""Environment variables keep precedence over the config file for untouched keys."""
write_config_file(config_eos_file, {"server": {"port": 9000}})
monkeypatch.setenv("EOS_SERVER__PORT", "9500")
@@ -694,8 +680,7 @@ def test_env_overrides_config_file_after_merge(config_eos_file, monkeypatch):
config_eos_file.reset_settings()
assert config_eos_file.server.port == 9600
def test_reset_settings_drops_runtime_settings(config_eos_file):
def test_reset_settings_drops_runtime_settings(self, config_eos_file):
"""Reset drops runtime settings and falls back to the config file."""
write_config_file(config_eos_file, {"optimization": {"genetic": {"individuals": 200}}})
@@ -705,8 +690,7 @@ def test_reset_settings_drops_runtime_settings(config_eos_file):
config_eos_file.reset_settings()
assert config_eos_file.optimization.genetic.individuals == 200
def test_set_nested_value_survives_merge(config_eos_file):
def test_set_nested_value_survives_merge(self, config_eos_file):
"""Granular updates are not lost by a later bulk update."""
write_config_file(
config_eos_file,
@@ -735,8 +719,7 @@ def test_set_nested_value_survives_merge(config_eos_file):
assert config_eos_file.pvforecast.planes[1].peakpower == 9.9
assert config_eos_file.general.latitude is None
def test_revert_settings_restores_backup(config_eos_file):
def test_revert_settings_restores_backup(self, config_eos_file):
"""Revert restores the backup values even if the config file differs."""
write_config_file(config_eos_file, {"optimization": {"genetic": {"individuals": 200}}})
@@ -757,8 +740,7 @@ def test_revert_settings_restores_backup(config_eos_file):
assert config_eos_file.optimization.genetic.individuals == 500
def test_config_from_env_on_first_init(config_eos, config_default_dirs, monkeypatch):
def test_config_from_env_on_first_init(self, config_eos, config_default_dirs, monkeypatch):
"""Environment variables are applied on the first configuration build."""
config_eos.reset_instance()
@@ -766,3 +748,109 @@ def test_config_from_env_on_first_init(config_eos, config_default_dirs, monkeypa
monkeypatch.setenv("EOS_SERVER__PORT", "8553")
assert ConfigEOS().server.port == 8553
class TestConfigEOSToConfigFile:
def test_to_config_file_writes_file(self, config_eos):
config_path = config_eos.general.config_file_path
# Remove file to test writing
config_path.unlink(missing_ok=True)
config_eos.to_config_file()
assert config_path.exists()
assert config_path.read_text().strip().startswith("{")
def test_to_config_file_excludes_computed_fields(self, config_eos):
config_path = config_eos.general.config_file_path
config_eos.to_config_file()
data = json.loads(config_path.read_text())
assert "timezone" not in data["general"]
assert "data_output_path" not in data["general"]
assert "config_folder_path" not in data["general"]
assert "config_file_path" not in data["general"]
def test_to_config_file_excludes_defaults(self, config_eos):
"""Ensure fields with default values are excluded when saving config."""
# Pick fields that have defaults
default_latitude = GeneralSettings.model_fields["latitude"].default
default_longitude = GeneralSettings.model_fields["longitude"].default
# Ensure fields are at default values
config_eos.general.latitude = default_latitude
config_eos.general.longitude = default_longitude
# Save the config using the correct path managed by config_eos
config_eos.to_config_file()
# Read back JSON from the correct path
config_file_path = config_eos.general.config_file_path
content = json.loads(config_file_path.read_text(encoding="utf-8"))
# Default fields should not appear
assert "latitude" not in content["general"]
assert "longitude" not in content["general"]
# Non-default value should appear
config_eos.general.latitude = 48.0
config_eos.to_config_file()
content = json.loads(config_file_path.read_text(encoding="utf-8"))
assert content["general"]["latitude"] == 48.0
def test_to_config_file_excludes_none_fields(self, config_eos):
config_eos.general.latitude = None
config_path = config_eos.general.config_file_path
config_eos.to_config_file()
data = json.loads(config_path.read_text())
assert "latitude" not in data["general"]
def test_to_config_file_includes_version(tmp_path, config_eos):
"""Ensure general.version is always included."""
# Save config
config_eos.to_config_file()
# Read back JSON
config_file_path = config_eos.general.config_file_path
content = json.loads(config_file_path.read_text(encoding="utf-8"))
# Assert 'version' is included even if default
assert content["general"]["version"] == config_eos.general.version
def test_to_config_file_roundtrip(self, config_eos):
config_eos.merge_settings_from_dict(
{
"general": {"latitude": 48.0},
"server": {"port": 9000},
}
)
config_path = config_eos.general.config_file_path
config_eos.to_config_file()
raw_data = json.loads(config_path.read_text())
reloaded = ConfigEOS.model_validate(raw_data)
assert reloaded.general.latitude == 48.0
assert reloaded.server.port == 9000
class TestDeviceSettings:
def test_device_map_supplies_stable_identity(self):
raw = {"batteries": {"house": {"capacity_wh": 12000}}}
a = DevicesCommonSettings.model_validate(raw)
b = DevicesCommonSettings.model_validate(raw)
assert a.batteries is not None and b.batteries is not None
assert a.batteries["house"].device_id == b.batteries["house"].device_id == "house"
assert "house-soc-factor" in a.measurement_keys
def test_device_map_rejects_conflicting_identity(self):
with pytest.raises(ValidationError, match="device_id"):
DevicesCommonSettings.model_validate({"batteries": {"house": {"device_id": "other"}}})
tests/test_configabc.py
+3 -12
View File
@@ -10,17 +10,11 @@ Timezone contract under test:
* When a timezone-aware datetime is supplied, ``start_time`` is
interpreted as wall-clock time **in that timezone** — no tz
conversion is applied to ``start_time`` itself.
* Constructing a ``TimeWindow`` with a naive ``start_time`` raises
``ValidationError``.
* A timezone-aware ``start_time`` is normalized to a naive wall-clock time.
"""
import datetime
import json
import os
import sys
sys.path.insert(0, os.path.dirname(__file__))
from typing import cast
import numpy as np
@@ -29,12 +23,9 @@ import pendulum
import pytest
from pydantic import ValidationError
from akkudoktoreos.config.configabc import ( # noqa — ensure Time is importable
from akkudoktoreos.config.configabc import (
CycleTimeWindowSequence,
)
from akkudoktoreos.config.configabc import TimeWindow
from akkudoktoreos.config.configabc import TimeWindow as _TW_check
from akkudoktoreos.config.configabc import ( # noqa — ensure Time is importable
TimeWindow,
TimeWindowSequence,
ValueTimeWindow,
ValueTimeWindowSequence,
tests/test_configfile.py
-99
View File
@@ -1,99 +0,0 @@
import json
from pathlib import Path
from unittest.mock import patch
import pytest
from akkudoktoreos.config.config import ConfigEOS, GeneralSettings
class TestConfigEOSToConfigFile:
def test_to_config_file_writes_file(self, config_eos):
config_path = config_eos.general.config_file_path
# Remove file to test writing
config_path.unlink(missing_ok=True)
config_eos.to_config_file()
assert config_path.exists()
assert config_path.read_text().strip().startswith("{")
def test_to_config_file_excludes_computed_fields(self, config_eos):
config_path = config_eos.general.config_file_path
config_eos.to_config_file()
data = json.loads(config_path.read_text())
assert "timezone" not in data["general"]
assert "data_output_path" not in data["general"]
assert "config_folder_path" not in data["general"]
assert "config_file_path" not in data["general"]
def test_to_config_file_excludes_defaults(self, config_eos):
"""Ensure fields with default values are excluded when saving config."""
# Pick fields that have defaults
default_latitude = GeneralSettings.model_fields["latitude"].default
default_longitude = GeneralSettings.model_fields["longitude"].default
# Ensure fields are at default values
config_eos.general.latitude = default_latitude
config_eos.general.longitude = default_longitude
# Save the config using the correct path managed by config_eos
config_eos.to_config_file()
# Read back JSON from the correct path
config_file_path = config_eos.general.config_file_path
content = json.loads(config_file_path.read_text(encoding="utf-8"))
# Default fields should not appear
assert "latitude" not in content["general"]
assert "longitude" not in content["general"]
# Non-default value should appear
config_eos.general.latitude = 48.0
config_eos.to_config_file()
content = json.loads(config_file_path.read_text(encoding="utf-8"))
assert content["general"]["latitude"] == 48.0
def test_to_config_file_excludes_none_fields(self, config_eos):
config_eos.general.latitude = None
config_path = config_eos.general.config_file_path
config_eos.to_config_file()
data = json.loads(config_path.read_text())
assert "latitude" not in data["general"]
def test_to_config_file_includes_version(tmp_path, config_eos):
"""Ensure general.version is always included."""
# Save config
config_eos.to_config_file()
# Read back JSON
config_file_path = config_eos.general.config_file_path
content = json.loads(config_file_path.read_text(encoding="utf-8"))
# Assert 'version' is included even if default
assert content["general"]["version"] == config_eos.general.version
def test_to_config_file_roundtrip(self, config_eos):
config_eos.merge_settings_from_dict(
{
"general": {"latitude": 48.0},
"server": {"port": 9000},
}
)
config_path = config_eos.general.config_file_path
config_eos.to_config_file()
raw_data = json.loads(config_path.read_text())
reloaded = ConfigEOS.model_validate(raw_data)
assert reloaded.general.latitude == 48.0
assert reloaded.server.port == 9000
tests/test_configmigrate.py
+16 -2
View File
@@ -1,6 +1,7 @@
# Test data directory and known migration pairs
import copy
import json
import shutil
import tempfile
from pathlib import Path
from typing import Any
@@ -8,9 +9,9 @@ import pytest
from akkudoktoreos.config import configmigrate
from akkudoktoreos.config.config import ConfigEOS, SettingsEOSDefaults
from akkudoktoreos.config.configmigrate import migrate_config_data
from akkudoktoreos.core.version import __version__
# Test data directory and known migration pairs
DIR_TESTDATA = Path(__file__).absolute().parent.joinpath("testdata")
MIGRATION_PAIRS = [
@@ -250,3 +251,16 @@ class TestConfigMigration:
# Remove the .new working file only if the test passed (failed == False)
if not failed and working_file.exists():
working_file.unlink(missing_ok=True)
class TestDeviceMigration:
@pytest.mark.parametrize("as_list", [False, True])
def test_migration_preserves_lcos_and_input(self, as_list):
battery = {"device_id": "house", "capacity_wh": 12000,
"levelized_cost_of_storage_kwh": 0.123}
raw = {"devices": {"batteries": [battery] if as_list else {"house": battery}}}
original = copy.deepcopy(raw)
migrated = migrate_config_data(raw)
assert migrated.devices.batteries is not None
assert migrated.devices.batteries["house"].levelized_cost_of_storage_amt_kwh == 0.123
assert raw == original
tests/test_consolidation_config.py
-44
View File
@@ -1,44 +0,0 @@
"""Contracts required by the combined EOS configuration."""
import copy
import pytest
from pydantic import ValidationError
from akkudoktoreos.config.configmigrate import migrate_config_data
from akkudoktoreos.devices.devices import DevicesCommonSettings
from akkudoktoreos.devices.settings.invertersettings import InverterCommonSettings
def test_device_map_supplies_stable_identity():
raw = {"batteries": {"house": {"capacity_wh": 12000}}}
a = DevicesCommonSettings.model_validate(raw)
b = DevicesCommonSettings.model_validate(raw)
assert a.batteries is not None and b.batteries is not None
assert a.batteries["house"].device_id == b.batteries["house"].device_id == "house"
assert "house-soc-factor" in a.measurement_keys
def test_device_map_rejects_conflicting_identity():
with pytest.raises(ValidationError, match="device_id"):
DevicesCommonSettings.model_validate({"batteries": {"house": {"device_id": "other"}}})
@pytest.mark.parametrize("as_list", [False, True])
def test_migration_preserves_lcos_and_input(as_list):
battery = {"device_id": "house", "capacity_wh": 12000,
"levelized_cost_of_storage_kwh": 0.123}
raw = {"devices": {"batteries": [battery] if as_list else {"house": battery}}}
original = copy.deepcopy(raw)
migrated = migrate_config_data(raw)
assert migrated.devices.batteries is not None
assert migrated.devices.batteries["house"].levelized_cost_of_storage_amt_kwh == 0.123
assert raw == original
@pytest.mark.parametrize("converter", ["to_genetic_param", "to_genetic0_param"])
def test_inverter_conversion_requires_output_limit(converter):
settings = InverterCommonSettings(device_id="inverter")
with pytest.raises(ValueError, match="max_power_w"):
getattr(settings, converter)()
settings.max_power_w = 4200
assert getattr(settings, converter)().max_power_wh == 4200
tests/test_consolidation_file_restore.py
-29
View File
@@ -1,29 +0,0 @@
"""Verify fallback JSON loading does not lose records through the singleton."""
from unittest.mock import AsyncMock
import pytest
from akkudoktoreos.core.coreabc import get_measurement
from akkudoktoreos.core.dataabc import DataSequence
from akkudoktoreos.utils.datetimeutil import to_datetime
@pytest.mark.asyncio
async def test_measurement_json_roundtrip(config_eos, tmp_path, monkeypatch):
m = get_measurement()
config_eos.measurement.load_emr_keys = ["meter"]
config_eos.general.data_folder_path = tmp_path
config_eos.database.provider = None
m._db_reset_state()
try:
await m.update_value(to_datetime("2026-09-16T08:00:00Z"), "meter", 123.5)
monkeypatch.setattr(DataSequence, "save", AsyncMock(return_value=False))
monkeypatch.setattr(DataSequence, "load", AsyncMock(return_value=False))
assert await m.save()
m._db_reset_state()
assert await m.load()
assert len(m.records) == 1
assert m.records[0]["meter"] == 123.5
finally:
m._db_reset_state()
tests/test_elecfeefixed.py
+1 -4
View File
@@ -1,12 +1,9 @@
"""Tests for fixed electricity fee prediction module."""
import asyncio
import json
from pathlib import Path
from unittest.mock import Mock, patch
import numpy as np
import pandas as pd
import pytest
from akkudoktoreos.config.configabc import ValueTimeWindow, ValueTimeWindowSequence
@@ -16,7 +13,7 @@ from akkudoktoreos.prediction.elecfeefixed import (
ElecFeeFixed,
ElecFeeFixedCommonSettings,
)
from akkudoktoreos.utils.datetimeutil import Duration, to_datetime
from akkudoktoreos.utils.datetimeutil import to_datetime
DIR_TESTDATA = Path(__file__).absolute().parent.joinpath("testdata")
FILE_TESTDATA_ELECFEEFIXED_CONFIG_JSON = DIR_TESTDATA.joinpath("elecfeefixed_config.json")
tests/test_elecpriceakkudoktor.py
-1
View File
@@ -1,4 +1,3 @@
import asyncio
import json
from pathlib import Path
from unittest.mock import Mock, patch
tests/test_elecpriceenergycharts.py
-7
View File
@@ -1,4 +1,3 @@
import asyncio
import json
from collections.abc import Callable
from pathlib import Path
@@ -8,16 +7,10 @@ import numpy as np
import pandas as pd
import pytest
import requests
from loguru import logger
from akkudoktoreos.core.cache import CacheFileStore
from akkudoktoreos.core.coreabc import get_ems
from akkudoktoreos.prediction.elecfeefixed import ElecFeeFixed
from akkudoktoreos.prediction.elecpriceakkudoktor import (
AkkudoktorElecPrice,
AkkudoktorElecPriceValue,
ElecPriceAkkudoktor,
)
from akkudoktoreos.prediction.elecpriceenergycharts import (
ElecPriceEnergyCharts,
EnergyChartsElecPrice,
tests/test_elecpricefixed.py
+1 -5
View File
@@ -1,12 +1,8 @@
"""Tests for fixed electricity price prediction module."""
import asyncio
import json
from pathlib import Path
from unittest.mock import Mock, patch
import numpy as np
import pandas as pd
import pytest
from akkudoktoreos.config.configabc import ValueTimeWindow, ValueTimeWindowSequence
@@ -16,7 +12,7 @@ from akkudoktoreos.prediction.elecpricefixed import (
ElecPriceFixed,
ElecPriceFixedCommonSettings,
)
from akkudoktoreos.utils.datetimeutil import Duration, to_datetime
from akkudoktoreos.utils.datetimeutil import to_datetime
DIR_TESTDATA = Path(__file__).absolute().parent.joinpath("testdata")
FILE_TESTDATA_ELECPRICEFIXED_CONFIG_JSON = DIR_TESTDATA.joinpath("elecpricefixed_config.json")
tests/test_emplan.py
+35 -2
View File
@@ -1,9 +1,7 @@
from typing import Optional
import pytest
from akkudoktoreos.core.emplan import (
BaseInstruction,
CommodityQuantity,
DDBCInstruction,
EnergyManagementInstruction,
@@ -365,3 +363,38 @@ class TestEnergyManagementPlan:
active_after = plan.get_active_instructions(exec_future)
assert {i.resource_id for i in active_after} == {"dev-1"}
class TestEnergyManagementPlanDST:
def test_repeated_hour_keeps_instruction_order_and_dispatch(self):
first = to_datetime("2026-10-25T02:45:00+02:00").in_timezone("Europe/Berlin")
second = to_datetime("2026-10-25T02:15:00+01:00").in_timezone("Europe/Berlin")
plan = EnergyManagementPlan(id="fold", generated_at=first, instructions=[])
later = FRBCInstruction(
resource_id="battery1",
actuator_id="battery1",
execution_time=second,
operation_mode_id="IDLE",
operation_mode_factor=1.0,
)
earlier = FRBCInstruction(
resource_id="battery1",
actuator_id="battery1",
execution_time=first,
operation_mode_id="GRID_SUPPORT_IMPORT",
operation_mode_factor=0.5,
)
plan.add_instruction(later)
plan.add_instruction(earlier)
assert [item.execution_time.timestamp() for item in plan.instructions] == [
first.timestamp(),
second.timestamp(),
]
assert plan.valid_from is not None
assert plan.valid_from.timestamp() == first.timestamp()
assert plan.valid_until is None
between = to_datetime("2026-10-25T02:00:00+01:00").in_timezone("Europe/Berlin")
assert plan.get_active_instructions(between) == [earlier]
assert plan.get_next_instruction(between) == later
assert plan.get_active_instructions(second) == [later]
assert plan.get_next_instruction(second) is None
tests/test_emplan_dst_instants.py
-38
View File
@@ -1,38 +0,0 @@
"""Repeated local clock times must dispatch by their real UTC instant."""
from akkudoktoreos.core.emplan import EnergyManagementPlan, FRBCInstruction
from akkudoktoreos.utils.datetimeutil import to_datetime
def test_repeated_hour_keeps_instruction_order_and_dispatch():
first = to_datetime("2026-10-25T02:45:00+02:00").in_timezone("Europe/Berlin")
second = to_datetime("2026-10-25T02:15:00+01:00").in_timezone("Europe/Berlin")
plan = EnergyManagementPlan(id="fold", generated_at=first, instructions=[])
later = FRBCInstruction(
resource_id="battery1",
actuator_id="battery1",
execution_time=second,
operation_mode_id="IDLE",
operation_mode_factor=1.0,
)
earlier = FRBCInstruction(
resource_id="battery1",
actuator_id="battery1",
execution_time=first,
operation_mode_id="GRID_SUPPORT_IMPORT",
operation_mode_factor=0.5,
)
plan.add_instruction(later)
plan.add_instruction(earlier)
assert [item.execution_time.timestamp() for item in plan.instructions] == [
first.timestamp(),
second.timestamp(),
]
assert plan.valid_from is not None
assert plan.valid_from.timestamp() == first.timestamp()
assert plan.valid_until is None
between = to_datetime("2026-10-25T02:00:00+01:00").in_timezone("Europe/Berlin")
assert plan.get_active_instructions(between) == [earlier]
assert plan.get_next_instruction(between) == later
assert plan.get_active_instructions(second) == [later]
assert plan.get_next_instruction(second) is None
tests/test_ev_minimum_current.py
-78
View File
@@ -1,78 +0,0 @@
import unittest
import numpy as np
from akkudoktoreos.devices.genetic.battery import (
Battery,
ElectricVehicleParameters,
SolarPanelBatteryParameters,
)
RATES = [0.0] + [amps / 16 for amps in range(6, 17)]
def battery(kind=ElectricVehicleParameters, soc=99, maximum=100):
parameters = kind(
device_id="ev-test", capacity_wh=60000,
initial_soc_percentage=int(soc), charging_efficiency=0.95,
discharging_efficiency=1.0, max_charge_power_w=11040,
max_soc_percentage=maximum, charge_rates=RATES,
)
value = Battery(parameters, prediction_hours=4, slot_duration_h=0.25)
value.reset()
value.soc_wh = soc / 100 * value.capacity_wh
value.charge_array = np.full(4, 0.375)
return value
class TestEVMinimumCurrent(unittest.TestCase):
def test_ev_finishes_within_final_slot_at_unchanged_minimum_current(self):
for soc in [98.5, 99.0, 99.9]:
with self.subTest(soc=soc):
value = battery(soc=soc)
missing = 60000 - value.soc_wh
stored, losses = value.charge_energy(None, 0, charge_factor=0.375)
self.assertAlmostEqual(stored, missing)
self.assertAlmostEqual(value.soc_wh, 60000)
self.assertAlmostEqual(losses, missing / 0.95 - missing)
self.assertEqual(value.charge_array[0], 0.375)
self.assertLessEqual(stored + losses, 4140 * 0.25)
def test_full_ev_has_no_charge_command_or_energy(self):
value = battery(soc=100)
self.assertEqual(value.charge_energy(None, 0, charge_factor=0.375), (0.0, 0.0))
self.assertEqual(value.charge_array[0], 0)
def test_complete_minimum_current_slot_preserves_watt_hours(self):
value = battery(soc=80)
stored, losses = value.charge_energy(None, 0, charge_factor=0.375)
self.assertAlmostEqual(stored + losses, 4140 * 0.25)
self.assertAlmostEqual(stored, 4140 * 0.25 * 0.95)
def test_ev_respects_a_lower_configured_maximum(self):
value = battery(soc=95.9, maximum=96)
value.charge_energy(None, 0, charge_factor=0.375)
self.assertAlmostEqual(value.soc_wh, 60000 * 0.96)
def test_stationary_battery_keeps_discrete_rate_behavior(self):
value = battery(SolarPanelBatteryParameters, soc=99)
self.assertEqual(value.charge_energy(None, 0, charge_factor=0.375), (0.0, 0.0))
self.assertEqual(value.soc_wh, 59400)
self.assertEqual(value.charge_array[0], 0)
def test_final_slot_preserves_each_configured_current(self):
for amps in range(6, 17):
with self.subTest(amps=amps):
value = battery(soc=99.9)
factor = amps / 16
value.charge_array[0] = factor
missing = value.max_soc_wh - value.soc_wh
stored, losses = value.charge_energy(None, 0, charge_factor=factor)
self.assertAlmostEqual(stored, missing)
self.assertAlmostEqual(value.soc_wh, value.max_soc_wh)
self.assertAlmostEqual(losses, missing / 0.95 - missing)
self.assertEqual(value.charge_array[0], factor)
if __name__ == "__main__":
unittest.main()
tests/test_genetic0_measured_soc.py
-160
View File
@@ -1,160 +0,0 @@
"""GENETIC0 must use the SoC read during the run, not an older reading (#1447)."""
from unittest.mock import AsyncMock, Mock, patch
import numpy as np
import pandas as pd
import pytest
import pytest_asyncio
from akkudoktoreos.core.coreabc import get_ems, get_measurement
from akkudoktoreos.optimization.genetic0.genetic0params import (
Genetic0OptimizationParameters,
)
from akkudoktoreos.utils.datetimeutil import to_datetime
@pytest_asyncio.fixture
async def isolated_measurement(config_eos):
"""Keep synthetic records out of the process-wide measurement singleton."""
measurement = get_measurement()
await measurement.delete_by_datetime(None, None)
try:
yield measurement
finally:
await measurement.delete_by_datetime(None, None)
@pytest.mark.asyncio
async def test_soc_read_late_in_slot_beats_previous_reading(config_eos, isolated_measurement):
config_eos.merge_settings_from_dict(
{
"prediction": {"hours": 48},
"optimization": {"genetic0": {"horizon_hours": 24}},
"elecfee": {"provider": None},
"devices": {
"max_batteries": 1,
"max_electric_vehicles": 0,
"max_inverters": 1,
"max_home_appliances": 0,
"batteries": {"storage": {"capacity_wh": 2000, "max_charge_power_w": 1000}},
"inverters": {"inverter": {"battery_id": "storage", "max_power_w": 1000}},
},
}
)
# A run at 14:40 belongs to the 14:00 slot. The previous run (13:40) is closer
# to the slot start than the reading taken now.
run_time = to_datetime("2026-08-01T14:40:00+00:00")
get_ems().set_start_datetime(run_time)
measurement = isolated_measurement
await measurement.update_value(run_time.subtract(hours=1), "storage-soc-factor", 0.10)
await measurement.update_value(run_time, "storage-soc-factor", 0.42)
tariff = pd.Series(
0.00008, index=pd.date_range("2026-08-01T00:00:00Z", periods=72 * 4, freq="15min")
)
async def read_array(**kwargs):
return np.full(48, 0.0003)
prediction = Mock(
update_data=AsyncMock(),
key_to_array=AsyncMock(side_effect=read_array),
key_to_raw_series=AsyncMock(return_value=tariff),
)
with patch.object(Genetic0OptimizationParameters, "prediction", prediction):
parameters = await Genetic0OptimizationParameters.prepare()
assert parameters is not None
assert parameters.pv_battery is not None
assert parameters.pv_battery.initial_soc_percentage == 42
@pytest.mark.asyncio
async def test_meter_only_record_does_not_mask_soc(config_eos, isolated_measurement):
config_eos.merge_settings_from_dict(
{
"prediction": {"hours": 48},
"optimization": {"genetic0": {"horizon_hours": 24}},
"elecfee": {"provider": None},
"measurement": {"load_emr_keys": ["load_meter"]},
"devices": {
"max_batteries": 1,
"max_electric_vehicles": 0,
"max_inverters": 1,
"max_home_appliances": 0,
"batteries": {"storage": {"capacity_wh": 2000, "max_charge_power_w": 1000}},
"inverters": {"inverter": {"battery_id": "storage", "max_power_w": 1000}},
},
}
)
run_time = to_datetime("2026-08-01T14:40:00+00:00")
get_ems().set_start_datetime(run_time)
measurement = isolated_measurement
# SoC through the measurement API at 13:30, the adapter's meter reading at the
# slot start (14:00); the run at 14:40 must still find the SoC.
await measurement.update_value(run_time.subtract(minutes=70), "storage-soc-factor", 0.42)
await measurement.update_value(get_ems().start_datetime, "load_meter", 1234.5)
tariff = pd.Series(
0.00008, index=pd.date_range("2026-08-01T00:00:00Z", periods=72 * 4, freq="15min")
)
async def read_array(**kwargs):
return np.full(48, 0.0003)
prediction = Mock(
update_data=AsyncMock(),
key_to_array=AsyncMock(side_effect=read_array),
key_to_raw_series=AsyncMock(return_value=tariff),
)
with patch.object(Genetic0OptimizationParameters, "prediction", prediction):
parameters = await Genetic0OptimizationParameters.prepare()
assert parameters is not None
assert parameters.pv_battery is not None
assert parameters.pv_battery.initial_soc_percentage == 42
@pytest.mark.asyncio
async def test_older_missing_value_does_not_spoil_current_soc(config_eos, isolated_measurement):
config_eos.merge_settings_from_dict(
{
"prediction": {"hours": 48},
"optimization": {"genetic0": {"horizon_hours": 24}},
"elecfee": {"provider": None},
"devices": {
"max_batteries": 1,
"max_electric_vehicles": 0,
"max_inverters": 1,
"max_home_appliances": 0,
"batteries": {"storage": {"capacity_wh": 2000, "max_charge_power_w": 1000}},
"inverters": {"inverter": {"battery_id": "storage", "max_power_w": 1000}},
},
}
)
run_time = to_datetime("2026-08-01T14:40:00+00:00")
get_ems().set_start_datetime(run_time)
measurement = isolated_measurement
# An older reading stored as pandas NA (e.g. a JSON null in a Float64 series).
await measurement.update_value(run_time.subtract(hours=1), "storage-soc-factor", pd.NA)
await measurement.update_value(run_time, "storage-soc-factor", 0.42)
tariff = pd.Series(
0.00008, index=pd.date_range("2026-08-01T00:00:00Z", periods=72 * 4, freq="15min")
)
async def read_array(**kwargs):
return np.full(48, 0.0003)
prediction = Mock(
update_data=AsyncMock(),
key_to_array=AsyncMock(side_effect=read_array),
key_to_raw_series=AsyncMock(return_value=tariff),
)
with patch.object(Genetic0OptimizationParameters, "prediction", prediction):
parameters = await Genetic0OptimizationParameters.prepare()
assert parameters is not None
assert parameters.pv_battery is not None
assert parameters.pv_battery.initial_soc_percentage == 42
tests/test_genetic0_tariff_coverage.py
-65
View File
@@ -1,65 +0,0 @@
"""GENETIC0 must not extend a published feed-in tariff beyond real coverage."""
from unittest.mock import AsyncMock, Mock, patch
import numpy as np
import pandas as pd
import pytest
from akkudoktoreos.core.coreabc import get_ems
from akkudoktoreos.optimization.genetic0.genetic0params import (
Genetic0OptimizationParameters,
)
from akkudoktoreos.utils.datetimeutil import to_datetime
@pytest.mark.asyncio
@pytest.mark.parametrize("published_hours", [20, 24])
async def test_dvhub_tariff_requires_real_control_horizon(
config_eos, published_hours, caplog
):
config_eos.merge_settings_from_dict(
{
"prediction": {"hours": 48},
"optimization": {"genetic0": {"horizon_hours": 24}},
"feedintariff": {"provider": "FeedInTariffDvhubOnline"},
"elecfee": {"provider": None},
"devices": {
"max_batteries": 0,
"max_electric_vehicles": 0,
"max_inverters": 0,
"max_home_appliances": 0,
},
}
)
start = to_datetime("2026-08-01T00:00:00+00:00")
get_ems().set_start_datetime(start)
prices = pd.Series(
np.arange(published_hours * 4, dtype=float) / 1_000_000,
index=pd.date_range(start, periods=published_hours * 4, freq="15min"),
)
async def read_array(**kwargs):
return np.full(48, 1.0)
prediction = Mock(
update_data=AsyncMock(),
key_to_array=AsyncMock(side_effect=read_array),
key_to_raw_series=AsyncMock(return_value=prices),
)
with patch.object(Genetic0OptimizationParameters, "prediction", prediction):
parameters = await Genetic0OptimizationParameters.prepare()
prediction.update_data.assert_awaited_once()
assert config_eos.feedintariff.provider == "FeedInTariffDvhubOnline"
if published_hours < 24:
assert parameters is None
assert "Missing feed-in tariff within the GENETIC0 control horizon" in caplog.text
else:
assert parameters is not None
tariffs = parameters.ems.feed_in_tariff_per_wh
assert isinstance(tariffs, list)
assert tariffs[:24] == pytest.approx(
[(4 * hour + 1.5) / 1_000_000 for hour in range(24)]
)
assert np.isnan(tariffs[24:]).all()
tests/test_genetic0params.py
+279
View File
@@ -0,0 +1,279 @@
"""GENETIC0 parameter preparation: time alignment, tariff coverage, and measured SoC."""
from unittest.mock import AsyncMock, Mock, patch
import numpy as np
import pandas as pd
import pendulum
import pytest
import pytest_asyncio
from akkudoktoreos.core.coreabc import get_ems, get_measurement
from akkudoktoreos.optimization.genetic0.genetic0params import (
Genetic0OptimizationParameters,
)
from akkudoktoreos.utils.datetimeutil import to_datetime
class TestGenetic0ParameterTimeAlignment:
@pytest.mark.asyncio
@pytest.mark.parametrize(
"start",
[
# Normal day: 23 elapsed hours since local midnight.
pendulum.datetime(2026, 10, 24, 23, tz="Europe/Berlin"),
# Fall back (CEST -> CET): 24 elapsed hours since local midnight.
pendulum.datetime(2026, 10, 25, 23, tz="Europe/Berlin"),
# Spring forward (CET -> CEST): 22 elapsed hours since local midnight.
pendulum.datetime(2026, 3, 29, 23, tz="Europe/Berlin"),
],
ids=["normal", "fall-back", "spring-forward"],
)
async def test_parameter_arrays_start_hour_aligned_on_dst_days(self, config_eos, start):
config_eos.merge_settings_from_dict(
{
"general": {"timezone_override": "Europe/Berlin"},
"prediction": {"hours": 48},
"optimization": {"genetic0": {"horizon_hours": 24}},
"feedintariff": {"provider": "FeedInTariffDvhubOnline"},
"elecfee": {"provider": None},
"devices": {
"max_batteries": 0,
"max_electric_vehicles": 0,
"max_inverters": 0,
"max_home_appliances": 0,
},
}
)
get_ems().set_start_datetime(start)
ems_start = get_ems().start_datetime
assert ems_start.hour == 23
# The tariff is only published from the start slot on, as a live
# provider would do late in the evening.
prices = pd.Series(
np.full(24 * 4, 0.0001),
index=pd.date_range(ems_start, periods=24 * 4, freq="15min"),
)
array_starts = []
async def read_array(**kwargs):
array_starts.append(kwargs["start_datetime"])
return np.full(48, 1.0)
prediction = Mock(
update_data=AsyncMock(),
key_to_array=AsyncMock(side_effect=read_array),
key_to_raw_series=AsyncMock(return_value=prices),
)
with patch.object(Genetic0OptimizationParameters, "prediction", prediction):
parameters = await Genetic0OptimizationParameters.prepare()
assert parameters is not None
# Index ``start.hour`` of every array is the start slot.
assert array_starts
for array_start in array_starts:
assert array_start.add(hours=ems_start.hour) == ems_start
tariffs = parameters.ems.feed_in_tariff_per_wh
assert isinstance(tariffs, list)
assert np.isfinite(tariffs[ems_start.hour : ems_start.hour + 24]).all()
class TestGenetic0TariffCoverage:
"""Tariff arrays indexed from UTC midnight, independent of the host timezone."""
@pytest.mark.asyncio
@pytest.mark.parametrize("published_hours", [20, 24])
async def test_dvhub_tariff_requires_real_control_horizon(self,
config_eos, published_hours, caplog
):
config_eos.merge_settings_from_dict(
{
"general": {"timezone_override": "UTC"},
"prediction": {"hours": 48},
"optimization": {"genetic0": {"horizon_hours": 24}},
"feedintariff": {"provider": "FeedInTariffDvhubOnline"},
"elecfee": {"provider": None},
"devices": {
"max_batteries": 0,
"max_electric_vehicles": 0,
"max_inverters": 0,
"max_home_appliances": 0,
},
}
)
start = to_datetime("2026-08-01T00:00:00+00:00", in_timezone="UTC")
get_ems().set_start_datetime(start)
prices = pd.Series(
np.arange(published_hours * 4, dtype=float) / 1_000_000,
index=pd.date_range(start, periods=published_hours * 4, freq="15min"),
)
async def read_array(**kwargs):
return np.full(48, 1.0)
prediction = Mock(
update_data=AsyncMock(),
key_to_array=AsyncMock(side_effect=read_array),
key_to_raw_series=AsyncMock(return_value=prices),
)
with patch.object(Genetic0OptimizationParameters, "prediction", prediction):
parameters = await Genetic0OptimizationParameters.prepare()
prediction.update_data.assert_awaited_once()
assert config_eos.feedintariff.provider == "FeedInTariffDvhubOnline"
if published_hours < 24:
assert parameters is None
assert "Missing feed-in tariff within the GENETIC0 control horizon" in caplog.text
else:
assert parameters is not None
tariffs = parameters.ems.feed_in_tariff_per_wh
assert isinstance(tariffs, list)
assert tariffs[:24] == pytest.approx(
[(4 * hour + 1.5) / 1_000_000 for hour in range(24)]
)
assert np.isnan(tariffs[24:]).all()
class TestGenetic0MeasuredSOC:
@pytest_asyncio.fixture
async def isolated_measurement(self, config_eos):
"""Keep synthetic records out of the process-wide measurement singleton."""
measurement = get_measurement()
await measurement.delete_by_datetime(None, None)
try:
yield measurement
finally:
await measurement.delete_by_datetime(None, None)
@pytest.mark.asyncio
async def test_soc_read_late_in_slot_beats_previous_reading(self, config_eos, isolated_measurement):
config_eos.merge_settings_from_dict(
{
"prediction": {"hours": 48},
"optimization": {"genetic0": {"horizon_hours": 24}},
"elecfee": {"provider": None},
"devices": {
"max_batteries": 1,
"max_electric_vehicles": 0,
"max_inverters": 1,
"max_home_appliances": 0,
"batteries": {"storage": {"capacity_wh": 2000, "max_charge_power_w": 1000}},
"inverters": {"inverter": {"battery_id": "storage", "max_power_w": 1000}},
},
}
)
# A run at 14:40 belongs to the 14:00 slot. The previous run (13:40) is closer
# to the slot start than the reading taken now.
run_time = to_datetime("2026-08-01T14:40:00+00:00")
get_ems().set_start_datetime(run_time)
measurement = isolated_measurement
await measurement.update_value(run_time.subtract(hours=1), "storage-soc-factor", 0.10)
await measurement.update_value(run_time, "storage-soc-factor", 0.42)
tariff = pd.Series(
0.00008, index=pd.date_range("2026-08-01T00:00:00Z", periods=72 * 4, freq="15min")
)
async def read_array(**kwargs):
return np.full(48, 0.0003)
prediction = Mock(
update_data=AsyncMock(),
key_to_array=AsyncMock(side_effect=read_array),
key_to_raw_series=AsyncMock(return_value=tariff),
)
with patch.object(Genetic0OptimizationParameters, "prediction", prediction):
parameters = await Genetic0OptimizationParameters.prepare()
assert parameters is not None
assert parameters.pv_battery is not None
assert parameters.pv_battery.initial_soc_percentage == 42
@pytest.mark.asyncio
async def test_meter_only_record_does_not_mask_soc(self, config_eos, isolated_measurement):
config_eos.merge_settings_from_dict(
{
"prediction": {"hours": 48},
"optimization": {"genetic0": {"horizon_hours": 24}},
"elecfee": {"provider": None},
"measurement": {"load_emr_keys": ["load_meter"]},
"devices": {
"max_batteries": 1,
"max_electric_vehicles": 0,
"max_inverters": 1,
"max_home_appliances": 0,
"batteries": {"storage": {"capacity_wh": 2000, "max_charge_power_w": 1000}},
"inverters": {"inverter": {"battery_id": "storage", "max_power_w": 1000}},
},
}
)
run_time = to_datetime("2026-08-01T14:40:00+00:00")
get_ems().set_start_datetime(run_time)
measurement = isolated_measurement
# SoC through the measurement API at 13:30, the adapter's meter reading at the
# slot start (14:00); the run at 14:40 must still find the SoC.
await measurement.update_value(run_time.subtract(minutes=70), "storage-soc-factor", 0.42)
await measurement.update_value(get_ems().start_datetime, "load_meter", 1234.5)
tariff = pd.Series(
0.00008, index=pd.date_range("2026-08-01T00:00:00Z", periods=72 * 4, freq="15min")
)
async def read_array(**kwargs):
return np.full(48, 0.0003)
prediction = Mock(
update_data=AsyncMock(),
key_to_array=AsyncMock(side_effect=read_array),
key_to_raw_series=AsyncMock(return_value=tariff),
)
with patch.object(Genetic0OptimizationParameters, "prediction", prediction):
parameters = await Genetic0OptimizationParameters.prepare()
assert parameters is not None
assert parameters.pv_battery is not None
assert parameters.pv_battery.initial_soc_percentage == 42
@pytest.mark.asyncio
async def test_older_missing_value_does_not_spoil_current_soc(self, config_eos, isolated_measurement):
config_eos.merge_settings_from_dict(
{
"prediction": {"hours": 48},
"optimization": {"genetic0": {"horizon_hours": 24}},
"elecfee": {"provider": None},
"devices": {
"max_batteries": 1,
"max_electric_vehicles": 0,
"max_inverters": 1,
"max_home_appliances": 0,
"batteries": {"storage": {"capacity_wh": 2000, "max_charge_power_w": 1000}},
"inverters": {"inverter": {"battery_id": "storage", "max_power_w": 1000}},
},
}
)
run_time = to_datetime("2026-08-01T14:40:00+00:00")
get_ems().set_start_datetime(run_time)
measurement = isolated_measurement
# An older reading stored as pandas NA (e.g. a JSON null in a Float64 series).
await measurement.update_value(run_time.subtract(hours=1), "storage-soc-factor", pd.NA)
await measurement.update_value(run_time, "storage-soc-factor", 0.42)
tariff = pd.Series(
0.00008, index=pd.date_range("2026-08-01T00:00:00Z", periods=72 * 4, freq="15min")
)
async def read_array(**kwargs):
return np.full(48, 0.0003)
prediction = Mock(
update_data=AsyncMock(),
key_to_array=AsyncMock(side_effect=read_array),
key_to_raw_series=AsyncMock(return_value=tariff),
)
with patch.object(Genetic0OptimizationParameters, "prediction", prediction):
parameters = await Genetic0OptimizationParameters.prepare()
assert parameters is not None
assert parameters.pv_battery is not None
assert parameters.pv_battery.initial_soc_percentage == 42
tests/test_genetic0params_dst.py
-76
View File
@@ -1,76 +0,0 @@
"""GENETIC0 parameter arrays must stay aligned with the start hour on DST days."""
from unittest.mock import AsyncMock, Mock, patch
import numpy as np
import pandas as pd
import pendulum
import pytest
from akkudoktoreos.core.coreabc import get_ems
from akkudoktoreos.optimization.genetic0.genetic0params import (
Genetic0OptimizationParameters,
)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"start",
[
# Normal day: 23 elapsed hours since local midnight.
pendulum.datetime(2026, 10, 24, 23, tz="Europe/Berlin"),
# Fall back (CEST -> CET): 24 elapsed hours since local midnight.
pendulum.datetime(2026, 10, 25, 23, tz="Europe/Berlin"),
# Spring forward (CET -> CEST): 22 elapsed hours since local midnight.
pendulum.datetime(2026, 3, 29, 23, tz="Europe/Berlin"),
],
ids=["normal", "fall-back", "spring-forward"],
)
async def test_parameter_arrays_start_hour_aligned_on_dst_days(config_eos, start):
config_eos.merge_settings_from_dict(
{
"general": {"timezone": "Europe/Berlin"},
"prediction": {"hours": 48},
"optimization": {"genetic0": {"horizon_hours": 24}},
"feedintariff": {"provider": "FeedInTariffDvhubOnline"},
"elecfee": {"provider": None},
"devices": {
"max_batteries": 0,
"max_electric_vehicles": 0,
"max_inverters": 0,
"max_home_appliances": 0,
},
}
)
get_ems().set_start_datetime(start)
ems_start = get_ems().start_datetime
assert ems_start.hour == 23
# The tariff is only published from the start slot on, as a live
# provider would do late in the evening.
prices = pd.Series(
np.full(24 * 4, 0.0001),
index=pd.date_range(ems_start, periods=24 * 4, freq="15min"),
)
array_starts = []
async def read_array(**kwargs):
array_starts.append(kwargs["start_datetime"])
return np.full(48, 1.0)
prediction = Mock(
update_data=AsyncMock(),
key_to_array=AsyncMock(side_effect=read_array),
key_to_raw_series=AsyncMock(return_value=prices),
)
with patch.object(Genetic0OptimizationParameters, "prediction", prediction):
parameters = await Genetic0OptimizationParameters.prepare()
assert parameters is not None
# Index ``start.hour`` of every array is the start slot.
assert array_starts
for array_start in array_starts:
assert array_start.add(hours=ems_start.hour) == ems_start
tariffs = parameters.ems.feed_in_tariff_per_wh
assert isinstance(tariffs, list)
assert np.isfinite(tariffs[ems_start.hour : ems_start.hour + 24]).all()
tests/test_genetic_seeding.py
+127 -1
View File
@@ -14,7 +14,11 @@ from akkudoktoreos.optimization.genetic.genetic import (
_release_freed_memory,
_unpack_genes,
)
from akkudoktoreos.utils.datetimeutil import to_datetime
from akkudoktoreos.optimization.genetic.geneticparams import (
GeneticEnergyManagementParameters,
GeneticOptimizationParameters,
)
from akkudoktoreos.utils.datetimeutil import DateTime, compare_datetimes, to_datetime
def _configure_hourly_grid(config_eos: ConfigEOS, *, start_hour: int = 0) -> None:
@@ -527,3 +531,125 @@ def test_offspring_do_not_inherit_immigrant_protection(config_eos: ConfigEOS):
]
offspring = opt._make_offspring(parents, 8, mutation_probability=1.0)
assert all(getattr(child, "immigrant_protection", 0) == 0 for child in offspring)
def _warm_start_optimizer(
config_eos: ConfigEOS,
*,
interval: int,
optimize_ev: bool = False,
n_appliance_genes: int = 0,
) -> tuple[GeneticOptimization, DateTime]:
config_eos.merge_settings_from_dict(
{
"prediction": {"hours": 48},
"optimization": {
"genetic": {"tail_horizon_hours": 0, "horizon_hours": 24, "interval_sec": interval}
},
}
)
slot0 = get_ems(init=True).set_start_datetime(to_datetime().set(hour=8, minute=0, second=0))
opt = GeneticOptimization(fixed_seed=42)
opt.optimize_ev = optimize_ev
opt.appliance_layout = SimpleNamespace(n_genes=n_appliance_genes, genes=[]) # type: ignore[assignment]
opt._slot0_datetime = slot0
return opt, slot0
def _warm_start_genes(start: int, stop: int) -> list[float]:
return [float(value) for value in range(start, stop)]
class TestWarmStartAlignment:
def test_quarter_hour_warm_start_moves_one_slot_forward(self, config_eos: ConfigEOS):
# 07:45 run: export genes (32) at 08:00 and 08:15. Unshifted, the 08:00 run
# would read them as 08:15 and 08:30.
opt, slot0 = _warm_start_optimizer(config_eos, interval=900)
previous = [19.0, 32.0, 32.0, 14.0] + [36.0] * (opt.control_slots - 4)
aligned = opt._start_solution_for_run_start(previous, slot0.subtract(minutes=15))
assert aligned == [32, 32, 14] + [36] * (opt.control_slots - 3)
def test_elapsed_slots_are_dropped_per_block_and_appliance_genes_kept(self, config_eos: ConfigEOS):
opt, slot0 = _warm_start_optimizer(config_eos, interval=3600, optimize_ev=True, n_appliance_genes=1)
slots = opt.control_slots
battery = _warm_start_genes(0, slots)
ev = _warm_start_genes(100, 100 + slots)
appliance = [3.0]
aligned = opt._start_solution_for_run_start(battery + ev + appliance, slot0.subtract(hours=3))
assert aligned == (
_warm_start_genes(3, slots)
+ [slots - 1.0] * 3
+ _warm_start_genes(103, 100 + slots)
+ [100 + slots - 1.0] * 3
+ appliance
)
def test_same_slot_or_unknown_start_keeps_warm_start(self, config_eos: ConfigEOS):
opt, slot0 = _warm_start_optimizer(config_eos, interval=900)
previous = _warm_start_genes(0, opt.control_slots)
assert opt._start_solution_for_run_start(previous, slot0) == previous
assert opt._start_solution_for_run_start(previous, None) == previous
assert opt._start_solution_for_run_start(None, slot0) is None
@pytest.mark.parametrize(
"offset_minutes",
[
pytest.param(-24 * 60, id="all-control-slots-elapsed"),
pytest.param(15, id="starts-after-this-run"),
],
)
def test_unusable_warm_start_is_dropped(self, config_eos: ConfigEOS, offset_minutes: int):
opt, slot0 = _warm_start_optimizer(config_eos, interval=900)
previous = _warm_start_genes(0, opt.control_slots)
assert opt._start_solution_for_run_start(previous, slot0.add(minutes=offset_minutes)) is None
def test_start_datetime_falls_back_to_last_solution_of_this_server(self,
config_eos: ConfigEOS, monkeypatch: pytest.MonkeyPatch
):
opt, slot0 = _warm_start_optimizer(config_eos, interval=900)
last_start = slot0.subtract(minutes=15)
monkeypatch.setattr(
type(get_ems()),
"_genetic_solution",
SimpleNamespace(start_solution=[19.0, 32.0, 14.0], start_solution_datetime=last_start),
)
same = SimpleNamespace(start_solution=[19, 32, 14], start_solution_datetime=None)
other = SimpleNamespace(start_solution=[19, 32, 15], start_solution_datetime=None)
explicit_start = slot0.subtract(minutes=30)
explicit = SimpleNamespace(start_solution=[19, 32, 14], start_solution_datetime=explicit_start)
assert opt._resolve_start_solution_datetime(same) == last_start # type: ignore[arg-type]
assert opt._resolve_start_solution_datetime(other) is None # type: ignore[arg-type]
assert opt._resolve_start_solution_datetime(explicit) == explicit_start # type: ignore[arg-type]
def test_parameters_accept_iso_start_solution_datetime(self, config_eos: ConfigEOS):
parameters = GeneticOptimizationParameters.model_validate(
dict(
ems=GeneticEnergyManagementParameters.model_validate(
dict(
pv_prognose_wh=[0.0, 0.0],
strompreis_euro_pro_wh=[0.0, 0.0],
einspeiseverguetung_euro_pro_wh=0.0,
preis_euro_pro_wh_akku=0.0,
gesamtlast=[0.0, 0.0],
)
),
pv_akku=None,
inverter=None,
eauto=None,
start_solution=[1.0, 2.0],
start_solution_datetime="2026-09-14T07:45:00+02:00",
)
)
assert parameters.start_solution_datetime is not None
assert compare_datetimes(
parameters.start_solution_datetime, to_datetime("2026-09-14T07:45:00+02:00")
).equal
tests/test_genetic_warm_start_alignment.py
-146
View File
@@ -1,146 +0,0 @@
"""A warm start from an earlier run is aligned with the slot this run starts in.
Genomes are run-relative: gene 0 controls the slot the run starts in. Reusing
the previous solution unchanged after a slot boundary describes every decision
one slot too late, and a search that keeps the seed postpones a planned action
by one slot per run.
"""
from types import SimpleNamespace
import pytest
from akkudoktoreos.config.config import ConfigEOS
from akkudoktoreos.core.coreabc import get_ems
from akkudoktoreos.optimization.genetic.genetic import GeneticOptimization
from akkudoktoreos.optimization.genetic.geneticparams import (
GeneticEnergyManagementParameters,
GeneticOptimizationParameters,
)
from akkudoktoreos.utils.datetimeutil import DateTime, compare_datetimes, to_datetime
def _optimizer(
config_eos: ConfigEOS,
*,
interval: int,
optimize_ev: bool = False,
n_appliance_genes: int = 0,
) -> tuple[GeneticOptimization, DateTime]:
config_eos.merge_settings_from_dict(
{
"prediction": {"hours": 48},
"optimization": {
"genetic": {"tail_horizon_hours": 0, "horizon_hours": 24, "interval_sec": interval}
},
}
)
slot0 = get_ems(init=True).set_start_datetime(to_datetime().set(hour=8, minute=0, second=0))
opt = GeneticOptimization(fixed_seed=42)
opt.optimize_ev = optimize_ev
opt.appliance_layout = SimpleNamespace(n_genes=n_appliance_genes, genes=[]) # type: ignore[assignment]
opt._slot0_datetime = slot0
return opt, slot0
def _genes(start: int, stop: int) -> list[float]:
return [float(value) for value in range(start, stop)]
def test_quarter_hour_warm_start_moves_one_slot_forward(config_eos: ConfigEOS):
# 07:45 run: export genes (32) at 08:00 and 08:15. Unshifted, the 08:00 run
# would read them as 08:15 and 08:30.
opt, slot0 = _optimizer(config_eos, interval=900)
previous = [19.0, 32.0, 32.0, 14.0] + [36.0] * (opt.control_slots - 4)
aligned = opt._start_solution_for_run_start(previous, slot0.subtract(minutes=15))
assert aligned == [32, 32, 14] + [36] * (opt.control_slots - 3)
def test_elapsed_slots_are_dropped_per_block_and_appliance_genes_kept(config_eos: ConfigEOS):
opt, slot0 = _optimizer(config_eos, interval=3600, optimize_ev=True, n_appliance_genes=1)
slots = opt.control_slots
battery = _genes(0, slots)
ev = _genes(100, 100 + slots)
appliance = [3.0]
aligned = opt._start_solution_for_run_start(battery + ev + appliance, slot0.subtract(hours=3))
assert aligned == (
_genes(3, slots)
+ [slots - 1.0] * 3
+ _genes(103, 100 + slots)
+ [100 + slots - 1.0] * 3
+ appliance
)
def test_same_slot_or_unknown_start_keeps_warm_start(config_eos: ConfigEOS):
opt, slot0 = _optimizer(config_eos, interval=900)
previous = _genes(0, opt.control_slots)
assert opt._start_solution_for_run_start(previous, slot0) == previous
assert opt._start_solution_for_run_start(previous, None) == previous
assert opt._start_solution_for_run_start(None, slot0) is None
@pytest.mark.parametrize(
"offset_minutes",
[
pytest.param(-24 * 60, id="all-control-slots-elapsed"),
pytest.param(15, id="starts-after-this-run"),
],
)
def test_unusable_warm_start_is_dropped(config_eos: ConfigEOS, offset_minutes: int):
opt, slot0 = _optimizer(config_eos, interval=900)
previous = _genes(0, opt.control_slots)
assert opt._start_solution_for_run_start(previous, slot0.add(minutes=offset_minutes)) is None
def test_start_datetime_falls_back_to_last_solution_of_this_server(
config_eos: ConfigEOS, monkeypatch: pytest.MonkeyPatch
):
opt, slot0 = _optimizer(config_eos, interval=900)
last_start = slot0.subtract(minutes=15)
monkeypatch.setattr(
type(get_ems()),
"_genetic_solution",
SimpleNamespace(start_solution=[19.0, 32.0, 14.0], start_solution_datetime=last_start),
)
same = SimpleNamespace(start_solution=[19, 32, 14], start_solution_datetime=None)
other = SimpleNamespace(start_solution=[19, 32, 15], start_solution_datetime=None)
explicit_start = slot0.subtract(minutes=30)
explicit = SimpleNamespace(start_solution=[19, 32, 14], start_solution_datetime=explicit_start)
assert opt._resolve_start_solution_datetime(same) == last_start # type: ignore[arg-type]
assert opt._resolve_start_solution_datetime(other) is None # type: ignore[arg-type]
assert opt._resolve_start_solution_datetime(explicit) == explicit_start # type: ignore[arg-type]
def test_parameters_accept_iso_start_solution_datetime(config_eos: ConfigEOS):
parameters = GeneticOptimizationParameters.model_validate(
dict(
ems=GeneticEnergyManagementParameters.model_validate(
dict(
pv_prognose_wh=[0.0, 0.0],
strompreis_euro_pro_wh=[0.0, 0.0],
einspeiseverguetung_euro_pro_wh=0.0,
preis_euro_pro_wh_akku=0.0,
gesamtlast=[0.0, 0.0],
)
),
pv_akku=None,
inverter=None,
eauto=None,
start_solution=[1.0, 2.0],
start_solution_datetime="2026-09-14T07:45:00+02:00",
)
)
assert parameters.start_solution_datetime is not None
assert compare_datetimes(
parameters.start_solution_datetime, to_datetime("2026-09-14T07:45:00+02:00")
).equal
tests/test_geneticvisualize.py
+206
View File
@@ -1,10 +1,26 @@
from io import BytesIO
from pathlib import Path
from types import SimpleNamespace
import matplotlib.dates as mdates
import matplotlib.pyplot as plt
import numpy as np
import pendulum
import pytest
from matplotlib.testing.compare import compare_images
from pypdf import PdfReader
from akkudoktoreos.optimization.genetic import geneticvisualize as visualize
from akkudoktoreos.optimization.genetic.geneticsolution import GeneticSolution
from akkudoktoreos.optimization.genetic.geneticvisualize import (
genetic_generate_example_report,
)
from akkudoktoreos.optimization.genetic.terminalvalue import (
TailDiagnostics,
TailPlanSlot,
TerminalValueCurve,
TerminalValueResult,
)
filename = "example_report.pdf"
@@ -34,3 +50,193 @@ def test_generate_pdf_example(config_eos):
reference_png_file.unlink()
output_file.unlink()
output_png_file.unlink()
class TestGeneticVisualizationIntervals:
@pytest.fixture
def report_solution(self):
start = pendulum.datetime(2026, 10, 25, 1, 45, tz="Europe/Berlin")
energy = [100.0, 150.0, 200.0, 250.0]
result = SimpleNamespace(
load_wh_per_hour=energy,
home_appliance_wh_per_hour=energy,
grid_feed_in_wh_per_hour=energy,
grid_consumption_wh_per_hour=energy,
losses_per_hour=[1.0] * 4,
battery_soc_per_hour=[20.0, 25.0, 30.0, 35.0],
ev_soc_per_hour=[0.0] * 4,
costs_per_hour=[0.03] * 4,
revenue_per_hour=[0.01] * 4,
total_costs=0.12,
total_revenue=0.04,
total_balance=0.08,
home_appliance_energy_wh={"water-heater": energy},
)
return SimpleNamespace(
parameters=SimpleNamespace(
ems=SimpleNamespace(
total_load=energy,
pv_forecast_wh=energy,
feed_in_tariff_per_wh=[0.00007] * 4,
electricity_price_per_wh=[0.0003] * 4,
),
temperature_forecast=[12.0] * 4,
),
interval_seconds=900,
start_solution_datetime=start,
controls_start_at_now=True,
start_hour=1,
ac_charge=[0.0, 0.25, 0.5, 1.0],
dc_charge=[1.0] * 4,
discharge_allowed=[0, 0, 1, 1],
battery_grid_export_allowed=[0, 0, 0, 1],
battery_grid_export_factor=[0.0, 0.0, 0.0, 0.5],
result=result,
extra_data=None,
fitness_history=None,
fixed_seed=None,
appliance_starts={"water-heater": [start.add(minutes=15)]},
appliance_deadline_missed={"water-heater": True},
terminal_value=TerminalValueResult(
mode="TAIL", control_horizon_hours=1, requested_tail_hours=2,
effective_tail_hours=0.25, tail_end_hour=1.25,
battery_energy_wh=500, credited_euro=0.13,
tail_operating_euro=0.1, continuation_value_euro=0.03,
continuation_mode="AUTO", reason="forecast gap limits tail",
curve=TerminalValueCurve(energy_wh=[0, 1000], value_euro=[0, 0.26]),
tail_diagnostics=TailDiagnostics(slots=1, slot_hours=0.25),
tail_plan=[TailPlanSlot(
slot=0, hour_from_start=1, action="discharge",
soc_start_percentage=40, soc_end_percentage=35,
pv_wh=0, load_wh=100, grid_import_wh=0, grid_export_wh=0,
battery_charge_wh=0, battery_discharge_wh=100,
import_price_euro_per_kwh=0.3, feed_in_tariff_euro_per_kwh=0.07,
slot_value_euro=0.03, remaining_value_euro=0.1,
ac_charge_factor=0, dc_charge_allowed=1, discharge_allowed=1,
battery_grid_export_factor=0,
)],
),
)
@pytest.mark.parametrize("interval_seconds", [900, 3600])
def test_report_preserves_run_timing_energy_and_export_controls(self,
config_eos, monkeypatch, report_solution, interval_seconds
):
report_solution.interval_seconds = interval_seconds
captured = []
original = visualize.GeneticVisualizationReport.create_line_chart_date
def capture(self, start_date, y_list, **kwargs):
captured.append((start_date, y_list, kwargs, self.interval_seconds))
return original(self, start_date, y_list, **kwargs)
monkeypatch.setattr(visualize.GeneticVisualizationReport, "create_line_chart_date", capture)
monkeypatch.setattr(visualize, "get_ems", lambda: pytest.fail("Snapshot must own report timing"))
pdf = visualize.genetic_prepare_visualize(report_solution)
assert pdf.startswith(b"%PDF-")
assert all(row[0] == report_solution.start_solution_datetime for row in captured)
assert all(row[3] == interval_seconds for row in captured)
load = next(row for row in captured if row[2].get("title") == "Load Profile")
assert load[1] == [[100, 150, 200, 250]] # No repeated wall-clock trimming or Wh scaling.
controls = next(row for row in captured if row[2].get("title") == "Executable Battery Controls")
assert controls[1][-1] == [0, 0, 0, 0.5]
assert len(controls[1]) == 5
text = " ".join(page.extract_text() for page in PdfReader(BytesIO(pdf)).pages)
for expected in (
"Energy Flow per Interval", "water-heater", "Deadline missed: True",
"forecast gap limits tail", "effective tail: 0.25 h", "0.13 EUR",
"not executable controls", "not realized revenue", "Residual Battery Value Curve",
):
assert expected in text
assert "Energy Flow per Hour" not in text
def test_short_quarterhour_chart_has_elapsed_fractional_hours(self, config_eos):
report = visualize.GeneticVisualizationReport(interval_seconds=900)
start = pendulum.datetime(2026, 10, 25, 2, 45, tz="Europe/Berlin", fold=0)
report.create_line_chart_date(start, [[100.0, 200.0, 300.0, 400.0]], ylabel="Wh")
fig, axis = plt.subplots()
try:
report.current_group[0]()
dates = mdates.num2date(axis.lines[0].get_xdata())
assert np.diff([value.timestamp() for value in dates]).tolist() == [900] * 3
assert np.asarray(axis.lines[0].get_ydata()).tolist() == [100, 200, 300, 400]
assert [label.get_text() for label in fig.axes[1].get_xticklabels()] == ["0", "0.25", "0.5", "0.75"]
finally:
plt.close(fig)
def test_empty_and_mismatched_date_series(self, config_eos):
report = visualize.GeneticVisualizationReport()
start = pendulum.now("UTC")
report.create_line_chart_date(start, [[]], ylabel="Wh")
assert report.current_group == []
with pytest.raises(ValueError, match="same intervals"):
report.create_line_chart_date(start, [[1.0, 2.0], [1.0]], ylabel="Wh")
def test_existing_hourly_solution_renders_without_terminal_metadata(self, config_eos, monkeypatch):
payload = (Path(__file__).parent / "testdata/genetic/optimize_result_1.json").read_text()
solution = GeneticSolution.model_validate_json(payload)
assert not solution.controls_start_at_now
assert solution.terminal_value is None
start = pendulum.datetime(2026, 9, 16, solution.start_hour, tz="UTC")
monkeypatch.setattr(visualize, "get_ems", lambda: SimpleNamespace(start_datetime=start))
pdf = visualize.genetic_prepare_visualize(solution)
assert len(PdfReader(BytesIO(pdf)).pages) >= 4
def test_fixed_fallback_does_not_invent_tail_charts(self, config_eos, report_solution):
report_solution.terminal_value = TerminalValueResult(
mode="FIXED", reason="No complete tail forecast", credited_euro=0.05,
battery_energy_wh=500, requested_tail_hours=24,
)
report_solution.result.home_appliance_energy_wh = {}
pdf = visualize.genetic_prepare_visualize(report_solution)
text = " ".join(page.extract_text() for page in PdfReader(BytesIO(pdf)).pages)
assert "No complete tail forecast" in text
assert "effective tail: 0 h" in text
assert "Deadline missed: True" in text # Report unscheduled consumers even with no energy series.
assert "Residual Battery Value Curve" not in text
assert "Tail Lookahead:" not in text
def test_tail_chart_starts_after_control_horizon(self, config_eos, report_solution):
report = visualize.GeneticVisualizationReport(interval_seconds=900)
start = report_solution.start_solution_datetime
visualize._add_solution_diagnostics(report, report_solution, start)
fig, axis = plt.subplots()
try:
report.groups[-1][0]()
expected = mdates.date2num(start.add(hours=1))
assert np.asarray(axis.lines[0].get_xdata()).tolist() == [expected]
assert np.asarray(axis.lines[3].get_ydata()).tolist() == [100]
assert all(line.get_marker() == "o" for line in axis.lines)
left, right = axis.get_xlim()
assert left < expected < right
assert (right - left) * 86400 == pytest.approx(900)
assert "Not Executable" in axis.get_title()
assert report_solution.discharge_allowed == [0, 0, 1, 1]
finally:
plt.close(fig)
@pytest.mark.parametrize("tariff", [0.00007, [0.00007] * 6])
def test_pdf_clips_forecast_tail_and_accepts_historical_diagnostics(self,
config_eos, monkeypatch, report_solution, tariff
):
report_solution.parameters.ems.total_load = [100.0] * 6
report_solution.parameters.ems.pv_forecast_wh = [200.0] * 6
report_solution.parameters.ems.feed_in_tariff_per_wh = tariff
report_solution.extra_data = {
"verluste": [1.0, 2.0, 3.0], "bilanz": [0.1, 0.2, 0.3],
"nebenbedingung": [0.0, 0.0, 0.0],
}
captured = []
original = visualize.GeneticVisualizationReport.create_line_chart_date
def capture(self, start_date, y_list, **kwargs):
captured.append((y_list, kwargs))
return original(self, start_date, y_list, **kwargs)
monkeypatch.setattr(visualize.GeneticVisualizationReport, "create_line_chart_date", capture)
pdf = visualize.genetic_prepare_visualize(report_solution)
assert pdf.startswith(b"%PDF-")
tariff_plot = next(series for series, kwargs in captured if kwargs.get("title") == "Remuneration")
assert np.asarray(tariff_plot[0]).tolist() == [0.00007] * 4
assert all(len(series[0]) == 4 for series, _ in captured)
assert "verluste" in report_solution.extra_data # Rendering must not mutate the retained run.
tests/test_geneticvisualize_intervals.py
-217
View File
@@ -1,217 +0,0 @@
"""Semantic report tests with synthetic data; no forecast provider or live EMS run."""
from io import BytesIO
from pathlib import Path
from types import SimpleNamespace
import matplotlib.dates as mdates
import matplotlib.pyplot as plt
import numpy as np
import pendulum
import pytest
from pypdf import PdfReader
from akkudoktoreos.optimization.genetic import geneticvisualize as visualize
from akkudoktoreos.optimization.genetic.geneticsolution import GeneticSolution
from akkudoktoreos.optimization.genetic.terminalvalue import (
TailDiagnostics,
TailPlanSlot,
TerminalValueCurve,
TerminalValueResult,
)
@pytest.fixture
def report_solution():
start = pendulum.datetime(2026, 10, 25, 1, 45, tz="Europe/Berlin")
energy = [100.0, 150.0, 200.0, 250.0]
result = SimpleNamespace(
load_wh_per_hour=energy,
home_appliance_wh_per_hour=energy,
grid_feed_in_wh_per_hour=energy,
grid_consumption_wh_per_hour=energy,
losses_per_hour=[1.0] * 4,
battery_soc_per_hour=[20.0, 25.0, 30.0, 35.0],
ev_soc_per_hour=[0.0] * 4,
costs_per_hour=[0.03] * 4,
revenue_per_hour=[0.01] * 4,
total_costs=0.12,
total_revenue=0.04,
total_balance=0.08,
home_appliance_energy_wh={"water-heater": energy},
)
return SimpleNamespace(
parameters=SimpleNamespace(
ems=SimpleNamespace(
total_load=energy,
pv_forecast_wh=energy,
feed_in_tariff_per_wh=[0.00007] * 4,
electricity_price_per_wh=[0.0003] * 4,
),
temperature_forecast=[12.0] * 4,
),
interval_seconds=900,
start_solution_datetime=start,
controls_start_at_now=True,
start_hour=1,
ac_charge=[0.0, 0.25, 0.5, 1.0],
dc_charge=[1.0] * 4,
discharge_allowed=[0, 0, 1, 1],
battery_grid_export_allowed=[0, 0, 0, 1],
battery_grid_export_factor=[0.0, 0.0, 0.0, 0.5],
result=result,
extra_data=None,
fitness_history=None,
fixed_seed=None,
appliance_starts={"water-heater": [start.add(minutes=15)]},
appliance_deadline_missed={"water-heater": True},
terminal_value=TerminalValueResult(
mode="TAIL", control_horizon_hours=1, requested_tail_hours=2,
effective_tail_hours=0.25, tail_end_hour=1.25,
battery_energy_wh=500, credited_euro=0.13,
tail_operating_euro=0.1, continuation_value_euro=0.03,
continuation_mode="AUTO", reason="forecast gap limits tail",
curve=TerminalValueCurve(energy_wh=[0, 1000], value_euro=[0, 0.26]),
tail_diagnostics=TailDiagnostics(slots=1, slot_hours=0.25),
tail_plan=[TailPlanSlot(
slot=0, hour_from_start=1, action="discharge",
soc_start_percentage=40, soc_end_percentage=35,
pv_wh=0, load_wh=100, grid_import_wh=0, grid_export_wh=0,
battery_charge_wh=0, battery_discharge_wh=100,
import_price_euro_per_kwh=0.3, feed_in_tariff_euro_per_kwh=0.07,
slot_value_euro=0.03, remaining_value_euro=0.1,
ac_charge_factor=0, dc_charge_allowed=1, discharge_allowed=1,
battery_grid_export_factor=0,
)],
),
)
@pytest.mark.parametrize("interval_seconds", [900, 3600])
def test_report_preserves_run_timing_energy_and_export_controls(
config_eos, monkeypatch, report_solution, interval_seconds
):
report_solution.interval_seconds = interval_seconds
captured = []
original = visualize.GeneticVisualizationReport.create_line_chart_date
def capture(self, start_date, y_list, **kwargs):
captured.append((start_date, y_list, kwargs, self.interval_seconds))
return original(self, start_date, y_list, **kwargs)
monkeypatch.setattr(visualize.GeneticVisualizationReport, "create_line_chart_date", capture)
monkeypatch.setattr(visualize, "get_ems", lambda: pytest.fail("Snapshot must own report timing"))
pdf = visualize.genetic_prepare_visualize(report_solution)
assert pdf.startswith(b"%PDF-")
assert all(row[0] == report_solution.start_solution_datetime for row in captured)
assert all(row[3] == interval_seconds for row in captured)
load = next(row for row in captured if row[2].get("title") == "Load Profile")
assert load[1] == [[100, 150, 200, 250]] # No repeated wall-clock trimming or Wh scaling.
controls = next(row for row in captured if row[2].get("title") == "Executable Battery Controls")
assert controls[1][-1] == [0, 0, 0, 0.5]
assert len(controls[1]) == 5
text = " ".join(page.extract_text() for page in PdfReader(BytesIO(pdf)).pages)
for expected in (
"Energy Flow per Interval", "water-heater", "Deadline missed: True",
"forecast gap limits tail", "effective tail: 0.25 h", "0.13 EUR",
"not executable controls", "not realized revenue", "Residual Battery Value Curve",
):
assert expected in text
assert "Energy Flow per Hour" not in text
def test_short_quarterhour_chart_has_elapsed_fractional_hours(config_eos):
report = visualize.GeneticVisualizationReport(interval_seconds=900)
start = pendulum.datetime(2026, 10, 25, 2, 45, tz="Europe/Berlin", fold=0)
report.create_line_chart_date(start, [[100.0, 200.0, 300.0, 400.0]], ylabel="Wh")
fig, axis = plt.subplots()
try:
report.current_group[0]()
dates = mdates.num2date(axis.lines[0].get_xdata())
assert np.diff([value.timestamp() for value in dates]).tolist() == [900] * 3
assert np.asarray(axis.lines[0].get_ydata()).tolist() == [100, 200, 300, 400]
assert [label.get_text() for label in fig.axes[1].get_xticklabels()] == ["0", "0.25", "0.5", "0.75"]
finally:
plt.close(fig)
def test_empty_and_mismatched_date_series(config_eos):
report = visualize.GeneticVisualizationReport()
start = pendulum.now("UTC")
report.create_line_chart_date(start, [[]], ylabel="Wh")
assert report.current_group == []
with pytest.raises(ValueError, match="same intervals"):
report.create_line_chart_date(start, [[1.0, 2.0], [1.0]], ylabel="Wh")
def test_existing_hourly_solution_renders_without_terminal_metadata(config_eos, monkeypatch):
payload = (Path(__file__).parent / "testdata/genetic/optimize_result_1.json").read_text()
solution = GeneticSolution.model_validate_json(payload)
assert not solution.controls_start_at_now
assert solution.terminal_value is None
start = pendulum.datetime(2026, 9, 16, solution.start_hour, tz="UTC")
monkeypatch.setattr(visualize, "get_ems", lambda: SimpleNamespace(start_datetime=start))
pdf = visualize.genetic_prepare_visualize(solution)
assert len(PdfReader(BytesIO(pdf)).pages) >= 4
def test_fixed_fallback_does_not_invent_tail_charts(config_eos, report_solution):
report_solution.terminal_value = TerminalValueResult(
mode="FIXED", reason="No complete tail forecast", credited_euro=0.05,
battery_energy_wh=500, requested_tail_hours=24,
)
report_solution.result.home_appliance_energy_wh = {}
pdf = visualize.genetic_prepare_visualize(report_solution)
text = " ".join(page.extract_text() for page in PdfReader(BytesIO(pdf)).pages)
assert "No complete tail forecast" in text
assert "effective tail: 0 h" in text
assert "Deadline missed: True" in text # Report unscheduled consumers even with no energy series.
assert "Residual Battery Value Curve" not in text
assert "Tail Lookahead:" not in text
def test_tail_chart_starts_after_control_horizon(config_eos, report_solution):
report = visualize.GeneticVisualizationReport(interval_seconds=900)
start = report_solution.start_solution_datetime
visualize._add_solution_diagnostics(report, report_solution, start)
fig, axis = plt.subplots()
try:
report.groups[-1][0]()
expected = mdates.date2num(start.add(hours=1))
assert np.asarray(axis.lines[0].get_xdata()).tolist() == [expected]
assert np.asarray(axis.lines[3].get_ydata()).tolist() == [100]
assert all(line.get_marker() == "o" for line in axis.lines)
left, right = axis.get_xlim()
assert left < expected < right
assert (right - left) * 86400 == pytest.approx(900)
assert "Not Executable" in axis.get_title()
assert report_solution.discharge_allowed == [0, 0, 1, 1]
finally:
plt.close(fig)
@pytest.mark.parametrize("tariff", [0.00007, [0.00007] * 6])
def test_pdf_clips_forecast_tail_and_accepts_historical_diagnostics(
config_eos, monkeypatch, report_solution, tariff
):
report_solution.parameters.ems.total_load = [100.0] * 6
report_solution.parameters.ems.pv_forecast_wh = [200.0] * 6
report_solution.parameters.ems.feed_in_tariff_per_wh = tariff
report_solution.extra_data = {
"verluste": [1.0, 2.0, 3.0], "bilanz": [0.1, 0.2, 0.3],
"nebenbedingung": [0.0, 0.0, 0.0],
}
captured = []
original = visualize.GeneticVisualizationReport.create_line_chart_date
def capture(self, start_date, y_list, **kwargs):
captured.append((y_list, kwargs))
return original(self, start_date, y_list, **kwargs)
monkeypatch.setattr(visualize.GeneticVisualizationReport, "create_line_chart_date", capture)
pdf = visualize.genetic_prepare_visualize(report_solution)
assert pdf.startswith(b"%PDF-")
tariff_plot = next(series for series, kwargs in captured if kwargs.get("title") == "Remuneration")
assert np.asarray(tariff_plot[0]).tolist() == [0.00007] * 4
assert all(len(series[0]) == 4 for series, _ in captured)
assert "verluste" in report_solution.extra_data # Rendering must not mutate the retained run.
tests/test_inverter.py
+11
View File
@@ -8,6 +8,7 @@ from akkudoktoreos.devices.genetic.battery import (
SolarPanelBatteryParameters,
)
from akkudoktoreos.devices.genetic.inverter import Inverter, InverterParameters
from akkudoktoreos.devices.settings.invertersettings import InverterCommonSettings
@pytest.fixture
@@ -487,3 +488,13 @@ def test_process_energy_zero_generation_full_battery_high_consumption(inverter,
mock_battery.charge_energy.assert_not_called()
mock_battery.discharge_energy.assert_called_once_with(500.0, hour)
inverter.self_consumption_predictor.calculate_expected_direct_consumption.assert_not_called()
class TestInverterSettings:
@pytest.mark.parametrize("converter", ["to_genetic_param", "to_genetic0_param"])
def test_inverter_conversion_requires_output_limit(self, converter):
settings = InverterCommonSettings(device_id="inverter")
with pytest.raises(ValueError, match="max_power_w"):
getattr(settings, converter)()
settings.max_power_w = 4200
assert getattr(settings, converter)().max_power_wh == 4200
tests/test_measurement.py
+166 -52
View File
@@ -1,28 +1,29 @@
import json
from unittest.mock import AsyncMock
import numpy as np
import pytest
import pytest_asyncio
from pendulum import datetime, duration
from pydantic import ValidationError
from akkudoktoreos.config.config import SettingsEOS
from akkudoktoreos.core.coreabc import get_measurement
from akkudoktoreos.core.dataabc import DataSequence
from akkudoktoreos.measurement.measurement import (
MeasurementChannelSettings,
MeasurementCommonSettings,
MeasurementDataRecord,
)
from akkudoktoreos.utils.datetimeutil import (
DateTime,
Duration,
compare_datetimes,
to_datetime,
to_duration,
)
class TestMeasurementDataRecord:
"""Test suite for the MeasurementDataRecord class.
"""Measurement-specific keys, values, and serialization.
Ensuring that both dictionary-like and attribute-style access work correctly for fields and
configured measurements.
Generic native-field access is covered by TestDataRecord in test_dataabcrecord.py.
"""
@pytest.fixture
@@ -67,11 +68,6 @@ class TestMeasurementDataRecord:
with pytest.raises(KeyError):
_ = record["nonexistent"]
def test_setitem_field(self, record):
"""Test setting a native field using __setitem__."""
record["date_time"] = "2025-01-01T12:00:00+00:00"
assert str(record.date_time).startswith("2025-01-01")
def test_setitem_measurement(self, record):
"""Test setting a known measurement key using __setitem__."""
record["temp"] = 25.5
@@ -82,12 +78,6 @@ class TestMeasurementDataRecord:
with pytest.raises(KeyError):
record["unknown_key"] = 123
def test_delitem_field(self, record):
"""Test deleting a native field using __delitem__."""
record["date_time"] = "2025-01-01T12:00:00+00:00"
del record["date_time"]
assert record.date_time is None
def test_delitem_measurement(self, record):
"""Test deleting a known measurement key using __delitem__."""
del record["solar_power"]
@@ -98,11 +88,6 @@ class TestMeasurementDataRecord:
with pytest.raises(KeyError):
del record["nonexistent"]
def test_attribute_get_existing_field(self, record):
"""Test accessing a native field via attribute."""
record.date_time = "2025-01-01T12:00:00+00:00"
assert record.date_time is not None
def test_attribute_get_existing_measurement(self, record):
"""Test accessing an existing measurement via attribute."""
assert record.dish_washer_mr == 123.0
@@ -116,11 +101,6 @@ class TestMeasurementDataRecord:
with pytest.raises(AttributeError):
_ = record.nonexistent
def test_attribute_set_existing_field(self, record):
"""Test setting a native field via attribute."""
record.date_time = "2025-06-25T12:00:00+00:00"
assert record.date_time is not None
def test_attribute_set_existing_measurement(self, record):
"""Test setting a known measurement key via attribute."""
record.temp = 99.9
@@ -131,12 +111,6 @@ class TestMeasurementDataRecord:
with pytest.raises(AttributeError):
record.invalid = 123
def test_delattr_field(self, record):
"""Test deleting a native field via attribute."""
record.date_time = "2025-06-25T12:00:00+00:00"
del record.date_time
assert record.date_time is None
def test_delattr_measurement(self, record):
"""Test deleting a known measurement key via attribute."""
record.temp = 88.0
@@ -154,12 +128,6 @@ class TestMeasurementDataRecord:
assert set(record.record_keys_writable()) == set(keys)
assert len(record) == len(keys)
def test_in_operator_includes_measurements(self, record):
"""Test that 'in' operator includes measurement keys."""
assert "dish_washer_mr" in record
assert "temp" in record # known key, even if not yet set
assert "nonexistent" not in record
def test_hasattr_behavior(self, record):
"""Test that hasattr returns True for fields and known measurements."""
assert hasattr(record, "date_time")
@@ -183,11 +151,6 @@ class TestMeasurementDataRecord:
assert copied.temp == 22.2
assert copied is not record
def test_equality_includes_measurements(self, record):
"""Test that equality includes the `measurements` content."""
other = record.model_copy()
assert record == other
def test_inequality_differs_with_measurements(self, record):
"""Test that records with different measurements are not equal."""
other = record.model_copy(deep=True)
@@ -203,13 +166,6 @@ class TestMeasurementDataRecord:
assert "temp" in record # allowed but not yet set
assert "unknown" not in record
def test_hasattr_equivalence_to_getattr(self, record):
"""hasattr should return True for all valid keys/measurements."""
assert hasattr(record, "dish_washer_mr")
assert hasattr(record, "temp")
assert hasattr(record, "date_time")
assert not hasattr(record, "nonexistent")
def test_dir_includes_measurement_keys(self, record):
"""`dir(record)` should include measurement keys for introspection.
It shall not include the internal 'measurements' attribute.
@@ -432,3 +388,161 @@ class TestMeasurement:
result = await measurement_eos.load_total_kwh(start_datetime=start_datetime, end_datetime=end_datetime, interval=interval)
expected = np.array([100]) # Only one complete interval covered
np.testing.assert_array_equal(result, expected)
class TestMeasurementFileRestore:
@pytest.fixture
def file_measurements(self, config_eos, tmp_path, monkeypatch):
measurement = get_measurement()
config_eos.measurement.load_emr_keys = ["meter"]
config_eos.general.data_folder_path = tmp_path
config_eos.database.provider = None
measurement._db_reset_state()
monkeypatch.setattr(DataSequence, "save", AsyncMock(return_value=False))
monkeypatch.setattr(DataSequence, "load", AsyncMock(return_value=False))
try:
yield measurement
finally:
measurement._db_reset_state()
@pytest.mark.asyncio
async def test_json_roundtrip_restores_records_into_existing_singleton(self, file_measurements):
measurement = file_measurements
await measurement.update_value(to_datetime("2026-09-16T08:00:00Z"), "meter", 123.5)
await measurement.update_value("2026-09-16T09:00:00Z", "meter", 124.0)
assert await measurement.save()
measurement._db_reset_state()
assert await measurement.load()
assert get_measurement() is measurement
assert [record["meter"] for record in measurement.records] == [123.5, 124.0]
restored_hours = []
for record in measurement.records:
assert record.date_time is not None
restored_hours.append(record.date_time.in_timezone("UTC").hour)
assert restored_hours == [8, 9]
# Reloading must update matching timestamps, without duplicating them.
assert await measurement.load()
assert len(measurement.records) == 2
@pytest.mark.asyncio
async def test_json_restore_preserves_other_in_memory_timestamps(self, file_measurements):
measurement = file_measurements
await measurement.update_value("2026-09-16T08:00:00Z", "meter", 123.5)
assert await measurement.save()
measurement._db_reset_state()
await measurement.update_value("2026-09-16T09:00:00Z", "meter", 124.0)
assert await measurement.load()
assert [record["meter"] for record in measurement.records] == [123.5, 124.0]
@pytest.mark.asyncio
@pytest.mark.parametrize("payload", [
"invalid json",
json.dumps({"records": [
{"date_time": "2026-09-16T08:00:00Z", "meter": 123.5},
{"date_time": "not a date", "meter": 124.0},
]}),
])
async def test_invalid_json_does_not_partially_restore(self, file_measurements, payload):
measurement = file_measurements
measurement._measurement_file_path().write_text(payload, encoding="utf-8")
assert await measurement.load() is False
assert measurement.records == []
@pytest.mark.asyncio
async def test_database_success_does_not_read_json(self, file_measurements, monkeypatch):
measurement = file_measurements
measurement._measurement_file_path().write_text("invalid json", encoding="utf-8")
monkeypatch.setattr(DataSequence, "load", AsyncMock(return_value=True))
assert await measurement.load()
@pytest.mark.asyncio
async def test_missing_json_reports_not_loaded(self, file_measurements):
assert await file_measurements.load() is False
CHANNELS = {
"house_power": dict(quantity="power", unit="W", integration_method="hold", max_gap_seconds=120),
"house_meter": dict(quantity="cumulative_energy", unit="kWh"),
"house_interval": dict(
quantity="interval_energy", unit="Wh", interval_seconds=900, timestamp_reference="start"
),
}
class TestMeasurementChannels:
@pytest.mark.parametrize(
"definition",
[
dict(quantity="power", unit="kWh", integration_method="hold", max_gap_seconds=60),
dict(quantity="power", unit="W"),
dict(quantity="power", unit="W", integration_method="hold", max_gap_seconds=0),
dict(quantity="cumulative_energy", unit="W"),
dict(quantity="cumulative_energy", unit="kWh", timestamp_reference="end"),
dict(quantity="interval_energy", unit="Wh", interval_seconds=900),
dict(
quantity="interval_energy", unit="Wh", interval_seconds=True, timestamp_reference="end"
),
],
)
def test_reject_ambiguous_channel(self, definition):
with pytest.raises(ValidationError):
MeasurementChannelSettings(**definition)
@pytest.mark.parametrize(
"legacy_field",
["load_emr_keys", "grid_import_emr_keys", "grid_export_emr_keys", "pv_production_emr_keys"],
)
def test_legacy_keys_keep_meter_semantics(self, legacy_field):
settings = MeasurementCommonSettings.model_validate(dict(**{legacy_field: ["legacy"]}, channels=CHANNELS))
assert settings.keys == sorted(["legacy", *CHANNELS])
compatible = MeasurementCommonSettings.model_validate(dict(
**{legacy_field: ["legacy"]}, channels={"legacy": CHANNELS["house_meter"]}
))
assert compatible.keys == ["legacy"]
with pytest.raises(ValidationError, match="must remain"):
MeasurementCommonSettings.model_validate(dict(
**{legacy_field: ["legacy"]}, channels={"legacy": CHANNELS["house_power"]}
))
@pytest.mark.parametrize("key", ["", " x", "date_time", "configured_data", "keys", "_private"])
def test_reject_reserved_key(self, key):
with pytest.raises(ValidationError):
MeasurementCommonSettings.model_validate(dict(channels={key: CHANNELS["house_meter"]}))
@pytest.mark.asyncio
async def test_existing_import_and_file_reload(self, config_eos, tmp_path, monkeypatch):
"""All three quantities retain their raw values through the existing file path."""
measurement = get_measurement()
previous_settings = config_eos.measurement
previous_records = measurement.records
previous_folder = config_eos.general.data_folder_path
try:
config_eos.measurement = MeasurementCommonSettings.model_validate(dict(channels=CHANNELS))
config_eos.general.data_folder_path = tmp_path
measurement._db_reset_state()
values = dict(house_power=800.0, house_meter=12345.6, house_interval=200.0)
for key, value in values.items():
(await measurement.update_value(to_datetime("2026-09-10T18:00:00Z"), key, value))
assert set(values).issubset(measurement.record_keys)
monkeypatch.setattr(DataSequence, "save", AsyncMock(return_value=False))
monkeypatch.setattr(DataSequence, "load", AsyncMock(return_value=False))
assert (await measurement.save())
measurement._db_reset_state()
assert (await measurement.load())
assert len(measurement.records) == 1
for key, value in values.items():
assert measurement.records[0][key] == value
restored = MeasurementDataRecord.model_validate_json(
measurement.records[0].model_dump_json()
)
assert restored.configured_data == values
settings = MeasurementCommonSettings.model_validate_json(
config_eos.measurement.model_dump_json()
)
assert settings.model_dump() == config_eos.measurement.model_dump()
finally:
measurement._db_reset_state()
measurement.records = previous_records
config_eos.measurement = previous_settings
config_eos.general.data_folder_path = previous_folder
tests/test_measurement_channels.py
-106
View File
@@ -1,106 +0,0 @@
from unittest.mock import AsyncMock
"""Contracts for typed channels sharing the existing measurement storage."""
# ruff: noqa: S101
import pytest
from pydantic import ValidationError
from akkudoktoreos.core.coreabc import get_measurement
from akkudoktoreos.measurement.measurement import (
MeasurementChannelSettings,
MeasurementCommonSettings,
MeasurementDataRecord,
)
from akkudoktoreos.utils.datetimeutil import to_datetime
CHANNELS = {
"house_power": dict(quantity="power", unit="W", integration_method="hold", max_gap_seconds=120),
"house_meter": dict(quantity="cumulative_energy", unit="kWh"),
"house_interval": dict(
quantity="interval_energy", unit="Wh", interval_seconds=900, timestamp_reference="start"
),
}
@pytest.mark.parametrize(
"definition",
[
dict(quantity="power", unit="kWh", integration_method="hold", max_gap_seconds=60),
dict(quantity="power", unit="W"),
dict(quantity="power", unit="W", integration_method="hold", max_gap_seconds=0),
dict(quantity="cumulative_energy", unit="W"),
dict(quantity="cumulative_energy", unit="kWh", timestamp_reference="end"),
dict(quantity="interval_energy", unit="Wh", interval_seconds=900),
dict(
quantity="interval_energy", unit="Wh", interval_seconds=True, timestamp_reference="end"
),
],
)
def test_reject_ambiguous_channel(definition):
with pytest.raises(ValidationError):
MeasurementChannelSettings(**definition)
@pytest.mark.parametrize(
"legacy_field",
["load_emr_keys", "grid_import_emr_keys", "grid_export_emr_keys", "pv_production_emr_keys"],
)
def test_legacy_keys_keep_meter_semantics(legacy_field):
settings = MeasurementCommonSettings.model_validate(dict(**{legacy_field: ["legacy"]}, channels=CHANNELS))
assert settings.keys == sorted(["legacy", *CHANNELS])
compatible = MeasurementCommonSettings.model_validate(dict(
**{legacy_field: ["legacy"]}, channels={"legacy": CHANNELS["house_meter"]}
))
assert compatible.keys == ["legacy"]
with pytest.raises(ValidationError, match="must remain"):
MeasurementCommonSettings.model_validate(dict(
**{legacy_field: ["legacy"]}, channels={"legacy": CHANNELS["house_power"]}
))
@pytest.mark.parametrize("key", ["", " x", "date_time", "configured_data", "keys", "_private"])
def test_reject_reserved_key(key):
with pytest.raises(ValidationError):
MeasurementCommonSettings.model_validate(dict(channels={key: CHANNELS["house_meter"]}))
@pytest.mark.asyncio
async def test_existing_import_and_file_reload(config_eos, tmp_path, monkeypatch):
"""All three quantities retain their raw values through the existing file path."""
from akkudoktoreos.core.dataabc import DataSequence
measurement = get_measurement()
previous_settings = config_eos.measurement
previous_records = measurement.records
previous_folder = config_eos.general.data_folder_path
try:
config_eos.measurement = MeasurementCommonSettings.model_validate(dict(channels=CHANNELS))
config_eos.general.data_folder_path = tmp_path
measurement._db_reset_state()
values = dict(house_power=800.0, house_meter=12345.6, house_interval=200.0)
for key, value in values.items():
(await measurement.update_value(to_datetime("2026-09-10T18:00:00Z"), key, value))
assert set(values).issubset(measurement.record_keys)
monkeypatch.setattr(DataSequence, "save", AsyncMock(return_value=False))
monkeypatch.setattr(DataSequence, "load", AsyncMock(return_value=False))
assert (await measurement.save())
measurement._db_reset_state()
assert (await measurement.load())
assert len(measurement.records) == 1
for key, value in values.items():
assert measurement.records[0][key] == value
restored = MeasurementDataRecord.model_validate_json(
measurement.records[0].model_dump_json()
)
assert restored.configured_data == values
settings = MeasurementCommonSettings.model_validate_json(
config_eos.measurement.model_dump_json()
)
assert settings.model_dump() == config_eos.measurement.model_dump()
finally:
measurement._db_reset_state()
measurement.records = previous_records
config_eos.measurement = previous_settings
config_eos.general.data_folder_path = previous_folder
tests/test_measurement_file_restore.py
-84
View File
@@ -1,84 +0,0 @@
"""Regression coverage for measurement persistence through the JSON fallback."""
import json
from unittest.mock import AsyncMock
import pytest
from akkudoktoreos.core.coreabc import get_measurement
from akkudoktoreos.core.dataabc import DataSequence
@pytest.fixture
def file_measurements(config_eos, tmp_path, monkeypatch):
measurement = get_measurement()
config_eos.measurement.load_emr_keys = ["meter"]
config_eos.general.data_folder_path = tmp_path
config_eos.database.provider = None
measurement._db_reset_state()
monkeypatch.setattr(DataSequence, "save", AsyncMock(return_value=False))
monkeypatch.setattr(DataSequence, "load", AsyncMock(return_value=False))
try:
yield measurement
finally:
measurement._db_reset_state()
@pytest.mark.asyncio
async def test_json_roundtrip_restores_records_into_existing_singleton(file_measurements):
measurement = file_measurements
await measurement.update_value("2026-09-16T08:00:00Z", "meter", 123.5)
await measurement.update_value("2026-09-16T09:00:00Z", "meter", 124.0)
assert await measurement.save()
measurement._db_reset_state()
assert await measurement.load()
assert get_measurement() is measurement
assert [record["meter"] for record in measurement.records] == [123.5, 124.0]
restored_hours = []
for record in measurement.records:
assert record.date_time is not None
restored_hours.append(record.date_time.in_timezone("UTC").hour)
assert restored_hours == [8, 9]
# Reloading must update matching timestamps, without duplicating them.
assert await measurement.load()
assert len(measurement.records) == 2
@pytest.mark.asyncio
async def test_json_restore_preserves_other_in_memory_timestamps(file_measurements):
measurement = file_measurements
await measurement.update_value("2026-09-16T08:00:00Z", "meter", 123.5)
assert await measurement.save()
measurement._db_reset_state()
await measurement.update_value("2026-09-16T09:00:00Z", "meter", 124.0)
assert await measurement.load()
assert [record["meter"] for record in measurement.records] == [123.5, 124.0]
@pytest.mark.asyncio
@pytest.mark.parametrize("payload", [
"invalid json",
json.dumps({"records": [
{"date_time": "2026-09-16T08:00:00Z", "meter": 123.5},
{"date_time": "not a date", "meter": 124.0},
]}),
])
async def test_invalid_json_does_not_partially_restore(file_measurements, payload):
measurement = file_measurements
measurement._measurement_file_path().write_text(payload, encoding="utf-8")
assert await measurement.load() is False
assert measurement.records == []
@pytest.mark.asyncio
async def test_database_success_does_not_read_json(file_measurements, monkeypatch):
measurement = file_measurements
measurement._measurement_file_path().write_text("invalid json", encoding="utf-8")
monkeypatch.setattr(DataSequence, "load", AsyncMock(return_value=True))
assert await measurement.load()
@pytest.mark.asyncio
async def test_missing_json_reports_not_loaded(file_measurements):
assert await file_measurements.load() is False
tests/test_pvforecastpvlib.py
+1 -27
View File
@@ -1,12 +1,9 @@
"""Tests for PVForecastPVLib prediction provider."""
import bz2
import pickle
from pathlib import Path
from unittest.mock import MagicMock, Mock, patch
from unittest.mock import MagicMock, patch
import numpy as np
import pandas as pd
import pytest
@@ -1054,7 +1051,6 @@ class TestProvider:
weather = pd.DataFrame()
with patch.object(
Prediction,
"keys_to_dataframe",
@@ -1285,28 +1281,6 @@ class TestProvider:
class TestCalculation:
"""Tests for PVLib power calculation."""
@pytest.mark.skip("PVLib bails out on empty weather data")
def test_calculate_empty_weather(self, provider, sample_settings_4planes):
"""Empty weather dataframe returns empty result."""
weather = pd.DataFrame(
columns=[
"ghi",
"dni",
"dhi",
"temp_air",
"wind_speed",
"relative_humidity",
"precipitable_water",
"cloud_cover",
],
index=pd.DatetimeIndex([], tz="Europe/Berlin"),
)
result = provider._calculate_pvlib_power(weather)
assert isinstance(result, pd.DataFrame)
assert result.empty
@patch("pvlib.modelchain.ModelChain")
@patch("pvlib.location.Location")
tests/test_tailvalue_physics.py
-199
View File
@@ -1,199 +0,0 @@
"""Economic tail scenarios and hard control/forecast boundaries."""
from unittest.mock import patch
import numpy as np
import pandas as pd
import pytest
from akkudoktoreos.config.config import SettingsEOSDefaults
from akkudoktoreos.core.coreabc import get_ems
from akkudoktoreos.devices.genetic.battery import Battery, SolarPanelBatteryParameters
from akkudoktoreos.devices.genetic.inverter import Inverter, InverterParameters
from akkudoktoreos.optimization.genetic.forecast import bounded_forecast_array
from akkudoktoreos.optimization.genetic.genetic import GeneticOptimization
from akkudoktoreos.optimization.genetic.geneticparams import (
GeneticOptimizationParameters,
)
from akkudoktoreos.optimization.genetic.tailvalue import build_tail_value_curve
from akkudoktoreos.optimization.genetic.terminalvalue import TerminalValueCurve
from akkudoktoreos.utils.datetimeutil import to_datetime, to_duration
def devices(power=1000, efficiency=1.0, lcos=0, ac_limit=None, export_power=5000):
bat = Battery(
SolarPanelBatteryParameters(
device_id="battery1",
capacity_wh=1000,
max_charge_power_w=power,
charging_efficiency=efficiency,
discharging_efficiency=efficiency,
initial_soc_percentage=50,
levelized_cost_of_storage_kwh=lcos,
charge_rates=[0, 0.5, 1],
),
prediction_hours=1,
)
inv = Inverter(
InverterParameters(
device_id="inverter1",
battery_id="battery1",
max_power_wh=export_power,
dc_to_ac_efficiency=1,
ac_to_dc_efficiency=1,
max_ac_charge_power_w=ac_limit,
),
battery=bat,
)
return bat, inv
def curve(
prices=(-0.1, 0.3),
tariffs=(0, 0.3),
direct=True,
continuation=None,
load=None,
pv=None,
**kwargs,
):
bat, inv = devices(**kwargs)
return build_tail_value_curve(
battery=bat,
inverter=inv,
prices_euro_per_wh=np.array(prices) / 1000,
feed_in_euro_per_wh=np.array(tariffs) / 1000,
load_wh=np.zeros(len(prices)) if load is None else np.array(load),
pv_wh=np.zeros(len(prices)) if pv is None else np.array(pv),
continuation=continuation or TerminalValueCurve(),
charge_rates=[0.5, 1],
export_rates=[1],
direct_marketing=direct,
)
def test_headroom_has_value_and_empty_state_can_earn():
c = curve()
assert c.value(0) == pytest.approx(0.4)
tail, continuation = c.component_values(0)
assert tail == pytest.approx(0.4)
assert continuation == pytest.approx(0.0)
assert c.value(0) == pytest.approx(tail + continuation)
assert c.value(500) > c.value(1000)
assert any(v < 0 for v in c.marginal_euro_per_kwh)
def test_chronology_changes_arbitrage():
forward = curve()
reverse = curve(prices=(0.3, -0.1), tariffs=(0.3, 0))
assert forward.value(0) > reverse.value(0)
def test_discharge_and_ac_power_limits():
limited = curve(prices=(1,), tariffs=(1,), power=100)
assert limited.value(1000) == pytest.approx(0.1)
limited_ac = curve(ac_limit=100)
assert limited_ac.value(0) == pytest.approx(0.04)
limited_inverter = curve(prices=(1,), tariffs=(1,), export_power=50)
assert limited_inverter.value(1000) == pytest.approx(0.05)
def test_losses_and_lcos_reduce_arbitrage():
ideal = curve(prices=(0.1, 0.3))
lossy = curve(prices=(0.1, 0.3), efficiency=0.8)
assert 0 < lossy.value(0) < ideal.value(0)
assert curve(prices=(0.1, 0.3), lcos=0.25).value(0) == pytest.approx(0)
def test_no_battery_export_without_permission():
assert curve(prices=(0.1, 0.3), direct=False).value(0) == pytest.approx(0)
def test_pv_surplus_can_be_stored_for_local_load():
c = curve(prices=(0.2, 0.3), tariffs=(0, 0), pv=[1000, 0], load=[0, 1000], direct=False)
assert c.value(0) == pytest.approx(0)
# Without PV the same empty battery must buy energy to serve the load.
assert curve(prices=(0.2, 0.3), tariffs=(0, 0), load=[0, 1000], direct=False).value(
0
) < c.value(0)
def test_continuation_survives_tail_end():
continuation = TerminalValueCurve(energy_wh=[0, 1000], value_euro=[0, 0.2])
c = curve(prices=(0.5,), tariffs=(0,), continuation=continuation)
assert c.value(1000) == pytest.approx(0.2)
tail, continuation_credit = c.component_values(1000)
assert tail == pytest.approx(0.0)
assert continuation_credit == pytest.approx(0.2)
def test_tail_diagnostic_plan_explains_the_selected_path():
c = curve()
plan = c.diagnostic_plan(0, control_horizon_hours=24)
assert len(plan) == 2
assert plan[0].hour_from_start == 24
assert plan[0].action == "GRID_CHARGE"
assert plan[0].soc_end_percentage > plan[0].soc_start_percentage
assert plan[0].grid_import_wh > 0
assert plan[1].action == "BATTERY_EXPORT"
assert plan[1].soc_end_percentage < plan[1].soc_start_percentage
assert plan[1].grid_export_wh > 0
assert sum(slot.slot_value_euro for slot in plan) == pytest.approx(c.value(0))
@pytest.mark.asyncio
async def test_provider_values_are_not_extrapolated():
from types import SimpleNamespace
start = to_datetime("2026-09-05T00:00:00Z")
series = pd.Series([1.0, 2.0], index=pd.date_range(start=start, periods=2, freq="h"))
from unittest.mock import AsyncMock
provider = SimpleNamespace(key_to_raw_series=AsyncMock(return_value=series))
result = await bounded_forecast_array(
provider,
key="price",
start_datetime=start,
end_datetime=start.add(hours=3),
interval=to_duration("15 minutes"),
)
assert result[:8].tolist() == [1.0] * 4 + [2.0] * 4
assert np.isnan(result[8:]).all()
def test_disabled_ac_conversion_cannot_earn_negative_price_revenue():
bat, inv = devices()
inv.parameters.ac_to_dc_efficiency = 0
c = build_tail_value_curve(
battery=bat,
inverter=inv,
prices_euro_per_wh=np.array([-0.001, 0.001]),
feed_in_euro_per_wh=np.array([0.0, 0.001]),
load_wh=np.zeros(2),
pv_wh=np.zeros(2),
continuation=TerminalValueCurve(),
charge_rates=[1],
export_rates=[1],
direct_marketing=True,
)
assert c.value(0) == pytest.approx(0)
assert bat.soc_wh == 500 # Building the tail never mutates the real battery.
@pytest.mark.asyncio
async def test_missing_provider_key_stays_missing():
from types import SimpleNamespace
async def unavailable(*a, **kw):
raise KeyError("price unavailable")
start = to_datetime("2026-09-05T00:00:00Z")
result = await bounded_forecast_array(
SimpleNamespace(key_to_raw_series=unavailable),
key="price",
start_datetime=start,
end_datetime=start.add(hours=2),
interval=to_duration("1 hour"),
)
assert np.isnan(result).all()
assert len(result) == 2
tests/test_weatherclearoutside.py
+1 -1
View File
@@ -1,5 +1,5 @@
import re
from datetime import datetime, timedelta, timezone
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any
from unittest.mock import Mock, patch