Add optional location based region to dwd_weather_warnings

homeassistant/components/dwd_weather_warnings/__init__.py

@@ -3,22 +3,57 @@
 from __future__ import annotations
 
 from dwdwfsapi import DwdWeatherWarningsAPI
+import voluptuous as vol
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
+from homeassistant.helpers import entity_registry as er
 
-from .const import CONF_REGION_IDENTIFIER, DOMAIN, PLATFORMS
+from .const import (
+    CONF_REGION_DEVICE_TRACKER,
+    CONF_REGION_IDENTIFIER,
+    DOMAIN,
+    LOGGER,
+    PLATFORMS,
+)
 from .coordinator import DwdWeatherWarningsCoordinator
+from .exceptions import EntityNotFoundError
+from .util import get_position_data
 
 
 async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Set up a config entry."""
-    region_identifier: str = entry.data[CONF_REGION_IDENTIFIER]
+    region_identifier: str = entry.data.get(CONF_REGION_IDENTIFIER, None)
+    device_tracker: str = entry.data.get(CONF_REGION_DEVICE_TRACKER, None)
 
-    # Initialize the API and coordinator.
-    api = await hass.async_add_executor_job(DwdWeatherWarningsAPI, region_identifier)
-    coordinator = DwdWeatherWarningsCoordinator(hass, api)
+    # Initialize the API and coordinator based on the specified data.
+    if region_identifier is not None:
+        api = await hass.async_add_executor_job(
+            DwdWeatherWarningsAPI, region_identifier
+        )
+    elif device_tracker is not None:
+        registry = er.async_get(hass)
+
+        try:
+            device_tracker = er.async_validate_entity_id(registry, device_tracker)
+        except vol.Invalid:
+            # The entity/UUID is invalid or not associated with an entity registry item.
+            LOGGER.error(
+                "Failed to setup dwd_weather_warnings for unknown entity %s",
+                device_tracker,
+            )
+            return False
+
+        try:
+            position = get_position_data(hass, device_tracker)
+        except (EntityNotFoundError, AttributeError) as err:
+            # The provided device_tracker is not available or missing required attributes.
+            LOGGER.error(f"Failed to setup dwd_weather_warnings: {repr(err)}")
+            return False
 
+        api = await hass.async_add_executor_job(DwdWeatherWarningsAPI, position)
+
+    coordinator = DwdWeatherWarningsCoordinator(hass, api)
     await coordinator.async_config_entry_first_refresh()
 
     hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator

homeassistant/components/dwd_weather_warnings/config_flow.py

@@ -8,9 +8,13 @@
 import voluptuous as vol
 
 from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.helpers import entity_registry as er
 import homeassistant.helpers.config_validation as cv
+from homeassistant.helpers.selector import EntitySelector, EntitySelectorConfig
 
-from .const import CONF_REGION_IDENTIFIER, DOMAIN
+from .const import CONF_REGION_DEVICE_TRACKER, CONF_REGION_IDENTIFIER, DOMAIN
+from .exceptions import EntityNotFoundError
+from .util import get_position_data
 
 
 class DwdWeatherWarningsConfigFlow(ConfigFlow, domain=DOMAIN):
@@ -25,27 +29,72 @@ async def async_step_user(
         errors: dict = {}
 
         if user_input is not None:
-            region_identifier = user_input[CONF_REGION_IDENTIFIER]
+            valid_options = (CONF_REGION_IDENTIFIER, CONF_REGION_DEVICE_TRACKER)
 
-            # Validate region identifier using the API
-            if not await self.hass.async_add_executor_job(
-                DwdWeatherWarningsAPI, region_identifier
-            ):
-                errors["base"] = "invalid_identifier"
+            # Check, if either CONF_REGION_IDENTIFIER or CONF_GPS_TRACKER has been set.
+            if all(k not in user_input for k in valid_options):
+                errors["base"] = "no_identifier"
+            elif all(k in user_input for k in valid_options):
+                errors["base"] = "ambiguous_identifier"
+            elif CONF_REGION_IDENTIFIER in user_input:
+                # Validate region identifier using the API
+                identifier = user_input[CONF_REGION_IDENTIFIER]
 
-            if not errors:
-                # Set the unique ID for this config entry.
-                await self.async_set_unique_id(region_identifier)
-                self._abort_if_unique_id_configured()
+                if not await self.hass.async_add_executor_job(
+                    DwdWeatherWarningsAPI, identifier
+                ):
+                    errors["base"] = "invalid_identifier"
 
-                return self.async_create_entry(title=region_identifier, data=user_input)
+                if not errors:
+                    # Set the unique ID for this config entry.
+                    await self.async_set_unique_id(identifier)
+                    self._abort_if_unique_id_configured()
+
+                    return self.async_create_entry(title=identifier, data=user_input)
+            elif CONF_REGION_DEVICE_TRACKER in user_input:
+                device_tracker = user_input[CONF_REGION_DEVICE_TRACKER]
+                registry = er.async_get(self.hass)
+                entity_entry = registry.async_get(device_tracker)
+
+                if entity_entry is None:
+                    errors["base"] = "entity_not_found"
+                else:
+                    try:
+                        position = get_position_data(self.hass, entity_entry.id)
+                    except EntityNotFoundError:
+                        errors["base"] = "entity_not_found"
+                    except AttributeError:
+                        errors["base"] = "attribute_not_found"
+                    else:
+                        # Validate position using the API
+                        if not await self.hass.async_add_executor_job(
+                            DwdWeatherWarningsAPI, position
+                        ):
+                            errors["base"] = "invalid_identifier"
+
+                # Position is valid here, because the API call was successful.
+                if not errors and position is not None and entity_entry is not None:
+                    # Set the unique ID for this config entry.
+                    await self.async_set_unique_id(entity_entry.id)
+                    self._abort_if_unique_id_configured()
+
+                    # Replace entity ID with registry ID for more stability.
+                    user_input[CONF_REGION_DEVICE_TRACKER] = entity_entry.id
+
+                    return self.async_create_entry(
+                        title=device_tracker.removeprefix("device_tracker."),
+                        data=user_input,
+                    )
 
         return self.async_show_form(
             step_id="user",
             errors=errors,
             data_schema=vol.Schema(
                 {
-                    vol.Required(CONF_REGION_IDENTIFIER): cv.string,
+                    vol.Optional(CONF_REGION_IDENTIFIER): cv.string,
+                    vol.Optional(CONF_REGION_DEVICE_TRACKER): EntitySelector(
+                        EntitySelectorConfig(domain="device_tracker")
+                    ),
                 }
             ),
         )

homeassistant/components/dwd_weather_warnings/const.py

@@ -14,6 +14,7 @@
 
 CONF_REGION_NAME: Final = "region_name"
 CONF_REGION_IDENTIFIER: Final = "region_identifier"
+CONF_REGION_DEVICE_TRACKER: Final = "region_device_tracker"
 
 ATTR_REGION_NAME: Final = "region_name"
 ATTR_REGION_ID: Final = "region_id"

homeassistant/components/dwd_weather_warnings/coordinator.py

@@ -4,15 +4,20 @@
 
 from dwdwfsapi import DwdWeatherWarningsAPI
 
+from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
-from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
-from .const import DEFAULT_SCAN_INTERVAL, DOMAIN, LOGGER
+from .const import CONF_REGION_DEVICE_TRACKER, DEFAULT_SCAN_INTERVAL, DOMAIN, LOGGER
+from .exceptions import EntityNotFoundError
+from .util import get_position_data
 
 
 class DwdWeatherWarningsCoordinator(DataUpdateCoordinator[None]):
     """Custom coordinator for the dwd_weather_warnings integration."""
 
+    config_entry: ConfigEntry
+
     def __init__(self, hass: HomeAssistant, api: DwdWeatherWarningsAPI) -> None:
         """Initialize the dwd_weather_warnings coordinator."""
         super().__init__(
@@ -23,4 +28,14 @@ def __init__(self, hass: HomeAssistant, api: DwdWeatherWarningsAPI) -> None:
 
     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)
+        if device_tracker := self.config_entry.data.get(CONF_REGION_DEVICE_TRACKER):
+            try:
+                position = get_position_data(self.hass, device_tracker)
+            except (EntityNotFoundError, AttributeError) as err:
+                raise UpdateFailed(f"Error fetching position: {repr(err)}") from err
+
+            self.api = await self.hass.async_add_executor_job(
+                DwdWeatherWarningsAPI, position
+            )
+        else:
+            await self.hass.async_add_executor_job(self.api.update)

homeassistant/components/dwd_weather_warnings/exceptions.py

@@ -0,0 +1,6 @@
+"""Exceptions for the dwd_weather_warnings integration."""
+from homeassistant.exceptions import HomeAssistantError
+
+
+class EntityNotFoundError(HomeAssistantError):
+    """When a referenced entity was not found."""

homeassistant/components/dwd_weather_warnings/sensor.py

@@ -11,6 +11,8 @@
 
 from __future__ import annotations
 
+from typing import Any
+
 from homeassistant.components.sensor import SensorEntity, SensorEntityDescription
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
@@ -96,15 +98,15 @@ def __init__(
         self.api = coordinator.api
 
     @property
-    def native_value(self):
+    def native_value(self) -> int | None:
         """Return the state of the sensor."""
         if self.entity_description.key == CURRENT_WARNING_SENSOR:
             return self.api.current_warning_level
 
         return self.api.expected_warning_level
 
     @property
-    def extra_state_attributes(self):
+    def extra_state_attributes(self) -> dict[str, Any]:
         """Return the state attributes of the sensor."""
         data = {
             ATTR_REGION_NAME: self.api.warncell_name,

homeassistant/components/dwd_weather_warnings/strings.json

@@ -2,17 +2,22 @@
   "config": {
     "step": {
       "user": {
-        "description": "To identify the desired region, the warncell ID / name is required.",
+        "description": "To identify the desired region, either the warncell ID / name or device tracker is required. The provided device tracker has to contain the attributes 'latitude' and 'longitude'.",
         "data": {
-          "region_identifier": "Warncell ID or name"
+          "region_identifier": "Warncell ID or name",
+          "region_device_tracker": "Device tracker entity"
         }
       }
     },
     "error": {
-      "invalid_identifier": "The specified region identifier is invalid."
+      "no_identifier": "Either the region identifier or device tracker is required.",
+      "ambiguous_identifier": "The region identifier and device tracker can not be specified together.",
+      "invalid_identifier": "The specified region identifier / device tracker is invalid.",
+      "entity_not_found": "The specified device tracker entity was not found.",
+      "attribute_not_found": "The required `latitude` or `longitude` attribute was not found in the specified device tracker."
     },
     "abort": {
-      "already_configured": "Warncell ID / name is already configured.",
+      "already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
       "invalid_identifier": "[%key:component::dwd_weather_warnings::config::error::invalid_identifier%]"
     }
   },

homeassistant/components/dwd_weather_warnings/util.py

@@ -0,0 +1,39 @@
+"""Util functions for the dwd_weather_warnings integration."""
+
+from __future__ import annotations
+
+from homeassistant.const import ATTR_LATITUDE, ATTR_LONGITUDE
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import entity_registry as er
+
+from .exceptions import EntityNotFoundError
+
+
+def get_position_data(
+    hass: HomeAssistant, registry_id: str
+) -> tuple[float, float] | None:
+    """Extract longitude and latitude from a device tracker."""
+    registry = er.async_get(hass)
+    registry_entry = registry.async_get(registry_id)
+    if registry_entry is None:
+        raise EntityNotFoundError(f"Failed to find registry entry {registry_id}")
+
+    entity = hass.states.get(registry_entry.entity_id)
+    if entity is None:
+        raise EntityNotFoundError(f"Failed to find entity {registry_entry.entity_id}")
+
+    latitude = entity.attributes.get(ATTR_LATITUDE)
+    if not latitude:
+        raise AttributeError(
+            f"Failed to find attribute '{ATTR_LATITUDE}' in {registry_entry.entity_id}",
+            ATTR_LATITUDE,
+        )
+
+    longitude = entity.attributes.get(ATTR_LONGITUDE)
+    if not longitude:
+        raise AttributeError(
+            f"Failed to find attribute '{ATTR_LONGITUDE}' in {registry_entry.entity_id}",
+            ATTR_LONGITUDE,
+        )
+
+    return (latitude, longitude)