jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-25 22:27:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove Life360 integration (#107805)

Browse Source
This commit is contained in:
Phil Bruckner 2024-01-11 14:49:39 -06:00 committed by GitHub
parent ff811a33f5
commit 24ddc939c0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 81 additions and 1303 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.coveragerc
View File

@ -664,10 +664,6 @@ omit =
homeassistant/components/lg_netcast/media_player.py homeassistant/components/lg_netcast/media_player.py
homeassistant/components/lg_soundbar/__init__.py homeassistant/components/lg_soundbar/__init__.py
homeassistant/components/lg_soundbar/media_player.py homeassistant/components/lg_soundbar/media_player.py
homeassistant/components/life360/__init__.py
homeassistant/components/life360/button.py
homeassistant/components/life360/coordinator.py
homeassistant/components/life360/device_tracker.py
homeassistant/components/lightwave/* homeassistant/components/lightwave/*
homeassistant/components/limitlessled/light.py homeassistant/components/limitlessled/light.py
homeassistant/components/linksys_smart/device_tracker.py homeassistant/components/linksys_smart/device_tracker.py

2
CODEOWNERS
View File

@ -706,8 +706,6 @@ build.json @home-assistant/supervisor
/homeassistant/components/lg_netcast/ @Drafteed /homeassistant/components/lg_netcast/ @Drafteed
/homeassistant/components/lidarr/ @tkdrob /homeassistant/components/lidarr/ @tkdrob
/tests/components/lidarr/ @tkdrob /tests/components/lidarr/ @tkdrob
/homeassistant/components/life360/ @pnbruckner
/tests/components/life360/ @pnbruckner
/homeassistant/components/light/ @home-assistant/core /homeassistant/components/light/ @home-assistant/core
/tests/components/light/ @home-assistant/core /tests/components/light/ @home-assistant/core
/homeassistant/components/linear_garage_door/ @IceBotYT /homeassistant/components/linear_garage_door/ @IceBotYT

68
homeassistant/components/life360/__init__.py
View File

@ -2,59 +2,35 @@
from __future__ import annotations from __future__ import annotations
from dataclasses import dataclass, field from homeassistant.config_entries import ConfigEntry, ConfigEntryState
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers import issue_registry as ir
from .const import DOMAIN DOMAIN = "life360"
from .coordinator import Life360DataUpdateCoordinator, MissingLocReason
PLATFORMS = [Platform.DEVICE_TRACKER, Platform.BUTTON]
@dataclass async def async_setup_entry(hass: HomeAssistant, _: ConfigEntry) -> bool:
class IntegData:
"""Integration data."""
# ConfigEntry.entry_id: Life360DataUpdateCoordinator
coordinators: dict[str, Life360DataUpdateCoordinator] = field(
init=False, default_factory=dict
)
# member_id: missing location reason
missing_loc_reason: dict[str, MissingLocReason] = field(
init=False, default_factory=dict
)
# member_id: ConfigEntry.entry_id
tracked_members: dict[str, str] = field(init=False, default_factory=dict)
logged_circles: list[str] = field(init=False, default_factory=list)
logged_places: list[str] = field(init=False, default_factory=list)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up config entry.""" """Set up config entry."""
hass.data.setdefault(DOMAIN, IntegData()) ir.async_create_issue(
hass,
coordinator = Life360DataUpdateCoordinator(hass, entry) DOMAIN,
await coordinator.async_config_entry_first_refresh() DOMAIN,
hass.data[DOMAIN].coordinators[entry.entry_id] = coordinator is_fixable=False,
severity=ir.IssueSeverity.ERROR,
# Set up components for our platforms. translation_key="integration_removed",
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) translation_placeholders={
"entries": "/config/integrations/integration/life360"
},
)
return True return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload config entry.""" """Unload config entry."""
if all(
# Unload components for our platforms. config_entry.state is ConfigEntryState.NOT_LOADED
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS): for config_entry in hass.config_entries.async_entries(DOMAIN)
del hass.data[DOMAIN].coordinators[entry.entry_id] if config_entry.entry_id != entry.entry_id
# Remove any members that were tracked by this entry. ):
for member_id, entry_id in hass.data[DOMAIN].tracked_members.copy().items(): ir.async_delete_issue(hass, DOMAIN, DOMAIN)
if entry_id == entry.entry_id: return True
del hass.data[DOMAIN].tracked_members[member_id]
return unload_ok

54
homeassistant/components/life360/button.py
View File

@ -1,54 +0,0 @@
"""Support for Life360 buttons."""
from homeassistant.components.button import ButtonEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import Life360DataUpdateCoordinator
from .const import DOMAIN
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up Life360 buttons."""
coordinator: Life360DataUpdateCoordinator = hass.data[DOMAIN].coordinators[
config_entry.entry_id
]
async_add_entities(
Life360UpdateLocationButton(coordinator, member.circle_id, member_id)
for member_id, member in coordinator.data.members.items()
)
class Life360UpdateLocationButton(
CoordinatorEntity[Life360DataUpdateCoordinator], ButtonEntity
):
"""Represent an Life360 Update Location button."""
_attr_has_entity_name = True
_attr_translation_key = "update_location"
def __init__(
self,
coordinator: Life360DataUpdateCoordinator,
circle_id: str,
member_id: str,
) -> None:
"""Initialize a new Life360 Update Location button."""
super().__init__(coordinator)
self._circle_id = circle_id
self._member_id = member_id
self._attr_unique_id = f"{member_id}-update-location"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, member_id)},
name=coordinator.data.members[member_id].name,
)
async def async_press(self) -> None:
"""Handle the button press."""
await self.coordinator.update_location(self._circle_id, self._member_id)

197
homeassistant/components/life360/config_flow.py
View File

@ -2,205 +2,12 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import Mapping from homeassistant.config_entries import ConfigFlow
from typing import Any, cast
from life360 import Life360, Life360Error, LoginError from . import DOMAIN
import voluptuous as vol
from homeassistant.config_entries import ConfigEntry, ConfigFlow, OptionsFlow
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
from .const import (
COMM_TIMEOUT,
CONF_AUTHORIZATION,
CONF_DRIVING_SPEED,
CONF_MAX_GPS_ACCURACY,
DEFAULT_OPTIONS,
DOMAIN,
LOGGER,
OPTIONS,
SHOW_DRIVING,
)
LIMIT_GPS_ACC = "limit_gps_acc"
SET_DRIVE_SPEED = "set_drive_speed"
def account_schema(
def_username: str | vol.UNDEFINED = vol.UNDEFINED,
def_password: str | vol.UNDEFINED = vol.UNDEFINED,
) -> dict[vol.Marker, Any]:
"""Return schema for an account with optional default values."""
return {
vol.Required(CONF_USERNAME, default=def_username): cv.string,
vol.Required(CONF_PASSWORD, default=def_password): cv.string,
}
def password_schema(
def_password: str | vol.UNDEFINED = vol.UNDEFINED,
) -> dict[vol.Marker, Any]:
"""Return schema for a password with optional default value."""
return {vol.Required(CONF_PASSWORD, default=def_password): cv.string}
class Life360ConfigFlow(ConfigFlow, domain=DOMAIN): class Life360ConfigFlow(ConfigFlow, domain=DOMAIN):
"""Life360 integration config flow.""" """Life360 integration config flow."""
VERSION = 1 VERSION = 1
_api: Life360 | None = None
_username: str | vol.UNDEFINED = vol.UNDEFINED
_password: str | vol.UNDEFINED = vol.UNDEFINED
_reauth_entry: ConfigEntry | None = None
@staticmethod
@callback
def async_get_options_flow(config_entry: ConfigEntry) -> Life360OptionsFlow:
"""Get the options flow for this handler."""
return Life360OptionsFlow(config_entry)
async def _async_verify(self, step_id: str) -> FlowResult:
"""Attempt to authorize the provided credentials."""
if not self._api:
self._api = Life360(
session=async_get_clientsession(self.hass), timeout=COMM_TIMEOUT
)
errors: dict[str, str] = {}
try:
authorization = await self._api.get_authorization(
self._username, self._password
)
except LoginError as exc:
LOGGER.debug("Login error: %s", exc)
errors["base"] = "invalid_auth"
except Life360Error as exc:
LOGGER.debug("Unexpected error communicating with Life360 server: %s", exc)
errors["base"] = "cannot_connect"
if errors:
if step_id == "user":
schema = account_schema(self._username, self._password)
else:
schema = password_schema(self._password)
return self.async_show_form(
step_id=step_id, data_schema=vol.Schema(schema), errors=errors
)
data = {
CONF_USERNAME: self._username,
CONF_PASSWORD: self._password,
CONF_AUTHORIZATION: authorization,
}
if self._reauth_entry:
LOGGER.debug("Reauthorization successful")
self.hass.config_entries.async_update_entry(self._reauth_entry, data=data)
self.hass.async_create_task(
self.hass.config_entries.async_reload(self._reauth_entry.entry_id)
)
return self.async_abort(reason="reauth_successful")
return self.async_create_entry(
title=cast(str, self.unique_id), data=data, options=DEFAULT_OPTIONS
)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a config flow initiated by the user."""
if not user_input:
return self.async_show_form(
step_id="user", data_schema=vol.Schema(account_schema())
)
self._username = user_input[CONF_USERNAME]
self._password = user_input[CONF_PASSWORD]
await self.async_set_unique_id(self._username.lower())
self._abort_if_unique_id_configured()
return await self._async_verify("user")
async def async_step_reauth(self, data: Mapping[str, Any]) -> FlowResult:
"""Handle reauthorization."""
self._username = data[CONF_USERNAME]
self._reauth_entry = self.hass.config_entries.async_get_entry(
self.context["entry_id"]
)
# Always start with current credentials since they may still be valid and a
# simple reauthorization will be successful.
return await self.async_step_reauth_confirm(dict(data))
async def async_step_reauth_confirm(self, user_input: dict[str, Any]) -> FlowResult:
"""Handle reauthorization completion."""
if not user_input:
return self.async_show_form(
step_id="reauth_confirm",
data_schema=vol.Schema(password_schema(self._password)),
errors={"base": "invalid_auth"},
)
self._password = user_input[CONF_PASSWORD]
return await self._async_verify("reauth_confirm")
class Life360OptionsFlow(OptionsFlow):
"""Life360 integration options flow."""
def __init__(self, config_entry: ConfigEntry) -> None:
"""Initialize."""
self.config_entry = config_entry
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle account options."""
options = self.config_entry.options
if user_input is not None:
new_options = _extract_account_options(user_input)
return self.async_create_entry(title="", data=new_options)
return self.async_show_form(
step_id="init", data_schema=vol.Schema(_account_options_schema(options))
)
def _account_options_schema(options: Mapping[str, Any]) -> dict[vol.Marker, Any]:
"""Create schema for account options form."""
def_limit_gps_acc = options[CONF_MAX_GPS_ACCURACY] is not None
def_max_gps = options[CONF_MAX_GPS_ACCURACY] or vol.UNDEFINED
def_set_drive_speed = options[CONF_DRIVING_SPEED] is not None
def_speed = options[CONF_DRIVING_SPEED] or vol.UNDEFINED
def_show_driving = options[SHOW_DRIVING]
return {
vol.Required(LIMIT_GPS_ACC, default=def_limit_gps_acc): bool,
vol.Optional(CONF_MAX_GPS_ACCURACY, default=def_max_gps): vol.Coerce(float),
vol.Required(SET_DRIVE_SPEED, default=def_set_drive_speed): bool,
vol.Optional(CONF_DRIVING_SPEED, default=def_speed): vol.Coerce(float),
vol.Optional(SHOW_DRIVING, default=def_show_driving): bool,
}
def _extract_account_options(user_input: dict) -> dict[str, Any]:
"""Remove options from user input and return as a separate dict."""
result = {}
for key in OPTIONS:
value = user_input.pop(key, None)
# Was "include" checkbox (if there was one) corresponding to option key True
# (meaning option should be included)?
incl = user_input.pop(
{
CONF_MAX_GPS_ACCURACY: LIMIT_GPS_ACC,
CONF_DRIVING_SPEED: SET_DRIVE_SPEED,
}.get(key),
True,
)
result[key] = value if incl else None
return result

37
homeassistant/components/life360/const.py
View File

@ -1,37 +0,0 @@
"""Constants for Life360 integration."""
from datetime import timedelta
import logging
from aiohttp import ClientTimeout
DOMAIN = "life360"
LOGGER = logging.getLogger(__package__)
ATTRIBUTION = "Data provided by life360.com"
COMM_MAX_RETRIES = 3
COMM_TIMEOUT = ClientTimeout(sock_connect=15, total=60)
SPEED_FACTOR_MPH = 2.25
SPEED_DIGITS = 1
UPDATE_INTERVAL = timedelta(seconds=10)
ATTR_ADDRESS = "address"
ATTR_AT_LOC_SINCE = "at_loc_since"
ATTR_DRIVING = "driving"
ATTR_LAST_SEEN = "last_seen"
ATTR_PLACE = "place"
ATTR_SPEED = "speed"
ATTR_WIFI_ON = "wifi_on"
CONF_AUTHORIZATION = "authorization"
CONF_DRIVING_SPEED = "driving_speed"
CONF_MAX_GPS_ACCURACY = "max_gps_accuracy"
SHOW_DRIVING = "driving"
DEFAULT_OPTIONS = {
CONF_DRIVING_SPEED: None,
CONF_MAX_GPS_ACCURACY: None,
SHOW_DRIVING: False,
}
OPTIONS = list(DEFAULT_OPTIONS.keys())

246
homeassistant/components/life360/coordinator.py
View File

@ -1,246 +0,0 @@
"""DataUpdateCoordinator for the Life360 integration."""
from __future__ import annotations
import asyncio
from contextlib import suppress
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any
from life360 import Life360, Life360Error, LoginError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import UnitOfLength
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.util import dt as dt_util
from homeassistant.util.unit_conversion import DistanceConverter
from homeassistant.util.unit_system import METRIC_SYSTEM
from .const import (
COMM_MAX_RETRIES,
COMM_TIMEOUT,
CONF_AUTHORIZATION,
DOMAIN,
LOGGER,
SPEED_DIGITS,
SPEED_FACTOR_MPH,
UPDATE_INTERVAL,
)
class MissingLocReason(Enum):
"""Reason member location information is missing."""
VAGUE_ERROR_REASON = "vague error reason"
EXPLICIT_ERROR_REASON = "explicit error reason"
@dataclass
class Life360Place:
"""Life360 Place data."""
name: str
latitude: float
longitude: float
radius: float
@dataclass
class Life360Circle:
"""Life360 Circle data."""
name: str
places: dict[str, Life360Place]
@dataclass
class Life360Member:
"""Life360 Member data."""
address: str | None
at_loc_since: datetime
battery_charging: bool
battery_level: int
circle_id: str
driving: bool
entity_picture: str
gps_accuracy: int
last_seen: datetime
latitude: float
longitude: float
name: str
place: str | None
speed: float
wifi_on: bool
@dataclass
class Life360Data:
"""Life360 data."""
circles: dict[str, Life360Circle] = field(init=False, default_factory=dict)
members: dict[str, Life360Member] = field(init=False, default_factory=dict)
class Life360DataUpdateCoordinator(DataUpdateCoordinator[Life360Data]):
"""Life360 data update coordinator."""
config_entry: ConfigEntry
def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Initialize data update coordinator."""
super().__init__(
hass,
LOGGER,
name=f"{DOMAIN} ({entry.unique_id})",
update_interval=UPDATE_INTERVAL,
)
self._hass = hass
self._api = Life360(
session=async_get_clientsession(hass),
timeout=COMM_TIMEOUT,
max_retries=COMM_MAX_RETRIES,
authorization=entry.data[CONF_AUTHORIZATION],
)
self._missing_loc_reason = hass.data[DOMAIN].missing_loc_reason
async def _retrieve_data(self, func: str, *args: Any) -> list[dict[str, Any]]:
"""Get data from Life360."""
try:
return await getattr(self._api, func)(*args)
except LoginError as exc:
LOGGER.debug("Login error: %s", exc)
raise ConfigEntryAuthFailed(exc) from exc
except Life360Error as exc:
LOGGER.debug("%s: %s", exc.__class__.__name__, exc)
raise UpdateFailed(exc) from exc
async def update_location(self, circle_id: str, member_id: str) -> None:
"""Update location for given Circle and Member."""
await self._retrieve_data("update_location", circle_id, member_id)
async def _async_update_data(self) -> Life360Data:
"""Get & process data from Life360."""
data = Life360Data()
for circle in await self._retrieve_data("get_circles"):
circle_id = circle["id"]
circle_members, circle_places = await asyncio.gather(
self._retrieve_data("get_circle_members", circle_id),
self._retrieve_data("get_circle_places", circle_id),
)
data.circles[circle_id] = Life360Circle(
circle["name"],
{
place["id"]: Life360Place(
place["name"],
float(place["latitude"]),
float(place["longitude"]),
float(place["radius"]),
)
for place in circle_places
},
)
for member in circle_members:
# Member isn't sharing location.
if not int(member["features"]["shareLocation"]):
continue
member_id = member["id"]
first = member["firstName"]
last = member["lastName"]
if first and last:
name = " ".join([first, last])
else:
name = first or last
cur_missing_reason = self._missing_loc_reason.get(member_id)
# Check if location information is missing. This can happen if server
# has not heard from member's device in a long time (e.g., has been off
# for a long time, or has lost service, etc.)
if loc := member["location"]:
with suppress(KeyError):
del self._missing_loc_reason[member_id]
else:
if explicit_reason := member["issues"]["title"]:
if extended_reason := member["issues"]["dialog"]:
explicit_reason += f": {extended_reason}"
# Note that different Circles can report missing location in
# different ways. E.g., one might report an explicit reason and
# another does not. If a vague reason has already been logged but a
# more explicit reason is now available, log that, too.
if (
cur_missing_reason is None
or cur_missing_reason == MissingLocReason.VAGUE_ERROR_REASON
and explicit_reason
):
if explicit_reason:
self._missing_loc_reason[
member_id
] = MissingLocReason.EXPLICIT_ERROR_REASON
err_msg = explicit_reason
else:
self._missing_loc_reason[
member_id
] = MissingLocReason.VAGUE_ERROR_REASON
err_msg = "Location information missing"
LOGGER.error("%s: %s", name, err_msg)
continue
# Note that member may be in more than one circle. If that's the case
# just go ahead and process the newly retrieved data (overwriting the
# older data), since it might be slightly newer than what was retrieved
# while processing another circle.
place = loc["name"] or None
address1: str | None = loc["address1"] or None
address2: str | None = loc["address2"] or None
if address1 and address2:
address: str | None = ", ".join([address1, address2])
else:
address = address1 or address2
speed = max(0, float(loc["speed"]) * SPEED_FACTOR_MPH)
if self._hass.config.units is METRIC_SYSTEM:
speed = DistanceConverter.convert(
speed, UnitOfLength.MILES, UnitOfLength.KILOMETERS
)
data.members[member_id] = Life360Member(
address,
dt_util.utc_from_timestamp(int(loc["since"])),
bool(int(loc["charge"])),
int(float(loc["battery"])),
circle_id,
bool(int(loc["isDriving"])),
member["avatar"],
# Life360 reports accuracy in feet, but Device Tracker expects
# gps_accuracy in meters.
round(
DistanceConverter.convert(
float(loc["accuracy"]),
UnitOfLength.FEET,
UnitOfLength.METERS,
)
),
dt_util.utc_from_timestamp(int(loc["timestamp"])),
float(loc["latitude"]),
float(loc["longitude"]),
name,
place,
round(speed, SPEED_DIGITS),
bool(int(loc["wifiState"])),
)
return data

326
homeassistant/components/life360/device_tracker.py
View File

@ -1,326 +0,0 @@
"""Support for Life360 device tracking."""
from __future__ import annotations
from collections.abc import Mapping
from contextlib import suppress
from typing import Any, cast
from homeassistant.components.device_tracker import SourceType, TrackerEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_BATTERY_CHARGING
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import (
ATTR_ADDRESS,
ATTR_AT_LOC_SINCE,
ATTR_DRIVING,
ATTR_LAST_SEEN,
ATTR_PLACE,
ATTR_SPEED,
ATTR_WIFI_ON,
ATTRIBUTION,
CONF_DRIVING_SPEED,
CONF_MAX_GPS_ACCURACY,
DOMAIN,
LOGGER,
SHOW_DRIVING,
)
from .coordinator import Life360DataUpdateCoordinator, Life360Member
_LOC_ATTRS = (
"address",
"at_loc_since",
"driving",
"gps_accuracy",
"last_seen",
"latitude",
"longitude",
"place",
"speed",
)
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up the device tracker platform."""
coordinator = hass.data[DOMAIN].coordinators[entry.entry_id]
tracked_members = hass.data[DOMAIN].tracked_members
logged_circles = hass.data[DOMAIN].logged_circles
logged_places = hass.data[DOMAIN].logged_places
@callback
def process_data(new_members_only: bool = True) -> None:
"""Process new Life360 data."""
for circle_id, circle in coordinator.data.circles.items():
if circle_id not in logged_circles:
logged_circles.append(circle_id)
LOGGER.debug("Circle: %s", circle.name)
new_places = []
for place_id, place in circle.places.items():
if place_id not in logged_places:
logged_places.append(place_id)
new_places.append(place)
if new_places:
msg = f"Places from {circle.name}:"
for place in new_places:
msg += f"\n- name: {place.name}"
msg += f"\n latitude: {place.latitude}"
msg += f"\n longitude: {place.longitude}"
msg += f"\n radius: {place.radius}"
LOGGER.debug(msg)
new_entities = []
for member_id, member in coordinator.data.members.items():
tracked_by_entry = tracked_members.get(member_id)
if new_member := not tracked_by_entry:
tracked_members[member_id] = entry.entry_id
LOGGER.debug("Member: %s (%s)", member.name, entry.unique_id)
if (
new_member
or tracked_by_entry == entry.entry_id
and not new_members_only
):
new_entities.append(Life360DeviceTracker(coordinator, member_id))
async_add_entities(new_entities)
process_data(new_members_only=False)
entry.async_on_unload(coordinator.async_add_listener(process_data))
class Life360DeviceTracker(
CoordinatorEntity[Life360DataUpdateCoordinator], TrackerEntity
):
"""Life360 Device Tracker."""
_attr_attribution = ATTRIBUTION
_attr_unique_id: str
_attr_has_entity_name = True
_attr_name = None
def __init__(
self, coordinator: Life360DataUpdateCoordinator, member_id: str
) -> None:
"""Initialize Life360 Entity."""
super().__init__(coordinator)
self._attr_unique_id = member_id
self._data: Life360Member | None = coordinator.data.members[member_id]
self._prev_data = self._data
self._name = self._data.name
self._attr_entity_picture = self._data.entity_picture
# Server sends a pair of address values on alternate updates. Keep the pair of
# values so they can be combined into the one address attribute.
# The pair will either be two different address values, or one address and a
# copy of the Place value (if the Member is in a Place.) In the latter case we
# won't duplicate the Place name, but rather just use one the address value. Use
# the value of None to hold one of the "slots" in the list so we'll know not to
# expect another address value.
if (address := self._data.address) == self._data.place:
address = None
self._addresses = [address]
@property
def device_info(self) -> DeviceInfo:
"""Return device info."""
return DeviceInfo(identifiers={(DOMAIN, self._attr_unique_id)}, name=self._name)
@property
def _options(self) -> Mapping[str, Any]:
"""Shortcut to config entry options."""
return cast(Mapping[str, Any], self.coordinator.config_entry.options)
@callback
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
# Get a shortcut to this Member's data. This needs to be updated each time since
# coordinator provides a new Life360Member object each time, and it's possible
# that there is no data for this Member on some updates.
if self.available:
self._data = self.coordinator.data.members.get(self._attr_unique_id)
else:
self._data = None
if self._data:
# Check if we should effectively throw out new location data.
last_seen = self._data.last_seen
prev_seen = self._prev_data.last_seen
max_gps_acc = self._options.get(CONF_MAX_GPS_ACCURACY)
bad_last_seen = last_seen < prev_seen
bad_accuracy = (
max_gps_acc is not None and self.location_accuracy > max_gps_acc
)
if bad_last_seen or bad_accuracy:
if bad_last_seen:
LOGGER.warning(
(
"%s: Ignoring location update because "
"last_seen (%s) < previous last_seen (%s)"
),
self.entity_id,
last_seen,
prev_seen,
)
if bad_accuracy:
LOGGER.warning(
(
"%s: Ignoring location update because "
"expected GPS accuracy (%0.1f) is not met: %i"
),
self.entity_id,
max_gps_acc,
self.location_accuracy,
)
# Overwrite new location related data with previous values.
for attr in _LOC_ATTRS:
setattr(self._data, attr, getattr(self._prev_data, attr))
else:
# Process address field.
# Check if we got the name of a Place, which we won't use.
if (address := self._data.address) == self._data.place:
address = None
if last_seen != prev_seen:
# We have new location data, so we might have a new pair of address
# values.
if address not in self._addresses:
# We do.
# Replace the old values with the first value of the new pair.
self._addresses = [address]
elif self._data.address != self._prev_data.address:
# Location data didn't change in general, but the address field did.
# There are three possibilities:
# 1. The new value is one of the pair we've already seen before.
# 2. The new value is the second of the pair we haven't seen yet.
# 3. The new value is the first of a new pair of values.
if address not in self._addresses:
if len(self._addresses) < 2:
self._addresses.append(address)
else:
self._addresses = [address]
self._prev_data = self._data
super()._handle_coordinator_update()
@property
def force_update(self) -> bool:
"""Return True if state updates should be forced.
Overridden because CoordinatorEntity sets `should_poll` to False,
which causes TrackerEntity to set `force_update` to True.
"""
return False
@property
def entity_picture(self) -> str | None:
"""Return the entity picture to use in the frontend, if any."""
if self._data:
self._attr_entity_picture = self._data.entity_picture
return super().entity_picture
@property
def battery_level(self) -> int | None:
"""Return the battery level of the device.
Percentage from 0-100.
"""
if not self._data:
return None
return self._data.battery_level
@property
def source_type(self) -> SourceType:
"""Return the source type, eg gps or router, of the device."""
return SourceType.GPS
@property
def location_accuracy(self) -> int:
"""Return the location accuracy of the device.
Value in meters.
"""
if not self._data:
return 0
return self._data.gps_accuracy
@property
def driving(self) -> bool:
"""Return if driving."""
if not self._data:
return False
if (driving_speed := self._options.get(CONF_DRIVING_SPEED)) is not None:
if self._data.speed >= driving_speed:
return True
return self._data.driving
@property
def location_name(self) -> str | None:
"""Return a location name for the current location of the device."""
if self._options.get(SHOW_DRIVING) and self.driving:
return "Driving"
return None
@property
def latitude(self) -> float | None:
"""Return latitude value of the device."""
if not self._data:
return None
return self._data.latitude
@property
def longitude(self) -> float | None:
"""Return longitude value of the device."""
if not self._data:
return None
return self._data.longitude
@property
def extra_state_attributes(self) -> Mapping[str, Any] | None:
"""Return entity specific state attributes."""
if not self._data:
return {
ATTR_ADDRESS: None,
ATTR_AT_LOC_SINCE: None,
ATTR_BATTERY_CHARGING: None,
ATTR_DRIVING: None,
ATTR_LAST_SEEN: None,
ATTR_PLACE: None,
ATTR_SPEED: None,
ATTR_WIFI_ON: None,
}
# Generate address attribute from pair of address values.
# There may be two, one or no values. If there are two, sort the strings since
# one value is typically a numbered street address and the other is a street,
# town or state name, and it's helpful to start with the more detailed address
# value. Also, sorting helps to generate the same result if we get a location
# update, and the same pair is sent afterwards, but where the value that comes
# first is swapped vs the order they came in before the update.
address1: str | None = None
address2: str | None = None
with suppress(IndexError):
address1 = self._addresses[0]
address2 = self._addresses[1]
if address1 and address2:
address: str | None = " / ".join(sorted([address1, address2]))
else:
address = address1 or address2
return {
ATTR_ADDRESS: 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,
}

7
homeassistant/components/life360/manifest.json
View File

@ -1,10 +1,9 @@
{ {
"domain": "life360", "domain": "life360",
"name": "Life360", "name": "Life360",
"codeowners": ["@pnbruckner"], "codeowners": [],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/life360", "documentation": "https://www.home-assistant.io/integrations/life360",
"integration_type": "system",
"iot_class": "cloud_polling", "iot_class": "cloud_polling",
"loggers": ["life360"], "requirements": []
"requirements": ["life360==6.0.1"]
} }

51
homeassistant/components/life360/strings.json
View File

@ -1,51 +1,8 @@
{ {
"config": { "issues": {
"step": { "integration_removed": {
"user": { "title": "The Life360 integration has been removed",
"title": "Configure Life360 Account", "description": "The Life360 integration has been removed from Home Assistant.\n\nLife360 has blocked all third-party integrations.\n\nTo resolve this issue, please remove the (now defunct) integration entries from your Home Assistant setup. [Click here to see your existing Life360 integration entries]({entries})."
"data": {
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]"
}
},
"reauth_confirm": {
"title": "[%key:common::config_flow::title::reauth%]",
"data": {
"password": "[%key:common::config_flow::data::password%]"
}
}
},
"error": {
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"abort": {
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
}
},
"entity": {
"button": {
"update_location": {
"name": "Update Location"
}
}
},
"options": {
"step": {
"init": {
"title": "Account Options",
"data": {
"limit_gps_acc": "Limit GPS accuracy",
"max_gps_accuracy": "Max GPS accuracy (meters)",
"set_drive_speed": "Set driving speed threshold",
"driving_speed": "Driving speed",
"driving": "Show driving as state"
}
}
} }
} }
} }

1
homeassistant/generated/config_flows.py
View File

@ -268,7 +268,6 @@ FLOWS = {
"led_ble", "led_ble",
"lg_soundbar", "lg_soundbar",
"lidarr", "lidarr",
"life360",
"lifx", "lifx",
"linear_garage_door", "linear_garage_door",
"litejet", "litejet",

6
homeassistant/generated/integrations.json
View File

@ -3093,12 +3093,6 @@
"config_flow": true, "config_flow": true,
"iot_class": "local_polling" "iot_class": "local_polling"
}, },
"life360": {
"name": "Life360",
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_polling"
},
"lifx": { "lifx": {
"name": "LIFX", "name": "LIFX",
"integration_type": "hub", "integration_type": "hub",

3
requirements_all.txt
View File

@ -1191,9 +1191,6 @@ librouteros==3.2.0
# homeassistant.components.soundtouch # homeassistant.components.soundtouch
libsoundtouch==0.8 libsoundtouch==0.8
# homeassistant.components.life360
life360==6.0.1
# homeassistant.components.osramlightify # homeassistant.components.osramlightify
lightify==1.0.7.3 lightify==1.0.7.3

3
requirements_test_all.txt
View File

@ -948,9 +948,6 @@ librouteros==3.2.0
# homeassistant.components.soundtouch # homeassistant.components.soundtouch
libsoundtouch==0.8 libsoundtouch==0.8
# homeassistant.components.life360
life360==6.0.1
# homeassistant.components.linear_garage_door # homeassistant.components.linear_garage_door
linear-garage-door==0.2.7 linear-garage-door==0.2.7

329
tests/components/life360/test_config_flow.py
View File

@ -1,329 +0,0 @@
"""Test the Life360 config flow."""
from unittest.mock import patch
from life360 import Life360Error, LoginError
import pytest
import voluptuous as vol
from homeassistant import config_entries, data_entry_flow
from homeassistant.components.life360.const import (
CONF_AUTHORIZATION,
CONF_DRIVING_SPEED,
CONF_MAX_GPS_ACCURACY,
DEFAULT_OPTIONS,
DOMAIN,
SHOW_DRIVING,
)
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
TEST_USER = "Test@Test.com"
TEST_PW = "password"
TEST_PW_3 = "password_3"
TEST_AUTHORIZATION = "authorization_string"
TEST_AUTHORIZATION_2 = "authorization_string_2"
TEST_AUTHORIZATION_3 = "authorization_string_3"
TEST_MAX_GPS_ACCURACY = "300"
TEST_DRIVING_SPEED = "18"
TEST_SHOW_DRIVING = True
USER_INPUT = {CONF_USERNAME: TEST_USER, CONF_PASSWORD: TEST_PW}
TEST_CONFIG_DATA = {
CONF_USERNAME: TEST_USER,
CONF_PASSWORD: TEST_PW,
CONF_AUTHORIZATION: TEST_AUTHORIZATION,
}
TEST_CONFIG_DATA_2 = {
CONF_USERNAME: TEST_USER,
CONF_PASSWORD: TEST_PW,
CONF_AUTHORIZATION: TEST_AUTHORIZATION_2,
}
TEST_CONFIG_DATA_3 = {
CONF_USERNAME: TEST_USER,
CONF_PASSWORD: TEST_PW_3,
CONF_AUTHORIZATION: TEST_AUTHORIZATION_3,
}
USER_OPTIONS = {
"limit_gps_acc": True,
CONF_MAX_GPS_ACCURACY: TEST_MAX_GPS_ACCURACY,
"set_drive_speed": True,
CONF_DRIVING_SPEED: TEST_DRIVING_SPEED,
SHOW_DRIVING: TEST_SHOW_DRIVING,
}
TEST_OPTIONS = {
CONF_MAX_GPS_ACCURACY: float(TEST_MAX_GPS_ACCURACY),
CONF_DRIVING_SPEED: float(TEST_DRIVING_SPEED),
SHOW_DRIVING: TEST_SHOW_DRIVING,
}
# ========== Common Fixtures & Functions ===============================================
@pytest.fixture(name="life360", autouse=True)
def life360_fixture():
"""Mock life360 config entry setup & unload."""
with patch(
"homeassistant.components.life360.async_setup_entry", return_value=True
), patch("homeassistant.components.life360.async_unload_entry", return_value=True):
yield
@pytest.fixture
def life360_api():
"""Mock Life360 api."""
with patch(
"homeassistant.components.life360.config_flow.Life360", autospec=True
) as mock:
yield mock.return_value
def create_config_entry(hass, state=None):
"""Create mock config entry."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data=TEST_CONFIG_DATA,
version=1,
state=state,
options=DEFAULT_OPTIONS,
unique_id=TEST_USER.lower(),
)
config_entry.add_to_hass(hass)
return config_entry
# ========== User Flow Tests ===========================================================
async def test_user_show_form(hass: HomeAssistant, life360_api) -> None:
"""Test that the form is served with no input."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
await hass.async_block_till_done()
life360_api.get_authorization.assert_not_called()
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "user"
assert not result["errors"]
schema = result["data_schema"].schema
assert set(schema) == set(USER_INPUT)
# username and password fields should be empty.
keys = list(schema)
for key in USER_INPUT:
assert keys[keys.index(key)].default == vol.UNDEFINED
async def test_user_config_flow_success(hass: HomeAssistant, life360_api) -> None:
"""Test a successful user config flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
await hass.async_block_till_done()
life360_api.get_authorization.return_value = TEST_AUTHORIZATION
result = await hass.config_entries.flow.async_configure(
result["flow_id"], USER_INPUT
)
await hass.async_block_till_done()
life360_api.get_authorization.assert_called_once()
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["title"] == TEST_USER.lower()
assert result["data"] == TEST_CONFIG_DATA
assert result["options"] == DEFAULT_OPTIONS
@pytest.mark.parametrize(
("exception", "error"),
[(LoginError, "invalid_auth"), (Life360Error, "cannot_connect")],
)
async def test_user_config_flow_error(
hass: HomeAssistant, life360_api, caplog: pytest.LogCaptureFixture, exception, error
) -> None:
"""Test a user config flow with an error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
await hass.async_block_till_done()
life360_api.get_authorization.side_effect = exception("test reason")
result = await hass.config_entries.flow.async_configure(
result["flow_id"], USER_INPUT
)
await hass.async_block_till_done()
life360_api.get_authorization.assert_called_once()
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "user"
assert result["errors"]
assert result["errors"]["base"] == error
assert "test reason" in caplog.text
schema = result["data_schema"].schema
assert set(schema) == set(USER_INPUT)
# username and password fields should be prefilled with current values.
keys = list(schema)
for key, val in USER_INPUT.items():
default = keys[keys.index(key)].default
assert default != vol.UNDEFINED
assert default() == val
async def test_user_config_flow_already_configured(
hass: HomeAssistant, life360_api
) -> None:
"""Test a user config flow with an account already configured."""
create_config_entry(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], USER_INPUT
)
await hass.async_block_till_done()
life360_api.get_authorization.assert_not_called()
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "already_configured"
# ========== Reauth Flow Tests =========================================================
@pytest.mark.parametrize("state", [None, config_entries.ConfigEntryState.LOADED])
async def test_reauth_config_flow_success(
hass: HomeAssistant, life360_api, caplog: pytest.LogCaptureFixture, state
) -> None:
"""Test a successful reauthorization config flow."""
config_entry = create_config_entry(hass, state=state)
# Simulate current username & password are still valid, but authorization string has
# expired, such that getting a new authorization string from server is successful.
life360_api.get_authorization.return_value = TEST_AUTHORIZATION_2
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={
"source": config_entries.SOURCE_REAUTH,
"entry_id": config_entry.entry_id,
"title_placeholders": {"name": config_entry.title},
"unique_id": config_entry.unique_id,
},
data=config_entry.data,
)
await hass.async_block_till_done()
life360_api.get_authorization.assert_called_once()
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
assert "Reauthorization successful" in caplog.text
assert config_entry.data == TEST_CONFIG_DATA_2
async def test_reauth_config_flow_login_error(
hass: HomeAssistant, life360_api, caplog: pytest.LogCaptureFixture
) -> None:
"""Test a reauthorization config flow with a login error."""
config_entry = create_config_entry(hass)
# Simulate current username & password are invalid, which results in a form
# requesting new password, with old password as default value.
life360_api.get_authorization.side_effect = LoginError("test reason")
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={
"source": config_entries.SOURCE_REAUTH,
"entry_id": config_entry.entry_id,
"title_placeholders": {"name": config_entry.title},
"unique_id": config_entry.unique_id,
},
data=config_entry.data,
)
await hass.async_block_till_done()
life360_api.get_authorization.assert_called_once()
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert result["errors"]
assert result["errors"]["base"] == "invalid_auth"
assert "test reason" in caplog.text
schema = result["data_schema"].schema
assert len(schema) == 1
assert "password" in schema
key = list(schema)[0]
assert key.default() == TEST_PW
# Simulate hitting RECONFIGURE button.
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
assert result["errors"]
assert result["errors"]["base"] == "invalid_auth"
# Simulate getting a new, valid password.
life360_api.get_authorization.reset_mock(side_effect=True)
life360_api.get_authorization.return_value = TEST_AUTHORIZATION_3
result = await hass.config_entries.flow.async_configure(
result["flow_id"], {CONF_PASSWORD: TEST_PW_3}
)
await hass.async_block_till_done()
life360_api.get_authorization.assert_called_once()
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
assert "Reauthorization successful" in caplog.text
assert config_entry.data == TEST_CONFIG_DATA_3
# ========== Option flow Tests =========================================================
async def test_options_flow(hass: HomeAssistant) -> None:
"""Test an options flow."""
config_entry = create_config_entry(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "init"
assert not result["errors"]
schema = result["data_schema"].schema
assert set(schema) == set(USER_OPTIONS)
flow_id = result["flow_id"]
result = await hass.config_entries.options.async_configure(flow_id, USER_OPTIONS)
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["data"] == TEST_OPTIONS
assert config_entry.options == TEST_OPTIONS

50
tests/components/life360/test_init.py Normal file
View File

@ -0,0 +1,50 @@
"""Tests for the MyQ Connected Services integration."""
from homeassistant.components.life360 import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.helpers import issue_registry as ir
from tests.common import MockConfigEntry
async def test_life360_repair_issue(
hass: HomeAssistant, issue_registry: ir.IssueRegistry
) -> None:
"""Test the Life360 configuration entry loading/unloading handles the repair."""
config_entry_1 = MockConfigEntry(
title="Example 1",
domain=DOMAIN,
)
config_entry_1.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry_1.entry_id)
await hass.async_block_till_done()
assert config_entry_1.state is ConfigEntryState.LOADED
# Add a second one
config_entry_2 = MockConfigEntry(
title="Example 2",
domain=DOMAIN,
)
config_entry_2.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry_2.entry_id)
await hass.async_block_till_done()
assert config_entry_2.state is ConfigEntryState.LOADED
assert issue_registry.async_get_issue(DOMAIN, DOMAIN)
# Remove the first one
await hass.config_entries.async_remove(config_entry_1.entry_id)
await hass.async_block_till_done()
assert config_entry_1.state is ConfigEntryState.NOT_LOADED
assert config_entry_2.state is ConfigEntryState.LOADED
assert issue_registry.async_get_issue(DOMAIN, DOMAIN)
# Remove the second one
await hass.config_entries.async_remove(config_entry_2.entry_id)
await hass.async_block_till_done()
assert config_entry_1.state is ConfigEntryState.NOT_LOADED
assert config_entry_2.state is ConfigEntryState.NOT_LOADED
assert issue_registry.async_get_issue(DOMAIN, DOMAIN) is None