Mirrors/EOS
1
0
Fork 0
mirror of https://github.com/Akkudoktor-EOS/EOS.git synced 2026-07-12 21:08:13 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

feat(prediction): add Forecast.Solar PV forecast provider

Browse Source 
Add PVForecastForecastSolar, a PV forecast provider for the free Forecast.Solar
API (https://forecast.solar), giving operators a no-account forecast source in
addition to Akkudoktor, VRM, Import and pvnode. An optional API key raises the
rate limit.

result.watts is the instantaneous AC power per timestamp, fed directly as
pvforecast_ac_power. Plants with several roof planes issue one request per plane
(Forecast.Solar is single-plane) and the powers are summed per timestamp.
Forecast.Solar azimuth (-180=N..0=S..90=W) is converted from EOS surface_azimuth
(north=0..south=180); local wall-clock timestamps are resolved via the response
timezone before resampling.

Registered in pvforecast.py and prediction.py. Adds tests for timezone
resolution, azimuth conversion, multi-plane summation and HTTP-error handling.
This commit is contained in:
Christin
2026-06-28 03:03:12 +00:00
parent 314e45cfaa
commit a1e2100206
4 changed files with 308 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/akkudoktoreos/prediction/prediction.py
View File

@@ -45,6 +45,7 @@ from akkudoktoreos.prediction.loadimport import LoadImport
from akkudoktoreos.prediction.loadvrm import LoadVrm
from akkudoktoreos.prediction.predictionabc import PredictionContainer
from akkudoktoreos.prediction.pvforecastakkudoktor import PVForecastAkkudoktor
from akkudoktoreos.prediction.pvforecastforecastsolar import PVForecastForecastSolar
from akkudoktoreos.prediction.pvforecastimport import PVForecastImport
from akkudoktoreos.prediction.pvforecastpvnode import PVForecastPVNode
from akkudoktoreos.prediction.pvforecastvrm import PVForecastVrm
@@ -86,6 +87,7 @@ loadforecast_import = LoadImport()
pvforecast_akkudoktor = PVForecastAkkudoktor()
pvforecast_vrm = PVForecastVrm()
pvforecast_pvnode = PVForecastPVNode()
pvforecast_forecastsolar = PVForecastForecastSolar()
pvforecast_import = PVForecastImport()
weather_brightsky = WeatherBrightSky()
weather_clearoutside = WeatherClearOutside()
@@ -108,6 +110,7 @@ def prediction_providers() -> list[
PVForecastAkkudoktor,
PVForecastVrm,
PVForecastPVNode,
PVForecastForecastSolar,
PVForecastImport,
WeatherBrightSky,
WeatherClearOutside,
@@ -133,6 +136,7 @@ def prediction_providers() -> list[
pvforecast_akkudoktor, \
pvforecast_vrm, \
pvforecast_pvnode, \
pvforecast_forecastsolar, \
pvforecast_import, \
weather_brightsky, \
weather_clearoutside, \
@@ -154,6 +158,7 @@ def prediction_providers() -> list[
pvforecast_akkudoktor,
pvforecast_vrm,
pvforecast_pvnode,
pvforecast_forecastsolar,
pvforecast_import,
weather_brightsky,
weather_clearoutside,
@@ -180,6 +185,7 @@ class Prediction(PredictionContainer):
PVForecastAkkudoktor,
PVForecastVrm,
PVForecastPVNode,
PVForecastForecastSolar,
PVForecastImport,
WeatherBrightSky,
WeatherClearOutside,

15
src/akkudoktoreos/prediction/pvforecast.py
View File

@@ -7,6 +7,9 @@ from pydantic import Field, computed_field, field_validator, model_validator
from akkudoktoreos.config.configabc import SettingsBaseModel
from akkudoktoreos.core.coreabc import get_prediction
from akkudoktoreos.prediction.pvforecastabc import PVForecastProvider
from akkudoktoreos.prediction.pvforecastforecastsolar import (
PVForecastForecastSolarCommonSettings,
)
from akkudoktoreos.prediction.pvforecastimport import PVForecastImportCommonSettings
from akkudoktoreos.prediction.pvforecastpvnode import PVForecastPVNodeCommonSettings
from akkudoktoreos.prediction.pvforecastvrm import PVForecastVrmCommonSettings
@@ -19,7 +22,13 @@ def pvforecast_provider_ids() -> list[str]:
except:
# Prediction may not be initialized
# Return at least provider used in example
return ["PVForecastAkkudoktor", "PVForecastImport", "PVForecastVrm", "PVForecastPVNode"]
return [
"PVForecastAkkudoktor",
"PVForecastImport",
"PVForecastVrm",
"PVForecastPVNode",
"PVForecastForecastSolar",
]
return [
provider.provider_id()
@@ -184,6 +193,10 @@ class PVForecastCommonProviderSettings(SettingsBaseModel):
default=None,
json_schema_extra={"description": "PVForecastPVNode settings", "examples": [None]},
)
PVForecastForecastSolar: Optional[PVForecastForecastSolarCommonSettings] = Field(
default=None,
json_schema_extra={"description": "PVForecastForecastSolar settings", "examples": [None]},
)
class PVForecastCommonSettings(SettingsBaseModel):

161
src/akkudoktoreos/prediction/pvforecastforecastsolar.py Normal file
View File

@@ -0,0 +1,161 @@
"""Retrieves PV forecast data from the Forecast.Solar API.
Forecast.Solar (https://forecast.solar) is a free public PV forecast service
(no API key required; an optional key raises the rate/feature limits). Each
request covers a single plane:
GET https://api.forecast.solar[/{api_key}]/estimate/{lat}/{lon}/{dec}/{az}/{kwp}
``result.watts`` is the instantaneous AC power per timestamp — exactly what the
optimizer consumes as ``pvforecast_ac_power``. EOS plants with several roof
planes issue one request per plane and the instantaneous powers are summed per
timestamp.
Note on conventions:
- Forecast.Solar azimuth is -180=N, -90=E, 0=S, 90=W, whereas EOS
``surface_azimuth`` is north=0, east=90, south=180, west=270. The provider
converts via ``az = surface_azimuth - 180``.
- Response timestamps are local wall-clock; ``message.info.timezone`` is used
to resolve them to absolute instants before EOS resamples them.
"""
import re
from typing import Any, Optional
import pendulum
import requests
from loguru import logger
from pydantic import Field
from akkudoktoreos.config.configabc import SettingsBaseModel
from akkudoktoreos.core.cache import cache_in_file
from akkudoktoreos.prediction.pvforecastabc import PVForecastProvider
from akkudoktoreos.utils.datetimeutil import to_datetime
FORECAST_SOLAR_BASE = "https://api.forecast.solar"
_TZ_SUFFIX = re.compile(r"([zZ]|[+-]\d\d:?\d\d)$")
class PVForecastForecastSolarCommonSettings(SettingsBaseModel):
"""Common settings for the Forecast.Solar PV forecast provider."""
api_key: Optional[str] = Field(
default=None,
json_schema_extra={
"description": (
"Forecast.Solar API key. Optional — the public endpoint works "
"without a key (lower rate limit)."
),
"examples": [None, "your-forecast-solar-key"],
},
)
class PVForecastForecastSolar(PVForecastProvider):
"""Fetch and process PV forecast data from the Forecast.Solar API."""
@classmethod
def provider_id(cls) -> str:
"""Return the unique identifier for the PV-Forecast-Provider."""
return "PVForecastForecastSolar"
@property
def _api_key(self) -> Optional[str]:
settings = self.config.pvforecast.provider_settings.PVForecastForecastSolar
return settings.api_key if settings is not None else None
def _to_utc_datetime(self, local_ts: Any, iana_tz: Optional[str]) -> Any:
"""Resolve a Forecast.Solar wall-clock timestamp to a timezone-aware datetime."""
s = str(local_ts).strip()
if _TZ_SUFFIX.search(s):
return to_datetime(s)
tz = iana_tz or str(self.config.general.timezone)
dt = pendulum.parse(s, tz=tz)
return to_datetime(dt.isoformat())
def _plane_url(self, plane: Any) -> str:
"""Build the single-plane estimate URL for the given plane configuration."""
latitude = self.config.general.latitude
longitude = self.config.general.longitude
if latitude is None or longitude is None:
raise ValueError("PVForecastForecastSolar needs general.latitude/longitude")
tilt = getattr(plane, "surface_tilt", None)
azimuth = getattr(plane, "surface_azimuth", None)
peakpower = getattr(plane, "peakpower", None)
if tilt is None or azimuth is None or peakpower is None:
raise ValueError(
"PVForecastForecastSolar needs surface_tilt, surface_azimuth and "
"peakpower on each pvforecast.planes entry"
)
# EOS azimuth (north=0..south=180) -> Forecast.Solar (north=-180..south=0).
fs_az = float(azimuth) - 180.0
base = FORECAST_SOLAR_BASE
api_key = self._api_key
if api_key:
base = f"{base}/{api_key}"
return f"{base}/estimate/{latitude}/{longitude}/{float(tilt)}/{fs_az}/{float(peakpower)}"
@cache_in_file(with_ttl="1 hour")
def _request_forecast(self) -> dict:
"""Fetch and aggregate the Forecast.Solar estimate across all configured planes."""
planes = self.config.pvforecast.planes or []
if not planes:
raise ValueError("PVForecastForecastSolar needs at least one pvforecast.planes entry")
summed: dict[str, float] = {}
timezone: Optional[str] = None
for plane in planes:
url = self._plane_url(plane)
logger.debug(f"Requesting Forecast.Solar estimate: {url}")
try:
response = requests.get(url, headers={"Accept": "application/json"}, timeout=30)
response.raise_for_status()
except requests.RequestException as e:
logger.error(f"Failed to fetch pvforecast from Forecast.Solar: {e}")
raise RuntimeError("Failed to fetch pvforecast from Forecast.Solar API") from e
data = response.json()
if timezone is None:
timezone = (data.get("message", {}).get("info", {}) or {}).get("timezone")
watts = (data.get("result", {}) or {}).get("watts", {}) or {}
for ts, power in watts.items():
try:
summed[ts] = summed.get(ts, 0.0) + float(power)
except (TypeError, ValueError):
continue
self.update_datetime = to_datetime(in_timezone=self.config.general.timezone)
return {"timezone": timezone, "watts": summed}
def _update_data(self, force_update: Optional[bool] = False) -> None:
"""Update forecast data in the PVForecastDataRecord format."""
if not self.enabled():
logger.info("PVForecastForecastSolar is disabled, skipping update.")
return
body = self._request_forecast(force_update=force_update) # type: ignore[call-arg]
timezone = body.get("timezone")
watts = body.get("watts", {})
count = 0
for ts, power_w in sorted(watts.items()):
try:
date = self._to_utc_datetime(ts, timezone)
except Exception as e: # noqa: BLE001 - skip unparseable rows
logger.warning(f"Forecast.Solar: skipping unparseable timestamp {ts!r}: {e}")
continue
value = round(float(power_w), 1)
self.update_value(
date,
{"pvforecast_ac_power": value, "pvforecast_dc_power": value},
)
count += 1
logger.debug(f"Updated pvforecast from Forecast.Solar with {count} entries.")
self.update_datetime = to_datetime(in_timezone=self.config.general.timezone)
# Example usage
if __name__ == "__main__":
pv = PVForecastForecastSolar()
pv._update_data()

127
tests/test_pvforecastforecastsolar.py Normal file
View File

@@ -0,0 +1,127 @@
from unittest.mock import call, patch
import pendulum
import pytest
import requests
from akkudoktoreos.prediction.pvforecastforecastsolar import PVForecastForecastSolar
def _config(config_eos, planes=None, api_key=None):
settings = {
"general": {"latitude": 52.5, "longitude": 13.4},
"pvforecast": {
"provider": "PVForecastForecastSolar",
"planes": planes
if planes is not None
else [{"surface_tilt": 30.0, "surface_azimuth": 180.0, "peakpower": 5.0}],
"provider_settings": {"PVForecastForecastSolar": {"api_key": api_key}},
},
}
config_eos.merge_settings_from_dict(settings)
return config_eos
@pytest.fixture
def pvforecast_instance(config_eos):
_config(config_eos)
start_dt = pendulum.datetime(2025, 1, 1, tz="Europe/Berlin")
return PVForecastForecastSolar(config=config_eos.load, start_datetime=start_dt)
def _http(watts, timezone="Europe/Berlin"):
return type(
"R",
(),
{
"raise_for_status": lambda self: None,
"json": lambda self: {
"result": {"watts": watts},
"message": {"info": {"timezone": timezone}},
},
},
)()
def test_provider_id(pvforecast_instance):
assert PVForecastForecastSolar.provider_id() == "PVForecastForecastSolar"
assert pvforecast_instance.enabled() is True
def test_update_data_resolves_tz_and_sets_power(pvforecast_instance):
body = {
"timezone": "Europe/Berlin",
"watts": {"2025-01-01 12:00:00": 1200.0, "2025-01-01 13:00:00": 1500.0},
}
with patch.object(pvforecast_instance, "_request_forecast", return_value=body), \
patch.object(PVForecastForecastSolar, "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, 13, 0, tz="Europe/Berlin"),
{"pvforecast_ac_power": 1500.0, "pvforecast_dc_power": 1500.0},
),
]
mock_update.assert_has_calls(expected, any_order=False)
# 12:00 Europe/Berlin (CET) is 11:00 UTC.
assert mock_update.call_args_list[0][0][0].in_timezone("UTC").hour == 11
def test_plane_url_converts_azimuth(config_eos):
"""EOS azimuth 270 (west) -> Forecast.Solar 90; the key + plane geometry land in the URL."""
_config(
config_eos,
planes=[{"surface_tilt": 25.0, "surface_azimuth": 270.0, "peakpower": 7.5}],
api_key="secret",
)
pv = PVForecastForecastSolar(
config=config_eos.load, start_datetime=pendulum.datetime(2025, 1, 1, tz="UTC")
)
with patch("requests.get", return_value=_http({})) as mock_get:
pv._request_forecast(force_update=True)
url = mock_get.call_args[0][0]
assert url == "https://api.forecast.solar/secret/estimate/52.5/13.4/25.0/90.0/7.5"
def test_request_forecast_sums_planes(config_eos):
"""Two planes -> two requests; instantaneous powers are summed per timestamp."""
_config(
config_eos,
planes=[
{"surface_tilt": 30.0, "surface_azimuth": 90.0, "peakpower": 3.0},
{"surface_tilt": 30.0, "surface_azimuth": 270.0, "peakpower": 3.0},
],
)
pv = PVForecastForecastSolar(
config=config_eos.load, start_datetime=pendulum.datetime(2025, 1, 1, tz="UTC")
)
responses = [
_http({"2025-01-01 12:00:00": 1000.0}),
_http({"2025-01-01 12:00:00": 800.0}),
]
with patch("requests.get", side_effect=responses) as mock_get:
body = pv._request_forecast(force_update=True)
assert mock_get.call_count == 2
assert body["watts"]["2025-01-01 12:00:00"] == 1800.0
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(PVForecastForecastSolar, "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 Forecast.Solar" in str(exc_info.value)