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
|
#!/usr/bin/env python3
"""A PresenceDetector that is waitable. This is not part of
base_presence.py because I do not want to bring these dependencies
into that lower-level module (especially state_tracker).
"""
import datetime
import logging
from typing import Optional, Tuple
from overrides import overrides
import base_presence
from type.locations import Location
import site_config
import state_tracker
logger = logging.getLogger(__name__)
class WaitablePresenceDetectorWithMemory(state_tracker.WaitableAutomaticStateTracker):
"""
This is a waitable class that keeps a PresenceDetector internally
and periodically polls it to detect changes in presence in a
particular location. Example suggested usage pattern:
detector = waitable_presence.WaitablePresenceDetectorWithMemory(60.0)
while True:
changed = detector.wait(timeout=60 * 5) # or, None for "forever"
(someone_is_home, since) = detector.is_someone_home()
if changed:
detector.reset()
logger.debug(
f'someone_is_home={someone_is_home}, since={since}, changed={changed}'
)
"""
def __init__(
self,
override_update_interval_sec: float = 60.0,
override_location: Location = site_config.get_location(),
) -> None:
self.last_someone_is_home: Optional[bool] = None
self.someone_is_home: Optional[bool] = None
self.everyone_gone_since: Optional[datetime.datetime] = None
self.someone_home_since: Optional[datetime.datetime] = None
self.location = override_location
self.detector: base_presence.PresenceDetection = base_presence.PresenceDetection()
super().__init__(
{
'poll_presence': override_update_interval_sec,
'check_detector': override_update_interval_sec * 5,
}
)
@overrides
def update(
self,
update_id: str,
now: datetime.datetime,
last_invocation: Optional[datetime.datetime],
) -> None:
if update_id == 'poll_presence':
self.poll_presence(now)
elif update_id == 'check_detector':
self.check_detector()
else:
raise Exception(f'Unknown update type {update_id} in {__file__}')
def poll_presence(self, now: datetime.datetime) -> None:
logger.debug(f'Checking presence in {self.location} now...')
self.detector.update()
if self.detector.is_anyone_in_location_now(self.location):
self.someone_is_home = True
self.someone_home_since = now
else:
self.someone_is_home = False
self.everyone_gone_since = now
if self.someone_is_home != self.last_someone_is_home:
self.something_changed()
self.last_someone_is_home = self.someone_is_home
def check_detector(self) -> None:
if len(self.detector.dark_locations) > 0:
logger.debug('PresenceDetector is incomplete; trying to reinitialize...')
self.detector = base_presence.PresenceDetection()
def is_someone_home(self) -> Tuple[bool, datetime.datetime]:
"""Returns a tuple of a bool that indicates whether someone is home
and a datetime that indicates how long either someone has been
home or no one has been home.
"""
if self.someone_is_home is None:
raise Exception("Too Soon!")
if self.someone_is_home:
return (True, self.someone_home_since)
else:
return (False, self.everyone_gone_since)
|
