waitable_presence.py

   1 #!/usr/bin/env python3
   2
   3 """A PresenceDetector that is waitable.  This is not part of
   4 base_presence.py because I do not want to bring these dependencies
   5 into that lower-level module (especially state_tracker).
   6
   7 """
   8
   9 import datetime
  10 import logging
  11 from typing import Optional, Tuple
  12
  13 from overrides import overrides
  14
  15 import base_presence
  16 from type.locations import Location
  17 import site_config
  18 import state_tracker
  19
  20 logger = logging.getLogger(__name__)
  21
  22
  23 class WaitablePresenceDetectorWithMemory(state_tracker.WaitableAutomaticStateTracker):
  24     """
  25     This is a waitable class that keeps a PresenceDetector internally
  26     and periodically polls it to detect changes in presence in a
  27     particular location.  Example suggested usage pattern:
  28
  29         detector = waitable_presence.WaitablePresenceDetectorWithMemory(60.0)
  30         while True:
  31             changed = detector.wait(timeout=60 * 5)  # or, None for "forever"
  32             (someone_is_home, since) = detector.is_someone_home()
  33             if changed:
  34                 detector.reset()
  35             logger.debug(
  36                 f'someone_is_home={someone_is_home}, since={since}, changed={changed}'
  37             )
  38     """
  39
  40     def __init__(
  41         self,
  42         override_update_interval_sec: float = 60.0,
  43         override_location: Location = site_config.get_location(),
  44     ) -> None:
  45         self.last_someone_is_home: Optional[bool] = None
  46         self.someone_is_home: Optional[bool] = None
  47         self.everyone_gone_since: Optional[datetime.datetime] = None
  48         self.someone_home_since: Optional[datetime.datetime] = None
  49         self.location = override_location
  50         self.detector: base_presence.PresenceDetection = (
  51             base_presence.PresenceDetection()
  52         )
  53         super().__init__(
  54             {
  55                 'poll_presence': override_update_interval_sec,
  56                 'check_detector': override_update_interval_sec * 5,
  57             }
  58         )
  59
  60     @overrides
  61     def update(
  62         self,
  63         update_id: str,
  64         now: datetime.datetime,
  65         last_invocation: Optional[datetime.datetime],
  66     ) -> None:
  67         if update_id == 'poll_presence':
  68             self.poll_presence(now)
  69         elif update_id == 'check_detector':
  70             self.check_detector()
  71         else:
  72             raise Exception(f'Unknown update type {update_id} in {__file__}')
  73
  74     def poll_presence(self, now: datetime.datetime) -> None:
  75         logger.debug(f'Checking presence in {self.location} now...')
  76         self.detector.update()
  77         if self.detector.is_anyone_in_location_now(self.location):
  78             self.someone_is_home = True
  79             self.someone_home_since = now
  80         else:
  81             self.someone_is_home = False
  82             self.everyone_gone_since = now
  83         if self.someone_is_home != self.last_someone_is_home:
  84             self.something_changed()
  85             self.last_someone_is_home = self.someone_is_home
  86
  87     def check_detector(self) -> None:
  88         if len(self.detector.dark_locations) > 0:
  89             logger.debug('PresenceDetector is incomplete; trying to reinitialize...')
  90             self.detector = base_presence.PresenceDetection()
  91
  92     def is_someone_home(self) -> Tuple[bool, datetime.datetime]:
  93         """Returns a tuple of a bool that indicates whether someone is home
  94         and a datetime that indicates how long either someone has been
  95         home or no one has been home.
  96
  97         """
  98         if self.someone_is_home is None:
  99             raise Exception("Too Soon!")
 100         if self.someone_is_home:
 101             return (True, self.someone_home_since)
 102         else:
 103             return (False, self.everyone_gone_since)