feat: accept an external consumption forecast as a time-series entity

CHANGELOG.md

@@ -6,6 +6,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
 ## [10.0.1] - 2026-08-02
 
+### Added
+
+- New `ha_consumption_series` consumption-forecast strategy reading a user-authored Home Assistant time-series entity. ([#428](https://github.com/johanzander/bess-manager/issues/428))
+
 ### Fixed
 
 - `HuaweiController.sync_soc_limits` now reads before writing SOC limits, instead of writing unconditionally on every sync. ([#427](https://github.com/johanzander/bess-manager/issues/427))

core/bess/battery_system_manager.py

@@ -1049,7 +1049,13 @@ def get_consumption_forecast_comparison(self) -> dict:
             hours without complete actual data).
         """
         active_strategy = self.home_settings.consumption_strategy
-        strategy_names = ["sensor", "fixed", "influxdb_7d_avg", "ha_statistics"]
+        strategy_names = [
+            "sensor",
+            "fixed",
+            "influxdb_7d_avg",
+            "ha_statistics",
+            "ha_consumption_series",
+        ]
         results = []
 
         for name in strategy_names:
@@ -1073,6 +1079,8 @@ def get_consumption_forecast_comparison(self) -> dict:
                     forecast = self._get_influxdb_7d_avg_forecast()
                 elif name == "ha_statistics":
                     forecast = self._get_ha_statistics_forecast()
+                elif name == "ha_consumption_series":
+                    forecast = self.controller.get_consumption_forecast_series()
                 else:
                     continue
 
@@ -1135,6 +1143,12 @@ def _get_consumption_forecast(self) -> list[float]:
         if strategy == "influxdb_7d_avg":
             return self._get_influxdb_7d_avg_forecast()
 
+        if strategy == "ha_consumption_series":
+            # No fallback on failure: this strategy's whole purpose is to let
+            # the user express a shaped load; silently degrading to a flat
+            # profile would hide exactly the thing they configured it for.
+            return self.controller.get_consumption_forecast_series()
+
         if strategy == "ha_statistics":
             # Data-insufficiency or missing-sensor errors are handled the same
             # way: fall back to fixed until the situation resolves.
@@ -1688,6 +1702,36 @@ def _get_current_battery_soc(self) -> float | None:
             logger.error(f"Failed to get battery SOC: {e}")
             return None
 
+    def _extend_consumption_predictions(
+        self, consumption_predictions: list[float], period_count: int
+    ) -> list[float]:
+        """Extend today's consumption predictions to cover `period_count` periods.
+
+        Every strategy except `ha_consumption_series` repeats today's
+        uniform pattern for the tomorrow-spanning periods — fine when the
+        forecast is already flat or a time-of-day profile with no real
+        "tomorrow" data. `ha_consumption_series` has genuine tomorrow data
+        (the same entity's `raw_tomorrow` attribute), so it extends with
+        that instead, mirroring how solar already fetches tomorrow's real
+        forecast rather than repeating today's.
+        """
+        if period_count <= len(consumption_predictions):
+            return consumption_predictions
+
+        if self.home_settings.consumption_strategy == "ha_consumption_series":
+            tomorrow_consumption = (
+                self.controller.get_consumption_forecast_series_tomorrow()
+            )
+        else:
+            tomorrow_consumption = consumption_predictions.copy()
+
+        extended = (consumption_predictions + tomorrow_consumption)[:period_count]
+        logger.info(
+            "Extended consumption predictions to %d periods for tomorrow horizon",
+            len(extended),
+        )
+        return extended
+
     def _fetch_tomorrow_solar_forecast(self) -> list[float]:
         """Fetch tomorrow's solar forecast, falling back to zeros if unavailable."""
         try:
@@ -1734,14 +1778,9 @@ def _gather_optimization_data(
             solar_predictions = self.controller.get_solar_forecast()
 
         # --- Extend arrays to match period_count when horizon spans tomorrow ---
-        if period_count > len(consumption_predictions):
-            # Consumption: repeat today's uniform pattern for tomorrow
-            tomorrow_consumption = consumption_predictions.copy()
-            consumption_predictions = consumption_predictions + tomorrow_consumption
-            logger.info(
-                "Extended consumption predictions to %d periods for tomorrow horizon",
-                len(consumption_predictions),
-            )
+        consumption_predictions = self._extend_consumption_predictions(
+            consumption_predictions, period_count
+        )
 
         if period_count > len(solar_predictions):
             if prepare_next_day:

core/bess/exceptions.py

@@ -44,6 +44,14 @@ def __init__(self, message: str | None = None):
         super().__init__(message or "HA Statistics data is not available")
 
 
+class ConsumptionForecastUnavailableError(BESSException):
+    """Raised when the ha_consumption_series entity is unconfigured, missing,
+    stale, malformed, or provides an unsupported record interval."""
+
+    def __init__(self, message: str | None = None):
+        super().__init__(message or "Consumption forecast series is not available")
+
+
 class HistoricalDataUnavailableError(BESSException):
     """Raised when InfluxDB historical energy-flow data is unavailable.
 

core/bess/ha_api_controller.py

@@ -10,12 +10,14 @@
 import ssl
 import time
 import urllib.parse
+from datetime import datetime, timedelta
 from typing import ClassVar
 
 import requests
 import websocket
 
-from .exceptions import SystemConfigurationError
+from . import time_utils
+from .exceptions import ConsumptionForecastUnavailableError, SystemConfigurationError
 from .runtime_failure_tracker import RuntimeFailureTracker
 
 logger = logging.getLogger(__name__)
@@ -236,6 +238,13 @@ def __init__(
             "precision": 1,
             "conversion_threshold": 1000,
         },
+        "get_consumption_forecast_series": {
+            "sensor_key": "consumption_forecast_series",
+            "name": "Consumption Forecast Series",
+            "unit": "list",
+            "precision": 1,
+            "conversion_threshold": None,
+        },
         # Solar forecast
         "get_solar_forecast": {
             "sensor_key": "solar_forecast_today",
@@ -1256,6 +1265,139 @@ def get_estimated_consumption(self):
         # Return 96 quarterly periods (24 hours * 4 quarters per hour)
         return [quarterly_consumption] * 96
 
+    def get_consumption_forecast_series(self) -> list[float]:
+        """Get today's consumption forecast from a user-authored HA time-series entity.
+
+        Mirrors the price-manager entity pattern: one entity, `raw_today` /
+        `raw_tomorrow` attributes holding timestamped {start, value} records
+        (value in kWh for that record's interval). Unlike the flat `sensor`
+        strategy this can express a shaped load (a known EV session, a
+        weather-driven aircon block, etc.) because the user's own HA
+        template — not BESS — builds the series; BESS only normalizes it
+        onto the DP's quarter-hour grid.
+
+        Returns:
+            list[float]: consumption values in kWh per quarter-hour period
+                (92-100 periods depending on DST).
+
+        Raises:
+            ConsumptionForecastUnavailableError: entity not configured,
+                missing, stale, malformed, short of the horizon, or using
+                an unsupported record interval. Never silently degrades to
+                a flat or fixed profile.
+        """
+        return self._parse_consumption_series_for_date(time_utils.today())
+
+    def get_consumption_forecast_series_tomorrow(self) -> list[float]:
+        """Get tomorrow's consumption forecast from the same series entity.
+
+        See `get_consumption_forecast_series` — reads `raw_tomorrow` instead
+        of `raw_today` from the same entity.
+        """
+        tomorrow = time_utils.today() + timedelta(days=1)
+        return self._parse_consumption_series_for_date(tomorrow)
+
+    def _parse_consumption_series_for_date(self, target_date) -> list[float]:
+        """Fetch and normalize the consumption series entity for `target_date`."""
+        sensor_key = "consumption_forecast_series"
+        entity_id = self.sensors.get(sensor_key)
+        if not entity_id:
+            raise ConsumptionForecastUnavailableError(
+                f"Consumption forecast series sensor '{sensor_key}' not configured"
+            )
+
+        response = self._api_request(
+            "get",
+            f"/api/states/{entity_id}",
+            operation="Get consumption forecast series",
+            category="sensor_read",
+        )
+        if not response or "attributes" not in response:
+            raise ConsumptionForecastUnavailableError(
+                f"No attributes found for consumption forecast series sensor {entity_id}"
+            )
+
+        attributes = response["attributes"]
+        raw_key = "raw_today" if target_date == time_utils.today() else "raw_tomorrow"
+        raw_data = attributes.get(raw_key)
+        if not raw_data:
+            raise ConsumptionForecastUnavailableError(
+                f"No '{raw_key}' data found on consumption forecast series sensor {entity_id}"
+            )
+
+        return self._normalize_consumption_records(raw_data, target_date, entity_id)
+
+    def _normalize_consumption_records(
+        self, raw_data: list, target_date, entity_id: str
+    ) -> list[float]:
+        """Parse timestamped {start, value} records and normalize to quarter-hour periods.
+
+        Accepts 15-minute (native) or 60-minute (upsampled /4) record
+        spacing. Any other interval, a date mismatch, or a malformed record
+        is an explicit failure — never a silent guess at the missing data.
+        """
+        try:
+            entries = sorted(raw_data, key=lambda entry: str(entry.get("start", "")))
+            starts = []
+            for entry in entries:
+                start_str = entry.get("start")
+                if not start_str:
+                    raise ConsumptionForecastUnavailableError(
+                        f"Consumption forecast series entry missing 'start' on {entity_id}"
+                    )
+                if "value" not in entry:
+                    raise ConsumptionForecastUnavailableError(
+                        f"Consumption forecast series entry missing 'value' on {entity_id}"
+                    )
+                if isinstance(start_str, str) and start_str.endswith(
+                    ("+02:00", "+01:00")
+                ):
+                    start_str = start_str[:-6]
+                starts.append(datetime.fromisoformat(start_str))
+        except (ValueError, TypeError) as e:
+            raise ConsumptionForecastUnavailableError(
+                f"Malformed consumption forecast series data on {entity_id}: {e}"
+            ) from e
+
+        if starts[0].date() != target_date:
+            raise ConsumptionForecastUnavailableError(
+                f"Consumption forecast series on {entity_id} is stale: first record "
+                f"is for {starts[0].date()}, expected {target_date}"
+            )
+
+        interval_minutes = (
+            round((starts[1] - starts[0]).total_seconds() / 60)
+            if len(starts) > 1
+            else 15
+        )
+
+        if interval_minutes == 15:
+            periods_per_record = 1
+            divisor = 1.0
+        elif interval_minutes == 60:
+            periods_per_record = 4
+            divisor = 4.0
+        else:
+            raise ConsumptionForecastUnavailableError(
+                f"Consumption forecast series on {entity_id} uses an unsupported "
+                f"record interval of {interval_minutes} minutes (only 15 or 60 "
+                "minute spacing is supported)"
+            )
+
+        periods: list[float] = []
+        for entry in entries:
+            period_value = float(entry["value"]) / divisor
+            periods.extend([period_value] * periods_per_record)
+
+        if not (92 <= len(periods) <= 100):
+            raise ConsumptionForecastUnavailableError(
+                f"Consumption forecast series on {entity_id} has {len(periods)} "
+                f"periods after normalization for {target_date}; expected 92-100 "
+                "(quarter-hour resolution, DST-adjusted)"
+            )
+
+        return periods
+
     def get_ha_config(self) -> dict:
         """Fetch Home Assistant configuration (timezone, location, etc.)."""
         response = self._api_request(

core/bess/sensor_collector.py

@@ -844,11 +844,14 @@ def check_prediction_health(self, consumption_strategy: str = "sensor") -> dict:
 
         Only validates the ``get_estimated_consumption`` sensor when the
         ``sensor`` strategy is active — other strategies (fixed,
-        influxdb_7d_avg, ha_statistics) do not rely on that HA sensor.
+        influxdb_7d_avg, ha_statistics, ha_consumption_series) do not rely
+        on that HA sensor.
         """
         all_methods = ["get_solar_forecast"]
         if consumption_strategy == "sensor":
             all_methods = ["get_estimated_consumption", *all_methods]
+        elif consumption_strategy == "ha_consumption_series":
+            all_methods = ["get_consumption_forecast_series", *all_methods]
 
         return perform_health_check(
             component_name="Energy Prediction",

core/bess/settings_store.py

@@ -49,6 +49,7 @@
         "solar_forecast_today",
         "solar_forecast_tomorrow",
         "48h_avg_grid_import",
+        "consumption_forecast_series",
         "current_l1",
         "current_l2",
         "current_l3",
@@ -80,6 +81,7 @@
         "solar_forecast_today",
         "solar_forecast_tomorrow",
         "48h_avg_grid_import",
+        "consumption_forecast_series",
         "current_l1",
         "current_l2",
         "current_l3",