#!/usr/bin/env python3 # © Copyright 2021-2022, Scott Gasch """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 import site_config import state_tracker from type.locations import Location 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(), injected_presence_detector: Optional[base_presence.PresenceDetection] = None, ) -> 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 if injected_presence_detector is not None: self.detector: base_presence.PresenceDetection = injected_presence_detector else: self.detector = 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('Checking presence in %s now...', self.location) 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) -> Optional[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: return None # checked too soon, wait a bit. if self.someone_is_home: assert self.someone_home_since is not None return (True, self.someone_home_since) else: assert self.everyone_gone_since is not None return (False, self.everyone_gone_since)