Add coordinator to dwd_weather_warnings (#93720)

* Add data coordinator

* Specify coordinator data type

* Apply code review changes

.coveragerc

@@ -227,6 +227,7 @@ omit =
     homeassistant/components/dunehd/__init__.py
     homeassistant/components/dunehd/media_player.py
     homeassistant/components/dwd_weather_warnings/const.py
+    homeassistant/components/dwd_weather_warnings/coordinator.py
     homeassistant/components/dwd_weather_warnings/sensor.py
     homeassistant/components/dweet/*
     homeassistant/components/ebox/sensor.py

homeassistant/components/dwd_weather_warnings/__init__.py

@@ -8,16 +8,20 @@ from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
 
 from .const import CONF_REGION_IDENTIFIER, DOMAIN, PLATFORMS
+from .coordinator import DwdWeatherWarningsCoordinator
 
 
 async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Set up a config entry."""
     region_identifier: str = entry.data[CONF_REGION_IDENTIFIER]
 
-    # Initialize the API.
+    # Initialize the API and coordinator.
     api = await hass.async_add_executor_job(DwdWeatherWarningsAPI, region_identifier)
+    coordinator = DwdWeatherWarningsCoordinator(hass, api)
 
-    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = api
+    await coordinator.async_config_entry_first_refresh()
+
+    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
     await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
 
     return True

homeassistant/components/dwd_weather_warnings/coordinator.py

@@ -0,0 +1,26 @@
+"""Data coordinator for the dwd_weather_warnings integration."""
+
+from __future__ import annotations
+
+from dwdwfsapi import DwdWeatherWarningsAPI
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .const import DEFAULT_SCAN_INTERVAL, DOMAIN, LOGGER
+
+
+class DwdWeatherWarningsCoordinator(DataUpdateCoordinator[None]):
+    """Custom coordinator for the dwd_weather_warnings integration."""
+
+    def __init__(self, hass: HomeAssistant, api: DwdWeatherWarningsAPI) -> None:
+        """Initialize the dwd_weather_warnings coordinator."""
+        super().__init__(
+            hass, LOGGER, name=DOMAIN, update_interval=DEFAULT_SCAN_INTERVAL
+        )
+
+        self.api = api
+
+    async def _async_update_data(self) -> None:
+        """Get the latest data from the DWD Weather Warnings API."""
+        await self.hass.async_add_executor_job(self.api.update)

homeassistant/components/dwd_weather_warnings/sensor.py

@@ -27,7 +27,7 @@ import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
 from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
-from homeassistant.util import Throttle
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
 from .const import (
     ADVANCE_WARNING_SENSOR,
@@ -48,10 +48,9 @@ from .const import (
     CONF_REGION_NAME,
     CURRENT_WARNING_SENSOR,
     DEFAULT_NAME,
-    DEFAULT_SCAN_INTERVAL,
     DOMAIN,
-    LOGGER,
 )
+from .coordinator import DwdWeatherWarningsCoordinator
 
 SENSOR_TYPES: tuple[SensorEntityDescription, ...] = (
     SensorEntityDescription(
@@ -109,55 +108,60 @@ async def async_setup_entry(
     hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
 ) -> None:
     """Set up entities from config entry."""
-    api = WrappedDwDWWAPI(hass.data[DOMAIN][entry.entry_id])
+    coordinator = hass.data[DOMAIN][entry.entry_id]
 
     async_add_entities(
         [
-            DwdWeatherWarningsSensor(api, entry, description)
+            DwdWeatherWarningsSensor(coordinator, entry, description)
             for description in SENSOR_TYPES
         ],
         True,
     )
 
 
-class DwdWeatherWarningsSensor(SensorEntity):
+class DwdWeatherWarningsSensor(
+    CoordinatorEntity[DwdWeatherWarningsCoordinator], SensorEntity
+):
     """Representation of a DWD-Weather-Warnings sensor."""
 
     _attr_attribution = "Data provided by DWD"
 
     def __init__(
         self,
-        api,
+        coordinator: DwdWeatherWarningsCoordinator,
         entry: ConfigEntry,
         description: SensorEntityDescription,
     ) -> None:
         """Initialize a DWD-Weather-Warnings sensor."""
+        super().__init__(coordinator)
+
         self.entity_description = description
         self._attr_name = f"{DEFAULT_NAME} {entry.title} {description.name}"
         self._attr_unique_id = f"{entry.unique_id}-{description.key}"
-        self._api = api
+
+        self.api = coordinator.api
 
     @property
     def native_value(self):
         """Return the state of the sensor."""
         if self.entity_description.key == CURRENT_WARNING_SENSOR:
-            return self._api.api.current_warning_level
+            return self.api.current_warning_level
 
-        return self._api.api.expected_warning_level
+        return self.api.expected_warning_level
 
     @property
     def extra_state_attributes(self):
         """Return the state attributes of the sensor."""
         data = {
-            ATTR_REGION_NAME: self._api.api.warncell_name,
+            ATTR_REGION_NAME: self.api.warncell_name,
-            ATTR_REGION_ID: self._api.api.warncell_id,
+            ATTR_REGION_ID: self.api.warncell_id,
-            ATTR_LAST_UPDATE: self._api.api.last_update,
+            ATTR_LAST_UPDATE: self.api.last_update,
         }
 
         if self.entity_description.key == CURRENT_WARNING_SENSOR:
-            searched_warnings = self._api.api.current_warnings
+            searched_warnings = self.api.current_warnings
         else:
-            searched_warnings = self._api.api.expected_warnings
+            searched_warnings = self.api.expected_warnings
 
         data[ATTR_WARNING_COUNT] = len(searched_warnings)
 
@@ -173,7 +177,7 @@ class DwdWeatherWarningsSensor(SensorEntity):
             data[f"warning_{i}_parameters"] = warning[API_ATTR_WARNING_PARAMETERS]
             data[f"warning_{i}_color"] = warning[API_ATTR_WARNING_COLOR]
 
-            # Dictionary for the attribute containing the complete warning
+            # Dictionary for the attribute containing the complete warning.
             warning_copy = warning.copy()
             warning_copy[API_ATTR_WARNING_START] = data[f"warning_{i}_start"]
             warning_copy[API_ATTR_WARNING_END] = data[f"warning_{i}_end"]
@@ -184,28 +188,4 @@ class DwdWeatherWarningsSensor(SensorEntity):
     @property
     def available(self) -> bool:
         """Could the device be accessed during the last update call."""
-        return self._api.api.data_valid
+        return self.api.data_valid
-
-    def update(self) -> None:
-        """Get the latest data from the DWD-Weather-Warnings API."""
-        LOGGER.debug(
-            "Update requested for %s (%s) by %s",
-            self._api.api.warncell_name,
-            self._api.api.warncell_id,
-            self.entity_description.key,
-        )
-        self._api.update()
-
-
-class WrappedDwDWWAPI:
-    """Wrapper for the DWD-Weather-Warnings api."""
-
-    def __init__(self, api):
-        """Initialize a DWD-Weather-Warnings wrapper."""
-        self.api = api
-
-    @Throttle(DEFAULT_SCAN_INTERVAL)
-    def update(self):
-        """Get the latest data from the DWD-Weather-Warnings API."""
-        self.api.update()
-        LOGGER.debug("Update performed")