feat(prediction): add Solcast PV forecast provider

Add PVForecastSolcast for the Solcast rooftop-site API. The operator registers a site in the Solcast web app and enters the API key + resource (site) id: GET /rooftop_sites/{site_id}/forecasts. pv_estimate (kW) is converted to watts and fed as pvforecast_ac_power; the timestamp is normalised to the period start (period_end - period) so it aligns with the resample axis. period_end is UTC. Completes the set of selectable cloud PV forecast providers (Akkudoktor, VRM, Import, pvnode, Forecast.Solar, Solcast). Adds tests for the kW->W conversion, period-start normalisation, ISO-8601 period parsing, the request URL/auth and HTTP-error handling.
2026-07-12 21:08:13 +00:00 · 2026-06-28 03:06:55 +00:00
parent a1e2100206
commit cec9e35aa9
4 changed files with 250 additions and 0 deletions
--- a/tests/test_pvforecastsolcast.py
+++ b/tests/test_pvforecastsolcast.py
@@ -0,0 +1,97 @@
+from unittest.mock import call, patch
+
+import pendulum
+import pytest
+import requests
+
+from akkudoktoreos.prediction.pvforecastsolcast import PVForecastSolcast
+
+
+@pytest.fixture
+def pvforecast_instance(config_eos):
+    settings = {
+        "pvforecast": {
+            "provider": "PVForecastSolcast",
+            "provider_settings": {
+                "PVForecastSolcast": {"api_key": "dummy-key", "site_id": "site-abc"},
+            },
+        },
+    }
+    config_eos.merge_settings_from_dict(settings)
+    start_dt = pendulum.datetime(2025, 1, 1, tz="Europe/Berlin")
+    return PVForecastSolcast(config=config_eos.load, start_datetime=start_dt)
+
+
+def _http(forecasts):
+    return type(
+        "R",
+        (),
+        {
+            "raise_for_status": lambda self: None,
+            "json": lambda self: {"forecasts": forecasts},
+        },
+    )()
+
+
+def test_provider_id(pvforecast_instance):
+    assert PVForecastSolcast.provider_id() == "PVForecastSolcast"
+    assert pvforecast_instance.enabled() is True
+
+
+@pytest.mark.parametrize(
+    "period,minutes",
+    [("PT30M", 30), ("PT15M", 15), ("PT5M", 5), ("PT1H", 60), ("PT1H30M", 90), ("", 0), (None, 0)],
+)
+def test_period_minutes(period, minutes):
+    assert PVForecastSolcast._period_minutes(period) == minutes
+
+
+def test_update_data_normalises_to_period_start_and_converts_kw(pvforecast_instance):
+    """pv_estimate (kW) -> W; timestamp = period_end - period; period_end is UTC."""
+    body = {
+        "forecasts": [
+            {"pv_estimate": 1.2, "period_end": "2025-01-01T12:30:00.0000000Z", "period": "PT30M"},
+            {"pv_estimate": 0.0, "period_end": "2025-01-01T13:00:00.0000000Z", "period": "PT30M"},
+        ]
+    }
+    with patch.object(pvforecast_instance, "_request_forecast", return_value=body), \
+         patch.object(PVForecastSolcast, "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="UTC"),
+                {"pvforecast_ac_power": 1200.0, "pvforecast_dc_power": 1200.0},
+            ),
+            call(
+                pendulum.datetime(2025, 1, 1, 12, 30, tz="UTC"),
+                {"pvforecast_ac_power": 0.0, "pvforecast_dc_power": 0.0},
+            ),
+        ]
+        mock_update.assert_has_calls(expected, any_order=False)
+
+
+def test_request_forecast_uses_site_and_bearer(pvforecast_instance):
+    with patch("requests.get", return_value=_http([])) as mock_get:
+        pvforecast_instance._request_forecast(force_update=True)
+        url = mock_get.call_args[0][0]
+        assert url.endswith("/rooftop_sites/site-abc/forecasts")
+        assert mock_get.call_args.kwargs["headers"]["Authorization"] == "Bearer dummy-key"
+        assert mock_get.call_args.kwargs["params"]["format"] == "json"
+
+
+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(PVForecastSolcast, "update_value") as mock_update:
+        pvforecast_instance._update_data()
+        mock_req.assert_not_called()
+        mock_update.assert_not_called()
+
+
+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 Solcast" in str(exc_info.value)