Add device tracker option

homeassistant/components/dwd_weather_warnings/__init__.py

@@ -3,22 +3,43 @@
 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_GPS_TRACKER, CONF_REGION_IDENTIFIER, DOMAIN, LOGGER, PLATFORMS
 from .coordinator import DwdWeatherWarningsCoordinator
+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)
+    gps_tracker: str = entry.data.get(CONF_GPS_TRACKER, None)
+
+    # 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 gps_tracker is not None:
+        try:
+            registry = er.async_get(hass)
+            gps_tracker = er.async_validate_entity_id(registry, gps_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",
+                gps_tracker,
+            )
+            return False
+
+        position = get_position_data(hass, gps_tracker)
+        api = await hass.async_add_executor_job(DwdWeatherWarningsAPI, position)
 
-    # Initialize the API and coordinator.
-    api = await hass.async_add_executor_job(DwdWeatherWarningsAPI, region_identifier)
     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

@@ -10,8 +10,10 @@
 from homeassistant.config_entries import ConfigFlow
 from homeassistant.data_entry_flow import FlowResult
 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_GPS_TRACKER, CONF_REGION_IDENTIFIER, DOMAIN
+from .util import get_position_data
 
 
 class DwdWeatherWarningsConfigFlow(ConfigFlow, domain=DOMAIN):
@@ -26,27 +28,58 @@ 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_GPS_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_GPS_TRACKER in user_input:
+                # Validate position using the API
+                device_tracker = user_input[CONF_GPS_TRACKER]
+                position = get_position_data(self.hass, device_tracker)
+
+                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:
+                    # Set the unique ID for this config entry.
+                    await self.async_set_unique_id(f"{position[0]}-{position[1]}")
+                    self._abort_if_unique_id_configured()
+
+                    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_GPS_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_GPS_TRACKER: Final = "gps_tracker"
 
 ATTR_REGION_NAME: Final = "region_name"
 ATTR_REGION_ID: Final = "region_id"

homeassistant/components/dwd_weather_warnings/coordinator.py

@@ -7,7 +7,8 @@
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 
-from .const import DEFAULT_SCAN_INTERVAL, DOMAIN, LOGGER
+from .const import CONF_GPS_TRACKER, DEFAULT_SCAN_INTERVAL, DOMAIN, LOGGER
+from .util import get_position_data
 
 
 class DwdWeatherWarningsCoordinator(DataUpdateCoordinator[None]):
@@ -23,4 +24,11 @@ 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 self.config_entry is not None:
+            if gps_tracker := self.config_entry.data.get(CONF_GPS_TRACKER, None):
+                position = get_position_data(self.hass, gps_tracker)
+                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/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,26 +98,26 @@ 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.coordinator.api.current_warning_level
 
-        return self.api.expected_warning_level
+        return self.coordinator.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,
-            ATTR_REGION_ID: self.api.warncell_id,
-            ATTR_LAST_UPDATE: self.api.last_update,
+            ATTR_REGION_NAME: self.coordinator.api.warncell_name,
+            ATTR_REGION_ID: self.coordinator.api.warncell_id,
+            ATTR_LAST_UPDATE: self.coordinator.api.last_update,
         }
 
         if self.entity_description.key == CURRENT_WARNING_SENSOR:
-            searched_warnings = self.api.current_warnings
+            searched_warnings = self.coordinator.api.current_warnings
         else:
-            searched_warnings = self.api.expected_warnings
+            searched_warnings = self.coordinator.api.expected_warnings
 
         data[ATTR_WARNING_COUNT] = len(searched_warnings)
 
@@ -142,4 +144,4 @@ def extra_state_attributes(self):
     @property
     def available(self) -> bool:
         """Could the device be accessed during the last update call."""
-        return self.api.data_valid
+        return self.coordinator.api.data_valid

homeassistant/components/dwd_weather_warnings/strings.json

@@ -2,17 +2,20 @@
   "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",
+          "gps_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."
     },
     "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,32 @@
+"""Util functions for the dwd_weather_warnings integration."""
+
+from __future__ import annotations
+
+from homeassistant.core import HomeAssistant
+
+from .const import LOGGER
+
+
+def get_position_data(
+    hass: HomeAssistant, device_tracker: str
+) -> tuple[float, float] | None:
+    """Extract longitude and latitude from a device tracker."""
+    entity = hass.states.get(device_tracker)
+    if entity is None:
+        return None
+
+    latitude = entity.attributes.get("latitude")
+    if not latitude:
+        LOGGER.warning(
+            "Failed to find attribute 'latitude' in device_tracker %s", entity
+        )
+        return None
+
+    longitude = entity.attributes.get("longitude")
+    if not longitude:
+        LOGGER.warning(
+            "Failed to find attribute 'longitude' in device_tracker %s", entity
+        )
+        return None
+
+    return (latitude, longitude)