feat: add cloud PV forecast providers: pvnode.com, forecast-solar and solcast (#1150)

This PR adds three native PV power forecast providers, giving operators more
cloud forecast sources to choose from via pvforecast.provider, alongside the
existing PVForecastAkkudoktor, PVForecastVrm and PVForecastImport:

    PVForecastPVNode — native 15-minute forecasts from the pvnode.com
    V2 API. Saved-site mode (GET /v2/forecast/{site_id}) where the operator enters their API key +
    site id, or inline mode (POST /v2/forecast/inline) using the configured planes.
    PVForecastForecastSolar — the free Forecast.Solar API (no key
    required for the public endpoint). Multi-plane systems issue one request per plane and the
    instantaneous powers are summed per timestamp.
    PVForecastSolcast — the Solcast rooftop-site API (API key + resource id).

Implementation notes

    All three populate the existing pvforecast_ac_power (and mirror pvforecast_dc_power)
    prediction keys, so they slot into the optimizer unchanged.
    Timezone handling: pvnode and Forecast.Solar return site-local wall-clock timestamps with an
    IANA timezone field, which are resolved to absolute instants before resampling; Solcast returns
    UTC period_end and is normalised to the period start.
    Each provider follows the existing provider pattern (provider_id(), _request_forecast() with
    @cache_in_file, _update_data()), is registered in pvforecast.py and prediction.py, and is
    upstream-neutral.

Validation

    27 unit tests (timezone resolution, null/zero handling, kW→W and period-start conversion,
    azimuth conversion, multi-plane summation, request URL/auth, HTTP-error handling).
    ruff check (F/D/S/bandit) and ruff format clean.
    Each provider was additionally validated against its live API with a real plant, confirming
    the response shapes (pvnode: 288 native 15-min slots; Forecast.Solar: instantaneous watts;
    Solcast: kW estimates with period_end/PT30M).

Documentation

    Provider descriptions and configuration examples added to docs/akkudoktoreos/prediction.md.
    CHANGELOG.md entry under Unreleased.
    Regenerated docs/_generated/configpvforecast.md and configexample.md.

Notes for reviewers

Forecast.Solar's free endpoint is rate-limited (12 req/hour) and Solcast's free tier limits daily
calls; both providers rely on the standard 1-hour cache_in_file TTL to stay within budget.

Authors:

The code is created by Christin. Only minor adaptions by Bobby.

Signed-off-by: Christin <info@bikinibottom.capital>
Signed-off-by: Bobby Noelte <b0661n0e17e@gmail.com>
Co-authored-by: Christin <info@bikinibottom.capital>

src/akkudoktoreos/prediction/prediction.py

@@ -45,7 +45,10 @@ 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.pvforecastsolcast import PVForecastSolcast
 from akkudoktoreos.prediction.pvforecastvrm import PVForecastVrm
 from akkudoktoreos.prediction.weatherbrightsky import WeatherBrightSky
 from akkudoktoreos.prediction.weatherclearoutside import WeatherClearOutside
@@ -84,6 +87,9 @@ loadforecast_vrm = LoadVrm()
 loadforecast_import = LoadImport()
 pvforecast_akkudoktor = PVForecastAkkudoktor()
 pvforecast_vrm = PVForecastVrm()
+pvforecast_pvnode = PVForecastPVNode()
+pvforecast_forecastsolar = PVForecastForecastSolar()
+pvforecast_solcast = PVForecastSolcast()
 pvforecast_import = PVForecastImport()
 weather_brightsky = WeatherBrightSky()
 weather_clearoutside = WeatherClearOutside()
@@ -105,6 +111,9 @@ def prediction_providers() -> list[
         LoadImport,
         PVForecastAkkudoktor,
         PVForecastVrm,
+        PVForecastPVNode,
+        PVForecastForecastSolar,
+        PVForecastSolcast,
         PVForecastImport,
         WeatherBrightSky,
         WeatherClearOutside,
@@ -129,6 +138,9 @@ def prediction_providers() -> list[
         loadforecast_import, \
         pvforecast_akkudoktor, \
         pvforecast_vrm, \
+        pvforecast_pvnode, \
+        pvforecast_forecastsolar, \
+        pvforecast_solcast, \
         pvforecast_import, \
         weather_brightsky, \
         weather_clearoutside, \
@@ -149,6 +161,9 @@ def prediction_providers() -> list[
         loadforecast_import,
         pvforecast_akkudoktor,
         pvforecast_vrm,
+        pvforecast_pvnode,
+        pvforecast_forecastsolar,
+        pvforecast_solcast,
         pvforecast_import,
         weather_brightsky,
         weather_clearoutside,
@@ -174,6 +189,9 @@ class Prediction(PredictionContainer):
             LoadImport,
             PVForecastAkkudoktor,
             PVForecastVrm,
+            PVForecastPVNode,
+            PVForecastForecastSolar,
+            PVForecastSolcast,
             PVForecastImport,
             WeatherBrightSky,
             WeatherClearOutside,

src/akkudoktoreos/prediction/pvforecast.py

@@ -7,7 +7,12 @@ 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.pvforecastsolcast import PVForecastSolcastCommonSettings
 from akkudoktoreos.prediction.pvforecastvrm import PVForecastVrmCommonSettings
 
 
@@ -18,7 +23,14 @@ def pvforecast_provider_ids() -> list[str]:
     except:
         # Prediction may not be initialized
         # Return at least provider used in example
-        return ["PVForecastAkkudoktor", "PVForecastImport", "PVForecastVrm"]
+        return [
+            "PVForecastAkkudoktor",
+            "PVForecastImport",
+            "PVForecastVrm",
+            "PVForecastPVNode",
+            "PVForecastForecastSolar",
+            "PVForecastSolcast",
+        ]
 
     return [
         provider.provider_id()
@@ -179,6 +191,18 @@ class PVForecastCommonProviderSettings(SettingsBaseModel):
         default=None,
         json_schema_extra={"description": "PVForecastVrm settings", "examples": [None]},
     )
+    PVForecastPVNode: Optional[PVForecastPVNodeCommonSettings] = Field(
+        default=None,
+        json_schema_extra={"description": "PVForecastPVNode settings", "examples": [None]},
+    )
+    PVForecastForecastSolar: Optional[PVForecastForecastSolarCommonSettings] = Field(
+        default=None,
+        json_schema_extra={"description": "PVForecastForecastSolar settings", "examples": [None]},
+    )
+    PVForecastSolcast: Optional[PVForecastSolcastCommonSettings] = Field(
+        default=None,
+        json_schema_extra={"description": "PVForecastSolcast settings", "examples": [None]},
+    )
 
 
 class PVForecastCommonSettings(SettingsBaseModel):

src/akkudoktoreos/prediction/pvforecastforecastsolar.py

@@ -0,0 +1,163 @@
+"""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}
+
+    async 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)
+            await 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__":
+    import asyncio
+
+    pv = PVForecastForecastSolar()
+    asyncio.run(pv._update_data())

src/akkudoktoreos/prediction/pvforecastpvnode.py

@@ -0,0 +1,244 @@
+"""Retrieves PV forecast data from the pvnode.com V2 API.
+
+pvnode.com delivers native 15-minute PV power forecasts. Two request modes,
+decided by configuration:
+
+* ``site_id`` set  -> ``GET /v2/forecast/{site_id}`` — a saved (and possibly
+  calibrated) site managed in the pvnode web app. This is the operator's primary
+  path: register the plant once on pvnode.com, then enter the site id + API key.
+* ``site_id`` empty -> ``POST /v2/forecast/inline`` — geometry is sent inline from
+  the configured ``pvforecast.planes`` (works without any web-app setup).
+
+V2 response timestamps are SITE-LOCAL wall-clock (no offset) accompanied by an
+IANA ``timezone`` field. We resolve them to absolute instants here so the rest of
+EOS keeps working in its own timezone. ``pv_power`` is nullable (e.g. at night) —
+null is treated as 0 W so the optimizer's linear resampling does not interpolate
+phantom production across the night.
+
+Notes:
+    - Requires ``pvforecast.provider_settings.PVForecastPVNode.api_key`` (Bearer auth).
+    - API: https://api.pvnode.com/v2  (15-minute resolution).
+"""
+
+import re
+import urllib.parse
+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
+
+PVNODE_BASE = "https://api.pvnode.com/v2"
+
+_TZ_SUFFIX = re.compile(r"([zZ]|[+-]\d\d:?\d\d)$")
+
+
+class PVForecastPVNodeCommonSettings(SettingsBaseModel):
+    """Common settings for the pvnode.com PV forecast provider."""
+
+    api_key: str = Field(
+        default="",
+        json_schema_extra={
+            "description": "pvnode.com API key (Bearer auth). Required.",
+            "examples": ["pvn_live_xxxxxxxxxxxxxxxx"],
+        },
+    )
+    site_id: Optional[str] = Field(
+        default=None,
+        json_schema_extra={
+            "description": (
+                "pvnode.com site id of the saved plant ('Anlagen-ID'). When set, the "
+                "saved (possibly calibrated) site is used. Leave empty to send the "
+                "configured pvforecast.planes inline instead."
+            ),
+            "examples": ["abcd-1234"],
+        },
+    )
+    forecast_days: int = Field(
+        default=2,
+        ge=1,
+        le=7,
+        json_schema_extra={
+            "description": "Forecast horizon in days (1-7, capped by the pvnode plan).",
+            "examples": [2],
+        },
+    )
+
+
+class PVForecastPVNode(PVForecastProvider):
+    """Fetch and process PV forecast data from the pvnode.com V2 API."""
+
+    @classmethod
+    def provider_id(cls) -> str:
+        """Return the unique identifier for the PV-Forecast-Provider."""
+        return "PVForecastPVNode"
+
+    @property
+    def _settings(self) -> PVForecastPVNodeCommonSettings:
+        settings = self.config.pvforecast.provider_settings.PVForecastPVNode
+        if settings is None:
+            settings = PVForecastPVNodeCommonSettings()
+        return settings
+
+    def _to_utc_datetime(self, local_ts: Any, iana_tz: Optional[str]) -> Any:
+        """Resolve a pvnode V2 wall-clock timestamp to a timezone-aware datetime.
+
+        V2 timestamps are local wall-clock without offset (e.g. "2026-06-22T14:00:00")
+        plus a response-level IANA ``timezone``. If the string already carries an
+        explicit offset or 'Z' it is trusted as-is.
+        """
+        s = str(local_ts).strip()
+        if _TZ_SUFFIX.search(s):
+            # Already absolute (offset or Z present) — parse as-is.
+            return to_datetime(s)
+        tz = iana_tz or str(self.config.general.timezone)
+        # Interpret the naive wall-clock string AS local time in tz, then resolve.
+        dt = pendulum.parse(s, tz=tz)
+        return to_datetime(dt.isoformat())
+
+    def _extract_values(self, body: Any) -> list[tuple[Any, float]]:
+        """Extract (datetime, power_w) rows from a pvnode V2 response body.
+
+        Canonical shape: ``{"timezone": ..., "values": [{"timestamp", "pv_power"}, ...]}``.
+        Tolerant of edge/legacy shapes (mirrors the production DVhub client).
+        """
+        tz: Optional[str] = None
+        arr: Any = None
+        if isinstance(body, list):
+            arr = body
+        elif isinstance(body, dict):
+            tz = body.get("timezone") if isinstance(body.get("timezone"), str) else None
+            for key in ("values", "forecasts", "data", "forecast"):
+                if isinstance(body.get(key), list):
+                    arr = body[key]
+                    break
+        if not isinstance(arr, list):
+            return []
+
+        rows: list[tuple[Any, float]] = []
+        for entry in arr:
+            if not isinstance(entry, dict):
+                continue
+            ts = (
+                entry.get("timestamp")
+                or entry.get("time")
+                or entry.get("ts")
+                or entry.get("ts_utc")
+                or entry.get("datetime")
+            )
+            if ts is None:
+                continue
+            # pv_power is nullable (night) -> treat missing as 0 W, not a gap, so the
+            # optimizer's linear resampling does not interpolate across the night.
+            raw_power = entry.get("pv_power")
+            if raw_power is None:
+                raw_power = entry.get("power_w")
+            if raw_power is None:
+                raw_power = entry.get("power")
+            if raw_power is None:
+                raw_power = entry.get("watts")
+            power = 0.0 if raw_power is None else float(raw_power)
+            try:
+                date = self._to_utc_datetime(ts, tz)
+            except Exception as e:  # noqa: BLE001 - skip unparseable rows
+                logger.warning(f"pvnode: skipping unparseable timestamp {ts!r}: {e}")
+                continue
+            rows.append((date, round(power, 1)))
+        return rows
+
+    @cache_in_file(with_ttl="1 hour")
+    def _request_forecast(self) -> Any:
+        """Fetch the PV forecast from pvnode.com (saved site or inline planes)."""
+        settings = self._settings
+        api_key = settings.api_key
+        if not api_key:
+            raise ValueError("PVForecastPVNode requires pvforecast...PVForecastPVNode.api_key")
+
+        headers = {"Authorization": f"Bearer {api_key}", "Accept": "application/json"}
+        params = {"forecast_days": str(settings.forecast_days)}
+        site_id = (settings.site_id or "").strip()
+
+        try:
+            if site_id:
+                url = f"{PVNODE_BASE}/forecast/{urllib.parse.quote(site_id, safe='')}"
+                response = requests.get(url, headers=headers, params=params, timeout=30)
+            else:
+                body = self._inline_body()
+                url = f"{PVNODE_BASE}/forecast/inline"
+                headers["Content-Type"] = "application/json"
+                response = requests.post(url, headers=headers, params=params, json=body, timeout=30)
+            logger.debug(f"Requesting pvnode forecast: {url}")
+            response.raise_for_status()
+        except requests.RequestException as e:
+            logger.error(f"Failed to fetch pvforecast from pvnode: {e}")
+            raise RuntimeError("Failed to fetch pvforecast from pvnode API") from e
+
+        self.update_datetime = to_datetime(in_timezone=self.config.general.timezone)
+        return response.json()
+
+    def _inline_body(self) -> dict:
+        """Build the inline-mode request body from latitude/longitude + planes."""
+        latitude = self.config.general.latitude
+        longitude = self.config.general.longitude
+        if latitude is None or longitude is None:
+            raise ValueError(
+                "PVForecastPVNode inline mode needs general.latitude/longitude "
+                "(or set pvforecast...PVForecastPVNode.site_id)"
+            )
+        planes = self.config.pvforecast.planes or []
+        strings = []
+        for plane in planes:
+            tilt = getattr(plane, "surface_tilt", None)
+            azimuth = getattr(plane, "surface_azimuth", None)
+            peakpower = getattr(plane, "peakpower", None)
+            if peakpower is None or tilt is None or azimuth is None:
+                continue
+            strings.append(
+                {
+                    "slope": float(tilt),
+                    # pvnode V2 azimuth convention (0=N, 90=E, 180=S, 270=W) matches
+                    # EOS surface_azimuth, so it is forwarded unchanged.
+                    "orientation": float(azimuth),
+                    "power_kw": float(peakpower),
+                }
+            )
+        if not strings:
+            raise ValueError(
+                "PVForecastPVNode inline mode needs at least one pvforecast.planes "
+                "entry with peakpower, surface_tilt and surface_azimuth"
+            )
+        return {"latitude": float(latitude), "longitude": float(longitude), "strings": strings}
+
+    async def _update_data(self, force_update: Optional[bool] = False) -> None:
+        """Update forecast data in the PVForecastDataRecord format."""
+        if not self.enabled():
+            logger.info("PVForecastPVNode is disabled, skipping update.")
+            return
+
+        body = self._request_forecast(force_update=force_update)  # type: ignore[call-arg]
+        rows = self._extract_values(body)
+
+        for date, power_w in rows:
+            # pvnode returns the plant's expected output power; feed it as AC power
+            # (the key the optimizer reads) and mirror it to DC for reporting.
+            await self.update_value(
+                date,
+                {"pvforecast_ac_power": power_w, "pvforecast_dc_power": power_w},
+            )
+
+        logger.debug(f"Updated pvforecast from pvnode with {len(rows)} entries.")
+        self.update_datetime = to_datetime(in_timezone=self.config.general.timezone)
+
+
+# Example usage
+if __name__ == "__main__":
+    import asyncio
+
+    pv = PVForecastPVNode()
+    asyncio.run(pv._update_data())

src/akkudoktoreos/prediction/pvforecastsolcast.py

@@ -0,0 +1,144 @@
+"""Retrieves PV forecast data from the Solcast API.
+
+Solcast (https://solcast.com) provides high-accuracy PV forecasts for a rooftop
+site that the operator registers in the Solcast web app. The operator enters the
+API key + the rooftop resource id (site id):
+
+    GET https://api.solcast.com.au/rooftop_sites/{site_id}/forecasts?format=json&hours=72
+
+Each forecast row carries ``pv_estimate`` (in kW) and ``period_end`` (UTC) plus
+an ISO-8601 ``period`` duration. The estimate is converted to watts and the
+timestamp is normalised to the period START (``period_end - period``) so it sits
+on the same axis EOS resamples onto.
+
+Notes:
+    - Solcast's free tier limits the number of calls per day; the response is
+      cached (1 hour TTL) to stay within budget.
+"""
+
+import re
+import urllib.parse
+from typing import Any, Optional
+
+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
+
+SOLCAST_BASE = "https://api.solcast.com.au/rooftop_sites"
+
+_PERIOD_RE = re.compile(r"^PT(?:(\d+)H)?(?:(\d+)M)?$")
+
+
+class PVForecastSolcastCommonSettings(SettingsBaseModel):
+    """Common settings for the Solcast PV forecast provider."""
+
+    api_key: str = Field(
+        default="",
+        json_schema_extra={
+            "description": "Solcast API key (Bearer auth). Required.",
+            "examples": ["your-solcast-key"],
+        },
+    )
+    site_id: str = Field(
+        default="",
+        json_schema_extra={
+            "description": "Solcast rooftop site (resource) id. Required.",
+            "examples": ["abcd-1234-efgh-5678"],
+        },
+    )
+
+
+class PVForecastSolcast(PVForecastProvider):
+    """Fetch and process PV forecast data from the Solcast API."""
+
+    @classmethod
+    def provider_id(cls) -> str:
+        """Return the unique identifier for the PV-Forecast-Provider."""
+        return "PVForecastSolcast"
+
+    @property
+    def _settings(self) -> PVForecastSolcastCommonSettings:
+        settings = self.config.pvforecast.provider_settings.PVForecastSolcast
+        if settings is None:
+            settings = PVForecastSolcastCommonSettings()
+        return settings
+
+    @staticmethod
+    def _period_minutes(period: Optional[str]) -> int:
+        """Parse an ISO-8601 period like 'PT30M' or 'PT1H' into minutes (0 if unknown)."""
+        match = _PERIOD_RE.match(str(period or ""))
+        if not match:
+            return 0
+        hours = int(match.group(1) or 0)
+        minutes = int(match.group(2) or 0)
+        return hours * 60 + minutes
+
+    @cache_in_file(with_ttl="1 hour")
+    def _request_forecast(self) -> Any:
+        """Fetch the rooftop-site forecast from Solcast."""
+        settings = self._settings
+        if not settings.api_key or not settings.site_id:
+            raise ValueError("PVForecastSolcast requires api_key and site_id")
+
+        url = f"{SOLCAST_BASE}/{urllib.parse.quote(settings.site_id, safe='')}/forecasts"
+        params = {"format": "json", "hours": "72"}
+        headers = {"Authorization": f"Bearer {settings.api_key}", "Accept": "application/json"}
+        logger.debug(f"Requesting Solcast forecast: {url}")
+        try:
+            response = requests.get(url, headers=headers, params=params, timeout=30)
+            response.raise_for_status()
+        except requests.RequestException as e:
+            logger.error(f"Failed to fetch pvforecast from Solcast: {e}")
+            raise RuntimeError("Failed to fetch pvforecast from Solcast API") from e
+
+        self.update_datetime = to_datetime(in_timezone=self.config.general.timezone)
+        return response.json()
+
+    async def _update_data(self, force_update: Optional[bool] = False) -> None:
+        """Update forecast data in the PVForecastDataRecord format."""
+        if not self.enabled():
+            logger.info("PVForecastSolcast is disabled, skipping update.")
+            return
+
+        body = self._request_forecast(force_update=force_update)  # type: ignore[call-arg]
+        forecasts = body.get("forecasts", []) if isinstance(body, dict) else []
+
+        count = 0
+        for entry in forecasts:
+            if not isinstance(entry, dict):
+                continue
+            estimate_kw = entry.get("pv_estimate")
+            if estimate_kw is None:
+                estimate_kw = entry.get("pv_estimate_period")
+            period_end = entry.get("period_end")
+            if estimate_kw is None or period_end is None:
+                continue
+            try:
+                end = to_datetime(period_end)
+            except Exception as e:  # noqa: BLE001 - skip unparseable rows
+                logger.warning(f"Solcast: skipping unparseable period_end {period_end!r}: {e}")
+                continue
+            minutes = self._period_minutes(entry.get("period"))
+            date = end.subtract(minutes=minutes) if minutes else end
+            power_w = round(float(estimate_kw) * 1000.0, 1)
+            await self.update_value(
+                date,
+                {"pvforecast_ac_power": power_w, "pvforecast_dc_power": power_w},
+            )
+            count += 1
+
+        logger.debug(f"Updated pvforecast from Solcast with {count} entries.")
+        self.update_datetime = to_datetime(in_timezone=self.config.general.timezone)
+
+
+# Example usage
+if __name__ == "__main__":
+    import asyncio
+
+    pv = PVForecastSolcast()
+    asyncio.run(pv._update_data())