Make presence detection work from cabin or house and deal with a

[python_utils.git] / presence.py

diff --git a/presence.py b/presence.py
index 5fad457aa1251c88ab66b2b693dd25dc00fbbeab..477b14ffa0a83c2a0cf8c18a90ab6537953cfeda 100755 (executable)
--- a/presence.py
+++ b/presence.py
@@ -2,16 +2,18 @@
 
 import datetime
 from collections import defaultdict
-import enum
 import logging
 import re
-import sys
-from typing import Dict, List
+from typing import Dict, List, Set
 
 # Note: this module is fairly early loaded.  Be aware of dependencies.
 import argparse_utils
 import bootstrap
 import config
+from locations import Location
+from people import Person
+import site_config
+
 
 logger = logging.getLogger(__name__)
 
@@ -28,23 +30,6 @@ cfg.add_argument(
 )
 
 
-class Person(enum.Enum):
-    UNKNOWN = 0
-    SCOTT = 1
-    LYNN = 2
-    ALEX = 3
-    AARON_AND_DANA = 4
-    AARON = 4
-    DANA = 4
-
-
[email protected]
-class Location(enum.Enum):
-    UNKNOWN = 0
-    HOUSE = 1
-    CABIN = 2
-
-
 class PresenceDetection(object):
     def __init__(self) -> None:
         # Note: list most important devices first.
@@ -74,14 +59,26 @@ class PresenceDetection(object):
                 "96:69:2C:88:7A:C3",
             ],
         }
+        self.run_location = site_config.get_location()
+        logger.debug(f"run_location is {self.run_location}")
         self.weird_mac_at_cabin = False
         self.location_ts_by_mac: Dict[
             Location, Dict[str, datetime.datetime]
         ] = defaultdict(dict)
         self.names_by_mac: Dict[str, str] = {}
+        self.dark_locations: Set[Location] = set()
         self.update()
 
     def update(self) -> None:
+        self.dark_locations = set()
+        if self.run_location is Location.HOUSE:
+            self.update_from_house()
+        elif self.run_location is Location.CABIN:
+            self.update_from_cabin()
+        else:
+            raise Exception("Where the hell is this running?!")
+
+    def update_from_house(self) -> None:
         from exec_utils import cmd_with_timeout
         try:
             persisted_macs = config.config['presence_macs_file']
@@ -96,10 +93,26 @@ class PresenceDetection(object):
             self.parse_raw_macs_file(raw, Location.CABIN)
         except Exception as e:
             logger.exception(e)
-            logger.error(
-                'Unable to fetch MAC Addresses from meerkat; can\'t do proper presence detection.'
+            logger.warning("Can't see the cabin right now; presence detection impared.")
+            self.dark_locations.add(Location.CABIN)
+
+    def update_from_cabin(self) -> None:
+        from exec_utils import cmd_with_timeout
+        try:
+            persisted_macs = config.config['presence_macs_file']
+        except KeyError:
+            persisted_macs = '/home/scott/cron/persisted_mac_addresses.txt'
+        self.read_persisted_macs_file(persisted_macs, Location.CABIN)
+        try:
+            raw = cmd_with_timeout(
+                "ssh [email protected] 'cat /home/scott/cron/persisted_mac_addresses.txt'",
+                timeout_seconds=10.0,
             )
-            sys.exit(1)
+            self.parse_raw_macs_file(raw, Location.HOUSE)
+        except Exception as e:
+            logger.exception(e)
+            logger.warning(f"Can't see the house right now; presence detection impared.")
+            self.dark_locations.add(Location.HOUSE)
 
     def read_persisted_macs_file(
         self, filename: str, location: Location
@@ -145,6 +158,8 @@ class PresenceDetection(object):
             self.weird_mac_at_cabin = True
 
     def is_anyone_in_location_now(self, location: Location) -> bool:
+        if location in self.dark_locations:
+            raise Exception("Can't see {location} right now; answer undefined.")
         for person in Person:
             if person is not None:
                 loc = self.where_is_person_now(person)
@@ -156,6 +171,10 @@ class PresenceDetection(object):
 
     def where_is_person_now(self, name: Person) -> Location:
         import dict_utils
+        if len(self.dark_locations) > 0:
+            logger.warning(
+                f"Can't see {self.dark_locations} right now; answer confidence impacted"
+            )
         logger.debug(f'Looking for {name}...')
 
         if name is Person.UNKNOWN: