waitable_presence.py

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