From: Scott Gasch Date: Fri, 29 Oct 2021 17:33:07 +0000 (-0700) Subject: Adds Arper, cleans up other stuff. X-Git-Url: https://wannabe.guru.org/gitweb/?a=commitdiff_plain;h=29ee1f98654a689e9cab76b0c7c68428faa43a8c;p=python_utils.git Adds Arper, cleans up other stuff. --- diff --git a/arper.py b/arper.py new file mode 100644 index 0000000..73d1dd6 --- /dev/null +++ b/arper.py @@ -0,0 +1,88 @@ +#!/usr/bin/env python3 + +import datetime +import logging +import os +from typing import Any, Optional + +import argparse_utils +from collect.bidict import BiDict +import config +import exec_utils +import persistent +import string_utils + +logger = logging.getLogger(__name__) + +cfg = config.add_commandline_args( + f'MAC <--> IP Address mapping table cache ({__file__})', + 'Commandline args related to MAC <--> IP Address mapping', +) +cfg.add_argument( + '--arper_cache_location', + default=f'{os.environ["HOME"]}/cache/.arp_table_cache', + metavar='FILENAME', + help='Where to cache the kernel ARP table', +) +cfg.add_argument( + '--arp_cache_max_staleness', + type=argparse_utils.valid_duration, + default=datetime.timedelta(seconds=60 * 5), + metavar='DURATION', + help='Max acceptable age of the kernel arp table cache' +) + + +@persistent.persistent_autoloaded_singleton() +class Arper(persistent.Persistent): + def __init__(self, cached_state: Optional[BiDict[str, str]] = None) -> None: + self.state = BiDict() + if cached_state is not None: + logger.debug('Loading Arper map from cached state.') + self.state = cached_state + else: + logger.debug('No usable cached state; calling /usr/sbin/arp') + output = exec_utils.cmd( + '/usr/sbin/arp -a', + timeout_seconds=5.0 + ) + for line in output.split('\n'): + line = str(line, 'ascii') + ip = string_utils.extract_ip_v4(line) + mac = string_utils.extract_mac_address(line) + mac = mac.lower() + logger.debug(f' {mac} => {ip}') + self.state[mac] = ip + + def get_ip_by_mac(self, mac: str) -> Optional[str]: + mac = mac.lower() + return self.state.get(mac, None) + + def get_mac_by_ip(self, ip: str) -> Optional[str]: + return self.state.inverse.get(ip, None) + + def save(self) -> bool: + logger.debug(f'Persisting state to {config.config["arp_cache_location"]}') + with open(config.config['arp_cache_location'], 'w') as wf: + for (mac, ip) in self.state.items(): + mac = mac.lower() + print(f'{mac}, {ip}', file=wf) + + @classmethod + def load(cls) -> Any: + if persistent.was_file_written_within_n_seconds( + config.config['arp_cache_location'], + config.config['arp_cache_max_staleness'].total_seconds(), + ): + logger.debug(f'Loading state from {config.config["arp_cache_location"]}') + cached_state = BiDict() + with open(config.config['arp_cache_location'], 'r') as rf: + for line in rd.readline(): + (mac, ip) = line.split(',') + mac = mac.strip() + mac = mac.lower() + ip = ip.strip() + cached_state[mac] = ip + return Arper(cached_state) + logger.debug('No usable saved state found') + return None diff --git a/collect/bidict.py b/collect/bidict.py index 1fa66dc..8153e54 100644 --- a/collect/bidict.py +++ b/collect/bidict.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 -class bidict(dict): +class BiDict(dict): def __init__(self, *args, **kwargs): """ A class that stores both a Mapping between keys and values and @@ -9,7 +9,7 @@ class bidict(dict): is possible to have several keys with the same value, using the inverse map returns a sequence of keys. - >>> d = bidict() + >>> d = BiDict() >>> d['a'] = 1 >>> d['b'] = 2 >>> d['c'] = 2 diff --git a/smart_home/cameras.py b/smart_home/cameras.py index 963f54e..40850a9 100644 --- a/smart_home/cameras.py +++ b/smart_home/cameras.py @@ -2,25 +2,13 @@ """Utilities for dealing with the webcams.""" -from abc import abstractmethod -import datetime -import json import logging -import os -import re -import subprocess -import sys -from typing import Any, Dict, List, Optional, Set - -import argparse_utils -import config -import logging_utils + import smart_home.device as dev -from google_assistant import ask_google, GoogleResponse -from decorator_utils import timeout, memoized logger = logging.getLogger(__name__) + class BaseCamera(dev.Device): camera_mapping = { 'cabin_drivewaycam': 'cabin_driveway', diff --git a/smart_home/device.py b/smart_home/device.py index 27860c5..0953b8d 100644 --- a/smart_home/device.py +++ b/smart_home/device.py @@ -1,7 +1,8 @@ #!/usr/bin/env python3 import re -from typing import Any, List, Optional, Tuple +from typing import List, Optional + class Device(object): def __init__( @@ -24,6 +25,9 @@ class Device(object): def get_mac(self) -> str: return self.mac + def get_ip(self) -> str: + pass + def get_keywords(self) -> Optional[List[str]]: return self.kws diff --git a/smart_home/lights.py b/smart_home/lights.py index 0752139..5446722 100644 --- a/smart_home/lights.py +++ b/smart_home/lights.py @@ -2,7 +2,7 @@ """Utilities for dealing with the smart lights.""" -from abc import ABC, abstractmethod +from abc import abstractmethod import datetime import json import logging @@ -10,7 +10,7 @@ import os import re import subprocess import sys -from typing import Any, Dict, List, Optional, Set +from typing import Any, Dict, List, Optional import tinytuya as tt diff --git a/smart_home/outlets.py b/smart_home/outlets.py index 81e10a7..f34d574 100644 --- a/smart_home/outlets.py +++ b/smart_home/outlets.py @@ -10,7 +10,7 @@ import os import re import subprocess import sys -from typing import Any, Dict, List, Optional, Set +from typing import Dict, List, Optional import argparse_utils import config