Feature/opendpe forecast

custom_components/rtetempo/forecast.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+import datetime
+import logging
+from dataclasses import dataclass
+from typing import List, Optional
+
+import aiohttp
+
+OPEN_DPE_URL = "https://open-dpe.fr/assets/tempo_days_lite.json"
+
+_LOGGER = logging.getLogger(__name__)
+
+
+#   Forecast model
+@dataclass
+class ForecastDay:
+    """Tempo forecast for a given day."""
+
+    date: datetime.date
+    color: str                     # "bleu", "blanc", "rouge" (normalized to lowercase)
+    probability: Optional[float]   # 0.67 for example (for 67%)
+    source: str = "open_dpe"
+
+
+#   Main function (Open-DPE)
+async def async_fetch_opendpe_forecast(
+    session: aiohttp.ClientSession,
+) -> List[ForecastDay]:
+    """Fetch Tempo forecasts from the Open DPE JSON."""
+
+    try:
+        async with session.get(OPEN_DPE_URL, timeout=10) as response:
+            if response.status != 200:
+                _LOGGER.error("Open-DPE: HTTP %s", response.status)
+                return []
+
+            data = await response.json()
+
+    except Exception as exc:
+        _LOGGER.error("Open DPE: erreur lors de la récupération JSON : %s", exc)
+        return []
+
+    forecasts: List[ForecastDay] = []
+
+    for entry in data:
+        try:
+            forecast_date = datetime.datetime.strptime(
+                entry["date"], "%Y-%m-%d"
+            ).date()
+            color = entry.get("couleur", "").lower()
+            prob = entry.get("probability", None)
+
+            forecasts.append(
+                ForecastDay(
+                    date=forecast_date,
+                    color=color,
+                    probability=prob,
+                    source="open_dpe",
+                )
+            )
+
+        except Exception as exc:
+            _LOGGER.warning("Open DPE: ligne ignorée (%s) : %s", exc, entry)
+            continue
+
+    return forecasts

custom_components/rtetempo/forecast_coordinator.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+import logging
+from datetime import timedelta
+from typing import List
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.event import async_track_time_change
+
+from .forecast import ForecastDay, async_fetch_opendpe_forecast
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class ForecastCoordinator(DataUpdateCoordinator[List[ForecastDay]]):
+    """Coordinator in charge of fetching Open-DPE forecasts."""
+
+    def __init__(self, hass: HomeAssistant):
+        """Initializing the coordinator."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            name="Tempo Forecast Coordinator",
+            update_interval=timedelta(hours=6),  # refresh every 6 hours
+        )
+
+        self.hass = hass
+        self.session = async_get_clientsession(hass)
+
+        # Daily uptade after midnight then every 6 hours (JSON is updated around 06:00)
+        async_track_time_change(
+            hass,
+            self._scheduled_refresh,
+            hour=7,
+            minute=0,
+            second=0,
+        )
+
+        _LOGGER.debug(
+            "ForecastCoordinator initialisé : refresh quotidien programmé à 07:00 + intervalle 6h"
+        )
+
+    async def _scheduled_refresh(self, now):
+        """Update at 07:00 every day."""
+        _LOGGER.debug("Open DPE: lancement du refresh programmé à 07:00")
+        await self.async_request_refresh()
+
+    async def _async_update_data(self) -> List[ForecastDay]:
+        """Open DPE data recovery."""
+        try:
+            forecasts = await async_fetch_opendpe_forecast(self.session)
+            _LOGGER.debug("Open DPE: %s jours récupérés", len(forecasts))
+            return forecasts
+
+        except Exception as exc:
+            _LOGGER.error("Open DPE: erreur lors de la mise à jour: %s", exc)
+            raise UpdateFailed(f"Erreur mise à jour des prévisions Open DPE: {exc}")
+

custom_components/rtetempo/sensor.py

@@ -43,6 +43,10 @@
 _LOGGER = logging.getLogger(__name__)
 
 
+# Importing coordinator and sensors for forecast data
+from .forecast_coordinator import ForecastCoordinator
+from .sensor_forecast import OpenDPEForecastSensor
+
 # config flow setup
 async def async_setup_entry(
     hass: HomeAssistant,
@@ -78,10 +82,23 @@ async def async_setup_entry(
         NextCycleTime(config_entry.entry_id),
         OffPeakChangeTime(config_entry.entry_id),
     ]
+
+    #   Add forecast sensors from Open DPE
+    forecast_coordinator = ForecastCoordinator(hass)
+    await forecast_coordinator.async_config_entry_first_refresh()
+
+    NUM_FORECAST_DAYS = 7  # J+1 à J+7
+
+    # Skip index 0 (J+1) because RTE provides the official J+1 sensor
+    for index in range(1, NUM_FORECAST_DAYS):
+        # Text version
+        sensors.append(OpenDPEForecastSensor(forecast_coordinator, index, visual=False))
+        # Visual version (emoji)
+        sensors.append(OpenDPEForecastSensor(forecast_coordinator, index, visual=True))
+
     # Add the entities to HA
     async_add_entities(sensors, True)
 
-
 class CurrentColor(SensorEntity):
     """Current Color Sensor Entity."""
 

custom_components/rtetempo/sensor_forecast.py

@@ -0,0 +1,151 @@
+from __future__ import annotations
+
+from typing import Optional
+from homeassistant.components.sensor import SensorEntity, SensorDeviceClass
+from homeassistant.helpers.entity import DeviceInfo
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import (
+    DOMAIN,
+    DEVICE_MANUFACTURER,
+    DEVICE_MODEL,
+    SENSOR_COLOR_BLUE_EMOJI,
+    SENSOR_COLOR_WHITE_EMOJI,
+    SENSOR_COLOR_RED_EMOJI,
+    SENSOR_COLOR_UNKNOWN_EMOJI,
+    SENSOR_COLOR_BLUE_NAME,
+    SENSOR_COLOR_WHITE_NAME,
+    SENSOR_COLOR_RED_NAME,
+    SENSOR_COLOR_UNKNOWN_NAME,
+)
+
+from .forecast_coordinator import ForecastCoordinator
+from .forecast import ForecastDay
+
+
+# -------- Helper functions (copied from sensor.py of RTE Tempo) ----------------
+
+def get_color_emoji(value: str) -> str:
+    if value == "rouge":
+        return SENSOR_COLOR_RED_EMOJI
+    if value == "blanc":
+        return SENSOR_COLOR_WHITE_EMOJI
+    if value == "bleu":
+        return SENSOR_COLOR_BLUE_EMOJI
+    return SENSOR_COLOR_UNKNOWN_EMOJI
+
+
+def get_color_name(value: str) -> str:
+    if value == "rouge":
+        return SENSOR_COLOR_RED_NAME
+    if value == "blanc":
+        return SENSOR_COLOR_WHITE_NAME
+    if value == "bleu":
+        return SENSOR_COLOR_BLUE_NAME
+    return SENSOR_COLOR_UNKNOWN_NAME
+
+
+def get_color_icon(value: str) -> str:
+    if value == "rouge":
+        return "mdi:alert"
+    if value == "blanc":
+        return "mdi:information-outline"
+    if value == "bleu":
+        return "mdi:check-bold"
+    return "mdi:palette"
+
+
+# ---------------- Forecast Sensor ----------------------
+
+
+class OpenDPEForecastSensor(CoordinatorEntity, SensorEntity):
+    """OpenDPE forecast sensor (text or visual version)."""
+
+    _attr_device_class = SensorDeviceClass.ENUM
+    _attr_has_entity_name = True
+
+    def __init__(self, coordinator: ForecastCoordinator, index: int, visual: bool):
+        super().__init__(coordinator)
+
+        self.index = index
+        self.visual = visual
+
+        # ----- Sensor naming and options -----
+        if visual:
+            self._attr_name = f"OpenDPE J{index + 1} (visuel)"
+            self._attr_unique_id = f"{DOMAIN}_forecast_opendpe_j{index + 1}_emoji"
+            self._attr_options = [
+                SENSOR_COLOR_BLUE_EMOJI,
+                SENSOR_COLOR_WHITE_EMOJI,
+                SENSOR_COLOR_RED_EMOJI,
+                SENSOR_COLOR_UNKNOWN_EMOJI,
+            ]
+            self._attr_icon = "mdi:palette"
+
+        else:
+            self._attr_name = f"OpenDPE J{index + 1}"
+            self._attr_unique_id = f"{DOMAIN}_forecast_opendpe_j{index + 1}"
+            self._attr_options = [
+                SENSOR_COLOR_BLUE_NAME,
+                SENSOR_COLOR_WHITE_NAME,
+                SENSOR_COLOR_RED_NAME,
+                SENSOR_COLOR_UNKNOWN_NAME,
+            ]
+            self._attr_icon = "mdi:calendar"
+
+        self._attr_native_value: Optional[str] = None
+        self._attr_extra_state_attributes = {}
+
+    # ---------------- Device Info ----------------------
+
+    @property
+    def device_info(self) -> DeviceInfo:
+        """Return device info shared by all forecast sensors."""
+        return DeviceInfo(
+            identifiers={(DOMAIN, "forecast")},
+            name="RTE Tempo Forecast",
+            manufacturer=DEVICE_MANUFACTURER,
+            model=DEVICE_MODEL,
+        )
+
+    # ---------------- Availability ----------------------
+
+    @property
+    def available(self) -> bool:
+        data = self.coordinator.data
+        return data is not None and len(data) > self.index
+
+    # ---------------- Coordinator update handler ----------------------
+
+    def _handle_coordinator_update(self) -> None:
+        data = self.coordinator.data
+
+        if not data or len(data) <= self.index:
+            self._attr_native_value = None
+            self._attr_extra_state_attributes = {}
+            self.async_write_ha_state()
+            return
+
+        forecast: ForecastDay = data[self.index]
+        color = forecast.color.lower()
+
+        if color not in ["bleu", "blanc", "rouge"]:
+            color = "inconnu"
+
+        # ----- VISUAL VERSION -----
+        if self.visual:
+            self._attr_native_value = get_color_emoji(color)
+            self._attr_icon = get_color_icon(color)
+
+        # ----- TEXT VERSION -----
+        else:
+            self._attr_native_value = get_color_name(color)
+
+        # Extra attributes for both sensors
+        self._attr_extra_state_attributes = {
+            "date": forecast.date.isoformat(),
+            "probability": forecast.probability,
+            "attribution": "Données Tempo : Open DPE (https://open-dpe.fr)",
+        }
+
+        self.async_write_ha_state()