feat: add gps location option

custom_components/nws_alerts/__init__.py

@@ -17,6 +17,7 @@
 
 from .const import (
     API_ENDPOINT,
+    CONF_GPS_LOC,
     CONF_INTERVAL,
     CONF_TIMEOUT,
     CONF_ZONE_ID,
@@ -146,14 +147,22 @@ async def update_alerts(config) -> dict:
 async def async_get_state(config) -> dict:
     """Query API for status."""
 
+    zone_id = ""
+    gps_loc = ""
+    url = "%s/alerts/active/count" % API_ENDPOINT
     values = {}
     headers = {"User-Agent": USER_AGENT, "Accept": "application/ld+json"}
     data = None
-    url = "%s/alerts/active/count" % API_ENDPOINT
-    zone_id = config[CONF_ZONE_ID]
+
+    if CONF_ZONE_ID in config:
+        zone_id = config[CONF_ZONE_ID]
+        _LOGGER.debug("getting state for %s from %s" % (zone_id, url))
+    elif CONF_GPS_LOC in config:
+        gps_loc = config[CONF_GPS_LOC]
+        _LOGGER.debug("getting state for %s from %s" % (gps_loc, url))        
+
     async with aiohttp.ClientSession() as session:
         async with session.get(url, headers=headers) as r:
-            _LOGGER.debug("getting state for %s from %s" % (zone_id, url))
             if r.status == 200:
                 data = await r.json()
 
@@ -173,22 +182,29 @@ async def async_get_state(config) -> dict:
         if "zones" in data:
             for zone in zone_id.split(","):
                 if zone in data["zones"]:
-                    values = await async_get_alerts(zone_id)
+                    values = await async_get_alerts(zone_id, gps_loc)
                     break
 
     return values
 
 
-async def async_get_alerts(zone_id: str) -> dict:
+async def async_get_alerts(zone_id: str = "", gps_loc: str = "") -> dict:
     """Query API for Alerts."""
 
+    url = ""
     values = {}
     headers = {"User-Agent": USER_AGENT, "Accept": "application/geo+json"}
     data = None
-    url = "%s/alerts/active?zone=%s" % (API_ENDPOINT, zone_id)
+
+    if zone_id != "":
+        url = "%s/alerts/active?zone=%s" % (API_ENDPOINT, zone_id)
+        _LOGGER.debug("getting alert for %s from %s" % (zone_id, url))
+    elif gps_loc != "":
+        url = '%s/alerts/active?point=%s' % (API_ENDPOINT, gps_loc)
+        _LOGGER.debug("getting alert for %s from %s" % (gps_loc, url))
+
     async with aiohttp.ClientSession() as session:
         async with session.get(url, headers=headers) as r:
-            _LOGGER.debug("getting alert for %s from %s" % (zone_id, url))
             if r.status == 200:
                 data = await r.json()
 

custom_components/nws_alerts/config_flow.py

@@ -13,6 +13,7 @@
 
 from .const import (
     API_ENDPOINT,
+    CONF_GPS_LOC,
     CONF_INTERVAL,
     CONF_TIMEOUT,
     CONF_ZONE_ID,
@@ -28,9 +29,10 @@
 JSON_ID = "id"
 
 _LOGGER = logging.getLogger(__name__)
+MENU_OPTIONS = ["zone", "gps_loc"]
 
 
-def _get_schema(hass: Any, user_input: list, default_dict: list) -> Any:
+def _get_schema_zone(hass: Any, user_input: list, default_dict: list) -> Any:
     """Gets a schema using the default_dict as a backup."""
     if user_input is None:
         user_input = {}
@@ -48,6 +50,23 @@ def _get_default(key):
         }
     )
 
+def _get_schema_gps(hass: Any, user_input: list, default_dict: list) -> Any:
+    """Gets a schema using the default_dict as a backup."""
+    if user_input is None:
+        user_input = {}
+
+    def _get_default(key):
+        """Gets default value for key."""
+        return user_input.get(key, default_dict.get(key))
+
+    return vol.Schema(
+        {
+            vol.Required(CONF_GPS_LOC, default=_get_default(CONF_GPS_LOC)): str,
+            vol.Optional(CONF_NAME, default=_get_default(CONF_NAME)): str,
+            vol.Optional(CONF_INTERVAL, default=_get_default(CONF_INTERVAL)): int,
+            vol.Optional(CONF_TIMEOUT, default=_get_default(CONF_TIMEOUT)): int,
+        }
+    )
 
 async def _get_zone_list(self) -> list | None:
     """Return list of zone by lat/lon"""
@@ -100,17 +119,52 @@ def __init__(self):
     #         return self.async_abort(reason=next(iter(errors.values())))
     #     return result
 
-    async def async_step_user(self, user_input={}):
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle the flow initialized by the user."""
+        return self.async_show_menu(step_id="user", menu_options=MENU_OPTIONS)
+
+    async def async_step_gps_loc(self, user_input={}):
+        """Handle a flow initialized by the user."""
+        lat = self.hass.config.latitude
+        lon = self.hass.config.longitude
+        self._errors = {}
+        self._gps_loc = f"{lat},{lon}"
+
+        if user_input is not None:
+            self._data.update(user_input)
+            return self.async_create_entry(title=self._data[CONF_NAME], data=self._data)
+        return await self._show_config_gps_loc(user_input)   
+
+    async def _show_config_gps_loc(self, user_input):
+        """Show the configuration form to edit location data."""
+
+        # Defaults
+        defaults = {
+            CONF_NAME: DEFAULT_NAME,
+            CONF_INTERVAL: DEFAULT_INTERVAL,
+            CONF_TIMEOUT: DEFAULT_TIMEOUT,
+            CONF_GPS_LOC: self._gps_loc,
+        }
+
+        return self.async_show_form(
+            step_id="gps_loc",
+            data_schema=_get_schema_gps(self.hass, user_input, defaults),
+            errors=self._errors,
+        )     
+
+    async def async_step_zone(self, user_input={}):
         """Handle a flow initialized by the user."""
         self._errors = {}
         self._zone_list = await _get_zone_list(self)
 
         if user_input is not None:
             self._data.update(user_input)
             return self.async_create_entry(title=self._data[CONF_NAME], data=self._data)
-        return await self._show_config_form(user_input)
+        return await self._show_config_zone(user_input)
 
-    async def _show_config_form(self, user_input):
+    async def _show_config_zone(self, user_input):
         """Show the configuration form to edit location data."""
 
         # Defaults
@@ -122,8 +176,8 @@ async def _show_config_form(self, user_input):
         }
 
         return self.async_show_form(
-            step_id="user",
-            data_schema=_get_schema(self.hass, user_input, defaults),
+            step_id="zone",
+            data_schema=_get_schema_zone(self.hass, user_input, defaults),
             errors=self._errors,
         )
 
@@ -154,6 +208,6 @@ async def _show_options_form(self, user_input):
 
         return self.async_show_form(
             step_id="init",
-            data_schema=_get_schema(self.hass, user_input, self._data),
+            data_schema=_get_schema_zone(self.hass, user_input, self._data),
             errors=self._errors,
         )

custom_components/nws_alerts/const.py

@@ -6,6 +6,7 @@
 CONF_TIMEOUT = "timeout"
 CONF_INTERVAL = "interval"
 CONF_ZONE_ID = "zone_id"
+CONF_GPS_LOC = "gps_loc"
 
 # Defaults
 DEFAULT_ICON = "mdi:alert"

custom_components/nws_alerts/sensor.py

@@ -13,6 +13,7 @@
 
 from .const import (
     ATTRIBUTION,
+    CONF_GPS_LOC,
     CONF_INTERVAL,
     CONF_TIMEOUT,
     CONF_ZONE_ID,
@@ -35,7 +36,8 @@
 
 PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
     {
-        vol.Required(CONF_ZONE_ID): cv.string,
+        vol.Optional(CONF_ZONE_ID): cv.string,
+        vol.Optional(CONF_GPS_LOC): cv.string,
         vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
         vol.Optional(CONF_INTERVAL, default=DEFAULT_INTERVAL): int,
         vol.Optional(CONF_TIMEOUT, default=DEFAULT_TIMEOUT): int,
@@ -47,10 +49,20 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
     """Configuration from yaml"""
     if DOMAIN not in hass.data.keys():
         hass.data.setdefault(DOMAIN, {})
-        config.entry_id = slugify(f"{config.get(CONF_ZONE_ID)}")
+        if CONF_ZONE_ID in config:
+            config.entry_id = slugify(f"{config.get(CONF_ZONE_ID)}")
+        elif CONF_GPS_LOC in config:
+            config.entry_id = slugify(f"{config.get(CONF_GPS_LOC)}")
+        elif CONF_GPS_LOC and CONF_ZONE_ID not in config:
+            raise ValueError("GPS or Zone needs to be configured.")
         config.data = config
     else:
-        config.entry_id = slugify(f"{config.get(CONF_ZONE_ID)}")
+        if CONF_ZONE_ID in config:
+            config.entry_id = slugify(f"{config.get(CONF_ZONE_ID)}")
+        elif CONF_GPS_LOC in config:
+            config.entry_id = slugify(f"{config.get(CONF_GPS_LOC)}")
+        elif CONF_GPS_LOC and CONF_ZONE_ID not in config:
+            raise ValueError("GPS or Zone needs to be configured.")            
         config.data = config
 
     # Setup the data coordinator
@@ -84,16 +96,6 @@ def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
         self._config = entry
         self._name = entry.data[CONF_NAME]
         self._icon = DEFAULT_ICON
-        self._state = 0
-        self._event = None
-        self._event_id = None
-        self._message_type = None
-        self._event_status = None
-        self._event_severity = None
-        self._event_expires = None
-        self._display_desc = None
-        self._spoken_desc = None
-        self._zone_id = entry.data[CONF_ZONE_ID].replace(" ", "")
         self.coordinator = hass.data[DOMAIN][entry.entry_id][COORDINATOR]
 
     @property
@@ -120,8 +122,7 @@ def state(self):
             return None
         elif "state" in self.coordinator.data.keys():
             return self.coordinator.data["state"]
-        else:
-            return None
+        return None
 
     @property
     def extra_state_attributes(self):

custom_components/nws_alerts/translations/en.json

@@ -2,6 +2,22 @@
   "config": {
     "step": {
       "user": {
+        "description": "Please select your NWS lookup method.",
+        "menu_options": {
+          "zone": "Zone ID (old method)",
+          "gps_loc": "GPS Location (new method)"
+        }
+      },
+      "gps_loc": {
+        "description": "Please enter your latitude and longitude, by default your coordinates from Home Assistant are used.",
+        "data": {
+          "name": "Friendly Name",
+          "gps_loc": "Your GPS coordinates",
+          "interval": "Update Interval (in minutes)",
+          "timeout":"Update Timeout (in seconds)"
+        }
+      },
+      "zone": {
         "data": {
           "name": "Friendly Name",
           "zone_id": "Zone ID(s)",
@@ -14,7 +30,23 @@
   },
   "options": {
     "step": {
-      "init": {
+      "user": {
+        "description": "Please select your NWS lookup method.",
+        "menu_options": {
+          "zone": "Zone ID (old method)",
+          "gps_loc": "GPS Location (new method)"
+        }
+      },
+      "gps_loc": {
+        "description": "Please enter your latitude and longitude, by default your coordinates from Home Assistant are used.",
+        "data": {
+          "name": "Friendly Name",
+          "gps_loc": "Your GPS coordinates",
+          "interval": "Update Interval (in minutes)",
+          "timeout":"Update Timeout (in seconds)"
+        }
+      },      
+      "zone": {
         "data": {
           "name": "Friendly Name",
           "zone_id": "Zone ID(s)",

tests/const.py

@@ -1,4 +1,6 @@
 """Constants for tests."""
 
 CONFIG_DATA = {"name": "NWS Alerts", "zone_id": "AZZ540,AZC013"}
-CONFIG_DATA_2 = {"name": "NWS Alerts YAML", "zone_id": "AZZ540"}
+CONFIG_DATA_2 = {"name": "NWS Alerts YAML", "zone_id": "AZZ540"}
+CONFIG_DATA_3 = {"name": "NWS Alerts", "gps_loc": "123,-456"}
+CONFIG_DATA_BAD = {"name": "NWS Alerts" }