Work around device tracker component deleting devices (take 2)

custom_components/gpslogger/__init__.py

@@ -23,6 +23,7 @@
     entity_registry as er,
 )
 from homeassistant.helpers.dispatcher import async_dispatcher_send
+from homeassistant.helpers.typing import ConfigType
 
 from .const import (
     ATTR_ACCURACY,
@@ -68,7 +69,32 @@ def _id(value: str) -> str:
 )
 
 
-async def handle_webhook(hass, webhook_id, request):
+async def async_setup(hass: HomeAssistant, _: ConfigType) -> bool:
+    """Set up integration."""
+    hass.data[DOMAIN] = {"devices": set(), "warned_no_last_seen": False}
+    ent_reg = er.async_get(hass)
+
+    async def device_work_around(_: Event) -> None:
+        """Work around for device tracker component deleting devices.
+
+        The device tracker component level code, at startup, deletes devices that are
+        associated only with device_tracker entities. Not only that, it will delete
+        those device_tracker entities from the entity registry as well. So, when HA
+        shuts down, remove references to devices from our device_tracker entity registry
+        entries. They'll get set back up automatically the next time our config is
+        loaded (i.e., setup.)
+        """
+        for c_entry in hass.config_entries.async_entries(DOMAIN):
+            for r_entry in er.async_entries_for_config_entry(ent_reg, c_entry.entry_id):
+                ent_reg.async_update_entity(r_entry.entity_id, device_id=None)
+
+    hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, device_work_around)
+    return True
+
+
+async def handle_webhook(
+    hass: HomeAssistant, webhook_id: str, request: web.Request
+) -> web.Response:
     """Handle incoming webhook with GPSLogger request."""
     try:
         data = WEBHOOK_SCHEMA(dict(await request.post()))
@@ -112,30 +138,10 @@ async def handle_webhook(hass, webhook_id, request):
 
 async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Configure based on config entry."""
-
-    async def device_work_around(_: Event):
-        """Work around for device tracker component deleting devices.
-
-        The device tracker component level code, at startup, deletes devices that are
-        associated only with device_tracker entities. Not only that, it will delete
-        those device_tracker entities from the entity registry as well. So, when HA
-        shuts down, remove references to devices from our device_tracker entity registry
-        entries. They'll get set back up automatically the next time our config is
-        loaded (i.e., setup.)
-        """
-        ent_reg = er.async_get(hass)
-        for r_entry in ent_reg.entities.get_entries_for_config_entry_id(entry.entry_id):
-            ent_reg.async_update_entity(r_entry.entity_id, device_id=None)
-
-    if DOMAIN not in hass.data:
-        hass.data[DOMAIN] = {"devices": set(), "warned_no_last_seen": False}
-        hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, device_work_around)
     webhook.async_register(
         hass, DOMAIN, "GPSLogger", entry.data[CONF_WEBHOOK_ID], handle_webhook
     )
-
     await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
-
     return True
 
 

custom_components/gpslogger/device_tracker.py

@@ -1,9 +1,10 @@
 """Support for the GPSLogger device tracking."""
 from __future__ import annotations
 
+from collections.abc import Mapping
 from datetime import datetime
 import logging
-from typing import cast
+from typing import Any, cast
 
 from homeassistant.components.device_tracker import SourceType, TrackerEntity
 from homeassistant.config_entries import ConfigEntry
@@ -41,7 +42,13 @@ async def async_setup_entry(
     """Configure a dispatcher connection based on a config entry."""
 
     @callback
-    def _receive_data(device, gps, battery, accuracy, attrs):
+    def _receive_data(
+        device: str,
+        gps: tuple[float, float],
+        battery: float,
+        accuracy: float,
+        attrs: dict[str, Any],
+    ) -> None:
         """Receive set location."""
         if device in hass.data[GPL_DOMAIN]["devices"]:
             return
@@ -64,7 +71,7 @@ def _receive_data(device, gps, battery, accuracy, attrs):
     entities = []
     for dev_id in dev_ids:
         hass.data[GPL_DOMAIN]["devices"].add(dev_id)
-        entity = GPSLoggerEntity(dev_id, None, None, None, None)
+        entity = GPSLoggerEntity(dev_id)
         entities.append(entity)
 
     async_add_entities(entities)
@@ -75,45 +82,51 @@ class GPSLoggerEntity(TrackerEntity, RestoreEntity):
 
     _attr_has_entity_name = True
     _attr_name = None
-    _prv_seen: datetime | None = None
 
-    def __init__(self, device, location, battery, accuracy, attributes):
+    def __init__(
+        self,
+        device: str,
+        location: tuple[float, float] | None = None,
+        battery: float | None = None,
+        accuracy: float | None = None,
+        attributes: dict[str, Any] | None = None,
+    ) -> None:
         """Set up GPSLogger entity."""
-        self._accuracy = accuracy or 0
-        self._attributes = attributes
+        self._accuracy = round(accuracy or 0)
+        self._attributes = attributes or {}
         self._name = device
-        self._battery = battery
+        self._battery = None if battery is None else int(battery)
         self._location = location
         self._unique_id = device
-        self._prv_seen = attributes and attributes.get(ATTR_LAST_SEEN)
+        self._prv_seen = cast(datetime | None, self._attributes.get(ATTR_LAST_SEEN))
 
     @property
-    def battery_level(self):
+    def battery_level(self) -> int | None:
         """Return battery value of the device."""
         return self._battery
 
     @property
-    def extra_state_attributes(self):
+    def extra_state_attributes(self) -> Mapping[str, Any] | None:
         """Return device specific attributes."""
         return self._attributes
 
     @property
-    def latitude(self):
+    def latitude(self) -> float | None:
         """Return latitude value of the device."""
-        return self._location[0]
+        return self._location and self._location[0]
 
     @property
-    def longitude(self):
+    def longitude(self) -> float | None:
         """Return longitude value of the device."""
-        return self._location[1]
+        return self._location and self._location[1]
 
     @property
-    def location_accuracy(self):
+    def location_accuracy(self) -> int:
         """Return the gps accuracy of the device."""
         return self._accuracy
 
     @property
-    def unique_id(self):
+    def unique_id(self) -> str | None:
         """Return the unique ID."""
         return self._unique_id
 
@@ -126,7 +139,7 @@ def device_info(self) -> DeviceInfo:
         )
 
     @property
-    def source_type(self) -> SourceType:
+    def source_type(self) -> SourceType | str:
         """Return the source type, eg gps or router, of the device."""
         return SourceType.GPS
 
@@ -144,7 +157,7 @@ async def async_added_to_hass(self) -> None:
             return
 
         if (state := await self.async_get_last_state()) is None:
-            self._location = (None, None)
+            self._location = None
             self._accuracy = 0
             self._attributes = {
                 ATTR_ACTIVITY: None,
@@ -160,8 +173,13 @@ async def async_added_to_hass(self) -> None:
             return
 
         attr = state.attributes
-        self._location = (attr.get(ATTR_LATITUDE), attr.get(ATTR_LONGITUDE))
-        self._accuracy = attr.get(ATTR_GPS_ACCURACY, 0)
+        lat = cast(float | None, attr.get(ATTR_LATITUDE))
+        lon = cast(float | None, attr.get(ATTR_LONGITUDE))
+        if lat is not None and lon is not None:
+            self._location = (lat, lon)
+        else:
+            self._location = None
+        self._accuracy = round(attr.get(ATTR_GPS_ACCURACY) or 0)
         # Python datetime objects are saved/restored as strings.
         # Convert back to datetime object.
         restored_last_seen = cast(str | None, attr.get(ATTR_LAST_SEEN))
@@ -179,10 +197,18 @@ async def async_added_to_hass(self) -> None:
             ATTR_PROVIDER: attr.get(ATTR_PROVIDER),
             ATTR_SPEED: attr.get(ATTR_SPEED),
         }
-        self._battery = attr.get(ATTR_BATTERY_LEVEL)
+        battery = attr.get(ATTR_BATTERY_LEVEL)
+        self._battery = None if battery is None else round(battery)
 
     @callback
-    def _async_receive_data(self, device, location, battery, accuracy, attributes):
+    def _async_receive_data(
+        self,
+        device: str,
+        location: tuple[float, float],
+        battery: float,
+        accuracy: float,
+        attributes: dict[str, Any],
+    ) -> None:
         """Mark the device as seen."""
         if device != self._name:
             return
@@ -198,8 +224,8 @@ def _async_receive_data(self, device, location, battery, accuracy, attributes):
             return
 
         self._location = location
-        self._battery = battery
-        self._accuracy = accuracy or 0
+        self._battery = round(battery)
+        self._accuracy = round(accuracy)
         self._attributes.update(attributes)
         self._prv_seen = last_seen
         self.async_write_ha_state()