X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=waitable_presence.py;h=9d79c6c5d61e1065f1444670c3ce24c1b9b7cd0f;hb=1803c7b2af8fde615b81ac21a6dc56e56dc20594;hp=9e0a9d0180663ad2337d7a6dc9ff42ccc677afb5;hpb=5f75cf834725ac26b289cc5f157af0cb71cd5f0e;p=python_utils.git diff --git a/waitable_presence.py b/waitable_presence.py index 9e0a9d0..9d79c6c 100644 --- a/waitable_presence.py +++ b/waitable_presence.py @@ -1,9 +1,10 @@ #!/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 @@ -13,9 +14,9 @@ from typing import Optional, Tuple from overrides import overrides import base_presence -from type.locations import Location import site_config import state_tracker +from type.locations import Location logger = logging.getLogger(__name__) @@ -24,7 +25,7 @@ class WaitablePresenceDetectorWithMemory(state_tracker.WaitableAutomaticStateTra """ 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: + particular location. Example suggested usage pattern:: detector = waitable_presence.WaitablePresenceDetectorWithMemory(60.0) while True: @@ -38,16 +39,20 @@ class WaitablePresenceDetectorWithMemory(state_tracker.WaitableAutomaticStateTra """ def __init__( - self, - override_update_interval_sec: float = 60.0, - override_location: Location = site_config.get_location(), + 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 - self.detector: base_presence.PresenceDetection = base_presence.PresenceDetection() + 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, @@ -70,7 +75,7 @@ class WaitablePresenceDetectorWithMemory(state_tracker.WaitableAutomaticStateTra 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...') + 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 @@ -87,15 +92,17 @@ class WaitablePresenceDetectorWithMemory(state_tracker.WaitableAutomaticStateTra logger.debug('PresenceDetector is incomplete; trying to reinitialize...') self.detector = base_presence.PresenceDetection() - def is_someone_home(self) -> Tuple[bool, datetime.datetime]: + 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: - raise Exception("Too Soon!") + 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)