From: Scott Date: Mon, 6 Dec 2021 04:50:59 +0000 (-0800) Subject: Make presence detection work from cabin or house and deal with a X-Git-Url: https://wannabe.guru.org/gitweb/?a=commitdiff_plain;h=c63d439fb7e1d6f50e849b580f8bc6cfe88a81b6;p=python_utils.git Make presence detection work from cabin or house and deal with a bad VPN link between them better. --- diff --git a/locations.py b/locations.py new file mode 100644 index 0000000..744f63a --- /dev/null +++ b/locations.py @@ -0,0 +1,9 @@ +#!/usr/bin/env python3 + +import enum + +@enum.unique +class Location(enum.Enum): + UNKNOWN = 0 + HOUSE = 1 + CABIN = 2 diff --git a/people.py b/people.py new file mode 100644 index 0000000..1dc0421 --- /dev/null +++ b/people.py @@ -0,0 +1,14 @@ +#!/usr/bin/env python3 + +import enum + + +class Person(enum.Enum): + UNKNOWN = 0 + SCOTT = 1 + LYNN = 2 + ALEX = 3 + AARON_AND_DANA = 4 + AARON = 4 + DANA = 4 + diff --git a/presence.py b/presence.py index 5fad457..477b14f 100755 --- 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 - - -@enum.unique -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 scott@wennabe.house '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: diff --git a/site_config.py b/site_config.py index e3b186d..2d0c4c3 100644 --- a/site_config.py +++ b/site_config.py @@ -7,9 +7,10 @@ from typing import Callable # Note: this module is fairly early loaded. Be aware of dependencies. import config -import presence +from locations import Location logger = logging.getLogger(__name__) + args = config.add_commandline_args( f'({__file__})', 'Args related to __file__' @@ -26,28 +27,43 @@ args.add_argument( @dataclass class SiteConfig(object): - location: str + location_name: str + location: Location network: str network_netmask: str network_router_ip: str - presence_location: presence.Location + presence_location: Location is_anyone_present: Callable[None, bool] arper_minimum_device_count: int -def get_location(): +def get_location_name(): """ Where are we? - >>> location = get_location() + >>> location = get_location_name() >>> location == 'HOUSE' or location == 'CABIN' True + """ + return get_config().location_name + + +def get_location(): + """ + Returns location as an enum instead of a string. + + >>> from locations import Location + >>> location = get_location() + >>> location == Location.HOUSE or location == Location.CABIN + True + """ return get_config().location -def is_anyone_present_wrapper(location: presence.Location): +def is_anyone_present_wrapper(location: Location): + import presence p = presence.PresenceDetection() return p.is_anyone_in_location_now(location) @@ -58,7 +74,7 @@ def get_config(): site-specific including the current running location. >>> cfg = get_config() - >>> cfg.location == 'HOUSE' or cfg.location == 'CABIN' + >>> cfg.location_name == 'HOUSE' or cfg.location_name == 'CABIN' True """ @@ -74,22 +90,24 @@ def get_config(): location = 'CABIN' if location == 'HOUSE': return SiteConfig( - location = 'HOUSE', + location_name = 'HOUSE', + location = Location.HOUSE, network = '10.0.0.0/24', network_netmask = '255.255.255.0', network_router_ip = '10.0.0.1', - presence_location = presence.Location.HOUSE, - is_anyone_present = lambda x=presence.Location.HOUSE: is_anyone_present_wrapper(x), + presence_location = Location.HOUSE, + is_anyone_present = lambda x=Location.HOUSE: is_anyone_present_wrapper(x), arper_minimum_device_count = 50, ) elif location == 'CABIN': return SiteConfig( - location = 'CABIN', + location_name = 'CABIN', + location = Location.CABIN, network = '192.168.0.0/24', network_netmask = '255.255.255.0', network_router_ip = '192.168.0.1', - presence_location = presence.Location.CABIN, - is_anyone_present = lambda x=presence.Location.CABIN: is_anyone_present_wrapper(x), + presence_location = Location.CABIN, + is_anyone_present = lambda x=Location.CABIN: is_anyone_present_wrapper(x), arper_minimum_device_count = 15, ) else: