-
Notifications
You must be signed in to change notification settings - Fork 299
/
composite.py
102 lines (86 loc) · 3.75 KB
/
composite.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
"""
A Device Tracker platform that combines one or more GPS based device trackers.
For more details about this platform, please refer to
https://github.com/pnbruckner/homeassistant-config#device_trackercompositepy
"""
import logging
import threading
import voluptuous as vol
from homeassistant.components.device_tracker import (
ATTR_BATTERY, PLATFORM_SCHEMA)
from homeassistant.const import (
ATTR_GPS_ACCURACY, ATTR_LATITUDE, ATTR_LONGITUDE, CONF_ENTITY_ID,
CONF_NAME, EVENT_HOMEASSISTANT_START)
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.event import track_state_change
from homeassistant.util import dt as dt_util
__version__ = '1.0.1'
_LOGGER = logging.getLogger(__name__)
ATTR_LAST_SEEN = 'last_seen'
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_NAME): cv.string,
vol.Required(CONF_ENTITY_ID): cv.entity_ids
})
def setup_scanner(hass, config, see, discovery_info=None):
def run_setup(event):
CompositeScanner(hass, config, see)
hass.bus.listen_once(EVENT_HOMEASSISTANT_START, run_setup)
return True
class CompositeScanner:
def __init__(self, hass, config, see):
self._hass = hass
self._see = see
entities = config[CONF_ENTITY_ID]
self._entities = dict.fromkeys(entities, False)
self._dev_id = config[CONF_NAME]
self._lock = threading.Lock()
self._prev_seen = None
self._remove = track_state_change(
hass, entities, self._update_info)
def _bad_entity(self, entity_id, message):
msg = "{} {}".format(entity_id, message)
# Has there already been a warning for this entity?
if self._entities[entity_id]:
_LOGGER.error(msg)
self._remove()
self._entities.pop(entity_id)
# Are there still any entities to watch?
if len(self._entities):
self._remove = track_state_change(
self._hass, self._entities.keys(), self._update_info)
else:
_LOGGER.warning(msg)
self._entities[entity_id] = True
def _update_info(self, entity_id, old_state, new_state):
with self._lock:
# Get time device was last seen, which is the entity's last_seen
# attribute, or if that doesn't exist, then last_updated from the
# new state object. Make sure last_seen is timezone aware in UTC.
# Note that dt_util.as_utc assumes naive datetime is in local
# timezone.
last_seen = dt_util.as_utc(
new_state.attributes.get(ATTR_LAST_SEEN,
new_state.last_updated))
# Is this newer info than last update?
if self._prev_seen and self._prev_seen >= last_seen:
_LOGGER.debug("Skipping: prv({}) >= new({})".format(
self._prev_seen, last_seen))
return
# GPS coordinates and accuracy are required.
# Battery level is optional.
try:
gps = (new_state.attributes[ATTR_LATITUDE],
new_state.attributes[ATTR_LONGITUDE])
except KeyError:
self._bad_entity(entity_id, "missing gps attributes")
return
try:
gps_accuracy = new_state.attributes[ATTR_GPS_ACCURACY]
except KeyError:
self._bad_entity(entity_id, "missing gps_accuracy attribute")
return
battery = new_state.attributes.get(ATTR_BATTERY)
attrs = {ATTR_LAST_SEEN: last_seen.replace(microsecond=0)}
self._see(dev_id=self._dev_id, gps=gps, gps_accuracy=gps_accuracy,
battery=battery, attributes=attrs)
self._prev_seen = last_seen