Hi @shaddow, you didn’t tag me so I didn’t know you mentioned me until I happened to go back and search this topic. Which, BTW, I did because I have an update for the custom Life360 device tracker platform I made. Maybe I should be posting this elsewhere, but oh well, here goes…
The full code is below, but since I posted the last update I’ve done quite a bit of work on it. I still use it everyday and I’m very happy with it. (And it still, unfortunately, depends on a low-level python integration for Life360 which isn’t an official package.)
Besides additional error handling, which I’ve added as I’ve monitored what can go wrong with the communications to/from Life360, I also recently added a feature that can fire an event if the time between updates for members goes beyond a configurable threshold. I use this in an automation to send myself an e-mail if this happens, which it does from time to time when the Life360 app on someone’s phone stops updating the Life360 server. The automation looks like this:
- alias: Life360 Overdue Update
trigger:
platform: event
event_type: device_tracker.life360_update_overdue
action:
service: notify.email_phil
data_template:
title: Life360 update overdue
message: >
Update for {{ state_attr(trigger.event.data.entity_id, 'friendly_name') }} is overdue.
And here’s what the platform configuration looks like:
device_tracker:
- platform: life360
username: life360_user_name
password: !secret life360_password
interval_seconds: 10
max_update_wait:
minutes: 30
And following is the code. I’m open to any comments, suggestions, criticisms, etc.
import datetime as dt
from requests import HTTPError, ConnectionError, Timeout
from json.decoder import JSONDecodeError
import logging
import voluptuous as vol
from homeassistant.components.device_tracker import (
PLATFORM_SCHEMA, CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL)
from homeassistant.const import CONF_USERNAME, CONF_PASSWORD
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.event import track_time_interval
from homeassistant import util
_LOGGER = logging.getLogger(__name__)
CONF_MAX_UPDATE_WAIT = 'max_update_wait'
_AUTHORIZATION_TOKEN = 'cFJFcXVnYWJSZXRyZTRFc3RldGhlcnVmcmVQdW1hbUV4dWNyRU'\
'h1YzptM2ZydXBSZXRSZXN3ZXJFQ2hBUHJFOTZxYWtFZHI0Vg=='
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_USERNAME): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
vol.Optional(CONF_MAX_UPDATE_WAIT): vol.All(
cv.time_period, cv.positive_timedelta)
})
def setup_scanner(hass, config, see, discovery_info=None):
from pylife360 import life360
api = life360(_AUTHORIZATION_TOKEN,
config[CONF_USERNAME], config[CONF_PASSWORD])
if not api.authenticate(timeout=(3.05, 5)):
_LOGGER.error('Authentication failed!')
return False
_LOGGER.info('Authentication successful!')
interval = config.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL)
max_update_wait = config.get(CONF_MAX_UPDATE_WAIT)
Life360Scanner(hass, see, interval, max_update_wait, api)
return True
class Life360Scanner(object):
def __init__(self, hass, see, interval, max_update_wait, api):
self._hass = hass
self._see = see
self._max_update_wait = max_update_wait
self._api = api
self._dev_data = {}
self._started = util.dt.utcnow()
track_time_interval(self._hass, self._update_life360, interval)
def _update_life360(self, now=None):
excs = (HTTPError, ConnectionError, Timeout, JSONDecodeError)
def exc_msg(exc, msg=None, extra=None):
_msg = '{}: {}'.format(exc.__class__.__name__, str(exc))
if msg:
_msg = '{}: '.format(msg) + _msg
if extra:
_msg += '; {}'.format(extra)
_LOGGER.error(_msg)
_LOGGER.debug('Checking members.')
try:
circles = self._api.get_circles(timeout=(3.05, 5))
except excs as exc:
exc_msg(exc, 'get_circles')
return
for circle in circles:
try:
members = self._api.get_circle(
circle['id'], timeout=(3.05, 5))['members']
except excs as exc:
exc_msg(exc, 'get_circle')
continue
for m in members:
try:
dev_id = util.slugify(
'_'.join([m['firstName'], m['lastName']])
.replace('-', '_'))
prev_update, reported = self._dev_data.get(
dev_id, (None, False))
loc = m.get('location')
last_update = (None if not loc else
util.dt.utc_from_timestamp(
float(loc['timestamp'])))
if self._max_update_wait:
update = last_update or prev_update or self._started
overdue = (util.dt.utcnow() - update
> self._max_update_wait)
if overdue and not reported:
self._hass.bus.fire(
'device_tracker.life360_update_overdue',
{'entity_id': dev_id})
reported = True
elif not overdue:
reported = False
if not loc:
err_msg = m['issues']['title']
if err_msg:
if m['issues']['dialog']:
err_msg += ': ' + m['issues']['dialog']
else:
err_msg = 'Location information missing'
_LOGGER.error('{}: {}'.format(dev_id, err_msg))
elif prev_update is None or last_update > prev_update:
msg = 'Updating {}.'.format(dev_id)
if prev_update is not None:
msg += ' Time since last update: {}.'.format(
last_update - prev_update)
_LOGGER.debug(msg)
lat = float(loc['latitude'])
lon = float(loc['longitude'])
loc_name = loc['name']
# Make sure Home is always seen as exactly as home,
# which is the special device_tracker state for home.
if loc_name is not None and loc_name.lower() == 'home':
loc_name = 'home'
attrs = {
'last_update' : str(util.dt.as_local(last_update)),
'circle': circle['name']
}
self._see(dev_id=dev_id, location_name=loc_name,
gps=(lat, lon),
gps_accuracy=round(float(loc['accuracy'])),
battery=round(float(loc['battery'])),
attributes=attrs)
self._dev_data[dev_id] = (
last_update or prev_update, reported)
except Exception as exc:
exc_msg(exc, extra='m = {}'.format(m))