Fix Life360 recovery from server errors (#76231)

This commit is contained in:
Phil Bruckner 2022-08-04 14:28:59 -05:00 committed by GitHub
parent 3d42c4ca87
commit 343508a015
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 53 additions and 37 deletions

View File

@ -91,9 +91,11 @@ class Life360Data:
members: dict[str, Life360Member] = field(init=False, default_factory=dict) members: dict[str, Life360Member] = field(init=False, default_factory=dict)
class Life360DataUpdateCoordinator(DataUpdateCoordinator): class Life360DataUpdateCoordinator(DataUpdateCoordinator[Life360Data]):
"""Life360 data update coordinator.""" """Life360 data update coordinator."""
config_entry: ConfigEntry
def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None: def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Initialize data update coordinator.""" """Initialize data update coordinator."""
super().__init__( super().__init__(

View File

@ -11,10 +11,7 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_BATTERY_CHARGING from homeassistant.const import ATTR_BATTERY_CHARGING
from homeassistant.core import HomeAssistant, callback from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import ( from homeassistant.helpers.update_coordinator import CoordinatorEntity
CoordinatorEntity,
DataUpdateCoordinator,
)
from .const import ( from .const import (
ATTR_ADDRESS, ATTR_ADDRESS,
@ -31,6 +28,7 @@ from .const import (
LOGGER, LOGGER,
SHOW_DRIVING, SHOW_DRIVING,
) )
from .coordinator import Life360DataUpdateCoordinator, Life360Member
_LOC_ATTRS = ( _LOC_ATTRS = (
"address", "address",
@ -95,23 +93,27 @@ async def async_setup_entry(
entry.async_on_unload(coordinator.async_add_listener(process_data)) entry.async_on_unload(coordinator.async_add_listener(process_data))
class Life360DeviceTracker(CoordinatorEntity, TrackerEntity): class Life360DeviceTracker(
CoordinatorEntity[Life360DataUpdateCoordinator], TrackerEntity
):
"""Life360 Device Tracker.""" """Life360 Device Tracker."""
_attr_attribution = ATTRIBUTION _attr_attribution = ATTRIBUTION
_attr_unique_id: str
def __init__(self, coordinator: DataUpdateCoordinator, member_id: str) -> None: def __init__(
self, coordinator: Life360DataUpdateCoordinator, member_id: str
) -> None:
"""Initialize Life360 Entity.""" """Initialize Life360 Entity."""
super().__init__(coordinator) super().__init__(coordinator)
self._attr_unique_id = member_id self._attr_unique_id = member_id
self._data = coordinator.data.members[self.unique_id] self._data: Life360Member | None = coordinator.data.members[member_id]
self._prev_data = self._data
self._attr_name = self._data.name self._attr_name = self._data.name
self._attr_entity_picture = self._data.entity_picture self._attr_entity_picture = self._data.entity_picture
self._prev_data = self._data
@property @property
def _options(self) -> Mapping[str, Any]: def _options(self) -> Mapping[str, Any]:
"""Shortcut to config entry options.""" """Shortcut to config entry options."""
@ -120,16 +122,15 @@ class Life360DeviceTracker(CoordinatorEntity, TrackerEntity):
@callback @callback
def _handle_coordinator_update(self) -> None: def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator.""" """Handle updated data from the coordinator."""
# Get a shortcut to this member's data. Can't guarantee it's the same dict every # Get a shortcut to this Member's data. This needs to be updated each time since
# update, or that there is even data for this member every update, so need to # coordinator provides a new Life360Member object each time, and it's possible
# update shortcut each time. # that there is no data for this Member on some updates.
self._data = self.coordinator.data.members.get(self.unique_id)
if self.available: if self.available:
# If nothing important has changed, then skip the update altogether. self._data = self.coordinator.data.members.get(self._attr_unique_id)
if self._data == self._prev_data: else:
return self._data = None
if self._data:
# Check if we should effectively throw out new location data. # Check if we should effectively throw out new location data.
last_seen = self._data.last_seen last_seen = self._data.last_seen
prev_seen = self._prev_data.last_seen prev_seen = self._prev_data.last_seen
@ -168,27 +169,21 @@ class Life360DeviceTracker(CoordinatorEntity, TrackerEntity):
"""Return True if state updates should be forced.""" """Return True if state updates should be forced."""
return False return False
@property
def available(self) -> bool:
"""Return if entity is available."""
# Guard against member not being in last update for some reason.
return super().available and self._data is not None
@property @property
def entity_picture(self) -> str | None: def entity_picture(self) -> str | None:
"""Return the entity picture to use in the frontend, if any.""" """Return the entity picture to use in the frontend, if any."""
if self.available: if self._data:
self._attr_entity_picture = self._data.entity_picture self._attr_entity_picture = self._data.entity_picture
return super().entity_picture return super().entity_picture
# All of the following will only be called if self.available is True.
@property @property
def battery_level(self) -> int | None: def battery_level(self) -> int | None:
"""Return the battery level of the device. """Return the battery level of the device.
Percentage from 0-100. Percentage from 0-100.
""" """
if not self._data:
return None
return self._data.battery_level return self._data.battery_level
@property @property
@ -202,11 +197,15 @@ class Life360DeviceTracker(CoordinatorEntity, TrackerEntity):
Value in meters. Value in meters.
""" """
if not self._data:
return 0
return self._data.gps_accuracy return self._data.gps_accuracy
@property @property
def driving(self) -> bool: def driving(self) -> bool:
"""Return if driving.""" """Return if driving."""
if not self._data:
return False
if (driving_speed := self._options.get(CONF_DRIVING_SPEED)) is not None: if (driving_speed := self._options.get(CONF_DRIVING_SPEED)) is not None:
if self._data.speed >= driving_speed: if self._data.speed >= driving_speed:
return True return True
@ -222,23 +221,38 @@ class Life360DeviceTracker(CoordinatorEntity, TrackerEntity):
@property @property
def latitude(self) -> float | None: def latitude(self) -> float | None:
"""Return latitude value of the device.""" """Return latitude value of the device."""
if not self._data:
return None
return self._data.latitude return self._data.latitude
@property @property
def longitude(self) -> float | None: def longitude(self) -> float | None:
"""Return longitude value of the device.""" """Return longitude value of the device."""
if not self._data:
return None
return self._data.longitude return self._data.longitude
@property @property
def extra_state_attributes(self) -> Mapping[str, Any] | None: def extra_state_attributes(self) -> Mapping[str, Any] | None:
"""Return entity specific state attributes.""" """Return entity specific state attributes."""
attrs = {} if not self._data:
attrs[ATTR_ADDRESS] = self._data.address return {
attrs[ATTR_AT_LOC_SINCE] = self._data.at_loc_since ATTR_ADDRESS: None,
attrs[ATTR_BATTERY_CHARGING] = self._data.battery_charging ATTR_AT_LOC_SINCE: None,
attrs[ATTR_DRIVING] = self.driving ATTR_BATTERY_CHARGING: None,
attrs[ATTR_LAST_SEEN] = self._data.last_seen ATTR_DRIVING: None,
attrs[ATTR_PLACE] = self._data.place ATTR_LAST_SEEN: None,
attrs[ATTR_SPEED] = self._data.speed ATTR_PLACE: None,
attrs[ATTR_WIFI_ON] = self._data.wifi_on ATTR_SPEED: None,
return attrs ATTR_WIFI_ON: None,
}
return {
ATTR_ADDRESS: self._data.address,
ATTR_AT_LOC_SINCE: self._data.at_loc_since,
ATTR_BATTERY_CHARGING: self._data.battery_charging,
ATTR_DRIVING: self.driving,
ATTR_LAST_SEEN: self._data.last_seen,
ATTR_PLACE: self._data.place,
ATTR_SPEED: self._data.speed,
ATTR_WIFI_ON: self._data.wifi_on,
}