diff --git a/custom_components/rtetempo/forecast.py b/custom_components/rtetempo/forecast.py new file mode 100644 index 0000000..d7a9f65 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/custom_components/rtetempo/forecast_coordinator.py b/custom_components/rtetempo/forecast_coordinator.py new file mode 100644 index 0000000..2b4454d --- /dev/null +++ b/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}") + diff --git a/custom_components/rtetempo/sensor.py b/custom_components/rtetempo/sensor.py index 606826d..6a65f9c 100644 --- a/custom_components/rtetempo/sensor.py +++ b/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.""" diff --git a/custom_components/rtetempo/sensor_forecast.py b/custom_components/rtetempo/sensor_forecast.py new file mode 100644 index 0000000..d7f05af --- /dev/null +++ b/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()