mirror of
https://github.com/Akkudoktor-EOS/EOS.git
synced 2026-07-13 05:18:12 +00:00
feat(prediction): add native pvnode.com PV forecast provider
Add PVForecastPVNode, a native 15-minute PV forecast provider for the
pvnode.com V2 API, giving operators another forecast source to choose from
alongside Akkudoktor, VRM and Import.
Two request modes, selected by configuration:
* site_id set -> GET /v2/forecast/{site_id} (a saved, possibly calibrated
site managed on pvnode.com — the operator enters site id + API key)
* site_id empty -> POST /v2/forecast/inline (geometry sent inline from the
configured pvforecast.planes; no web-app setup required)
V2 response timestamps are site-local wall-clock accompanied by an IANA
timezone; they are resolved to absolute instants before EOS resamples them.
Nullable pv_power (e.g. at night) is treated as 0 W so the optimizer's linear
resampling does not interpolate phantom production across the night.
Registered in pvforecast.py (provider settings + id list) and prediction.py
(singleton, factory list, container union). Adds tests covering timezone
resolution, null handling, both request modes and HTTP-error propagation.
This commit is contained in:
143
tests/test_pvforecastpvnode.py
Normal file
143
tests/test_pvforecastpvnode.py
Normal file
@@ -0,0 +1,143 @@
|
||||
from unittest.mock import call, patch
|
||||
|
||||
import pendulum
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
from akkudoktoreos.prediction.pvforecastpvnode import PVForecastPVNode
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def pvforecast_instance(config_eos):
|
||||
settings = {
|
||||
"general": {"latitude": 52.5, "longitude": 13.4},
|
||||
"pvforecast": {
|
||||
"provider": "PVForecastPVNode",
|
||||
"provider_settings": {
|
||||
"PVForecastPVNode": {
|
||||
"api_key": "dummy-key",
|
||||
"site_id": "test-site-123",
|
||||
"forecast_days": 2,
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
config_eos.merge_settings_from_dict(settings)
|
||||
start_dt = pendulum.datetime(2025, 1, 1, tz="Europe/Berlin")
|
||||
return PVForecastPVNode(config=config_eos.load, start_datetime=start_dt)
|
||||
|
||||
|
||||
def mock_v2_body():
|
||||
"""A canonical pvnode V2 response: site-local wall-clock + IANA timezone.
|
||||
|
||||
The second slot carries a null ``pv_power`` (night) which must become 0 W.
|
||||
"""
|
||||
return {
|
||||
"timezone": "Europe/Berlin",
|
||||
"values": [
|
||||
{"timestamp": "2025-01-01T12:00:00", "pv_power": 1200.0},
|
||||
{"timestamp": "2025-01-01T12:15:00", "pv_power": None},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def test_provider_id(pvforecast_instance):
|
||||
assert PVForecastPVNode.provider_id() == "PVForecastPVNode"
|
||||
assert pvforecast_instance.enabled() is True
|
||||
|
||||
|
||||
def test_extract_values_resolves_local_wall_clock_to_instant(pvforecast_instance):
|
||||
"""V2 timestamps are local wall-clock; 12:00 Europe/Berlin (CET) == 11:00 UTC."""
|
||||
rows = pvforecast_instance._extract_values(mock_v2_body())
|
||||
|
||||
assert len(rows) == 2
|
||||
# Same instant, regardless of representation.
|
||||
assert rows[0][0] == pendulum.datetime(2025, 1, 1, 12, 0, tz="Europe/Berlin")
|
||||
assert rows[0][0].in_timezone("UTC").hour == 11
|
||||
assert rows[0][1] == 1200.0
|
||||
# Null pv_power -> 0 W (not a gap, to avoid phantom night interpolation).
|
||||
assert rows[1][1] == 0.0
|
||||
|
||||
|
||||
def test_extract_values_trusts_explicit_offset(pvforecast_instance):
|
||||
"""A timestamp already carrying an offset is trusted as-is (no double shift)."""
|
||||
body = {
|
||||
"timezone": "Europe/Berlin",
|
||||
"values": [{"timestamp": "2025-01-01T12:00:00+00:00", "pv_power": 500.0}],
|
||||
}
|
||||
rows = pvforecast_instance._extract_values(body)
|
||||
assert rows[0][0].in_timezone("UTC").hour == 12
|
||||
|
||||
|
||||
def test_update_data_sets_ac_and_dc_power(pvforecast_instance):
|
||||
with patch.object(pvforecast_instance, "_request_forecast", return_value=mock_v2_body()), \
|
||||
patch.object(PVForecastPVNode, "update_value") as mock_update:
|
||||
pvforecast_instance._update_data()
|
||||
|
||||
assert mock_update.call_count == 2
|
||||
expected = [
|
||||
call(
|
||||
pendulum.datetime(2025, 1, 1, 12, 0, tz="Europe/Berlin"),
|
||||
{"pvforecast_ac_power": 1200.0, "pvforecast_dc_power": 1200.0},
|
||||
),
|
||||
call(
|
||||
pendulum.datetime(2025, 1, 1, 12, 15, tz="Europe/Berlin"),
|
||||
{"pvforecast_ac_power": 0.0, "pvforecast_dc_power": 0.0},
|
||||
),
|
||||
]
|
||||
mock_update.assert_has_calls(expected, any_order=False)
|
||||
|
||||
|
||||
def test_update_data_skips_when_disabled(pvforecast_instance, config_eos):
|
||||
config_eos.merge_settings_from_dict({"pvforecast": {"provider": "PVForecastAkkudoktor"}})
|
||||
with patch.object(pvforecast_instance, "_request_forecast") as mock_req, \
|
||||
patch.object(PVForecastPVNode, "update_value") as mock_update:
|
||||
pvforecast_instance._update_data()
|
||||
mock_req.assert_not_called()
|
||||
mock_update.assert_not_called()
|
||||
|
||||
|
||||
def test_update_data_skips_on_empty_forecast(pvforecast_instance):
|
||||
with patch.object(pvforecast_instance, "_request_forecast", return_value={"values": []}), \
|
||||
patch.object(PVForecastPVNode, "update_value") as mock_update:
|
||||
pvforecast_instance._update_data()
|
||||
mock_update.assert_not_called()
|
||||
|
||||
|
||||
def test_request_forecast_uses_saved_site_get(pvforecast_instance):
|
||||
"""site_id set -> GET /v2/forecast/{site_id} with Bearer auth."""
|
||||
fake = type("R", (), {"raise_for_status": lambda self: None, "json": lambda self: {"values": []}})()
|
||||
with patch("requests.get", return_value=fake) as mock_get:
|
||||
pvforecast_instance._request_forecast(force_update=True)
|
||||
url = mock_get.call_args[0][0]
|
||||
assert url.endswith("/v2/forecast/test-site-123")
|
||||
assert mock_get.call_args.kwargs["headers"]["Authorization"] == "Bearer dummy-key"
|
||||
|
||||
|
||||
def test_request_forecast_inline_post_when_no_site(config_eos):
|
||||
"""No site_id -> POST /v2/forecast/inline with planes geometry."""
|
||||
config_eos.merge_settings_from_dict(
|
||||
{
|
||||
"general": {"latitude": 52.5, "longitude": 13.4},
|
||||
"pvforecast": {
|
||||
"provider": "PVForecastPVNode",
|
||||
"planes": [{"surface_tilt": 30.0, "surface_azimuth": 180.0, "peakpower": 5.0}],
|
||||
"provider_settings": {"PVForecastPVNode": {"api_key": "k", "site_id": None}},
|
||||
},
|
||||
}
|
||||
)
|
||||
pv = PVForecastPVNode(config=config_eos.load, start_datetime=pendulum.datetime(2025, 1, 1, tz="UTC"))
|
||||
fake = type("R", (), {"raise_for_status": lambda self: None, "json": lambda self: {"values": []}})()
|
||||
with patch("requests.post", return_value=fake) as mock_post:
|
||||
pv._request_forecast(force_update=True)
|
||||
url = mock_post.call_args[0][0]
|
||||
assert url.endswith("/v2/forecast/inline")
|
||||
body = mock_post.call_args.kwargs["json"]
|
||||
assert body["strings"][0] == {"slope": 30.0, "orientation": 180.0, "power_kw": 5.0}
|
||||
|
||||
|
||||
def test_request_forecast_raises_on_http_error(pvforecast_instance):
|
||||
with patch("requests.get", side_effect=requests.Timeout("timed out")):
|
||||
with pytest.raises(RuntimeError) as exc_info:
|
||||
pvforecast_instance._request_forecast(force_update=True)
|
||||
assert "Failed to fetch pvforecast from pvnode" in str(exc_info.value)
|
||||
Reference in New Issue
Block a user