X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;ds=sidebyside;f=smart_home%2Fregistry.py;h=8ca7f3b96dad4cc25807b59d948803e95e8c72ca;hb=HEAD;hp=23584e119173e00f8d86dd38858126a990222f39;hpb=5f75cf834725ac26b289cc5f157af0cb71cd5f0e;p=python_utils.git diff --git a/smart_home/registry.py b/smart_home/registry.py index 23584e1..8ca7f3b 100644 --- a/smart_home/registry.py +++ b/smart_home/registry.py @@ -1,22 +1,23 @@ #!/usr/bin/env python3 +# © Copyright 2021-2022, Scott Gasch + +"""A searchable registry of known smart home devices and a factory for +constructing our wrappers around them.""" + import logging import re -from typing import List, Optional, Set +from typing import Dict, List, Optional, Set import argparse_utils import config import file_utils import logical_search -import smart_home.device as device -import smart_home.cameras as cameras -import smart_home.chromecasts as chromecasts -import smart_home.lights as lights -import smart_home.outlets as outlets +from smart_home import cameras, chromecasts, device, lights, outlets args = config.add_commandline_args( f"Smart Home Registry ({__file__})", - "Args related to the smart home configuration registry." + "Args related to the smart home configuration registry.", ) args.add_argument( '--smart_home_registry_file_location', @@ -27,30 +28,31 @@ args.add_argument( ) -logger = logging.getLogger(__file__) +logger = logging.getLogger(__name__) class SmartHomeRegistry(object): + """A searchable registry of known smart home devices and a factory for + constructing our wrappers around them.""" + def __init__( - self, - registry_file: Optional[str] = None, - filters: List[str] = ['smart'], + self, + registry_file: Optional[str] = None, + filters: List[str] = ['smart'], ) -> None: - self._macs_by_name = {} - self._keywords_by_name = {} - self._keywords_by_mac = {} - self._names_by_mac = {} - self._corpus = logical_search.Corpus() + self._macs_by_name: Dict[str, str] = {} + self._keywords_by_name: Dict[str, str] = {} + self._keywords_by_mac: Dict[str, str] = {} + self._names_by_mac: Dict[str, str] = {} + self._corpus: logical_search.Corpus = logical_search.Corpus() # Read the disk config file... if registry_file is None: - registry_file = config.config[ - 'smart_home_registry_file_location' - ] + registry_file = config.config['smart_home_registry_file_location'] assert file_utils.does_file_exist(registry_file) - logger.debug(f'Reading {registry_file}') - with open(registry_file, "r") as f: - contents = f.readlines() + logger.debug('Reading %s', registry_file) + with open(registry_file, "r") as rf: + contents = rf.readlines() # Parse the contents... for line in contents: @@ -59,11 +61,11 @@ class SmartHomeRegistry(object): line = line.strip() if line == "": continue - logger.debug(f'SH-CONFIG> {line}') + logger.debug('SH-CONFIG> %s', line) try: (mac, name, keywords) = line.split(",") except ValueError: - logger.warning(f'SH-CONFIG> {line} is malformed?!') + logger.warning('SH-CONFIG> "%s" is malformed?! Skipping it.', line) continue mac = mac.strip() name = name.strip() @@ -73,7 +75,7 @@ class SmartHomeRegistry(object): if filters is not None: for f in filters: if f not in keywords: - logger.debug(f'Skipping this entry b/c of filter {f}') + logger.debug('Skipping this entry b/c of filter: %s', f) skip = True break if not skip: @@ -81,9 +83,9 @@ class SmartHomeRegistry(object): self._keywords_by_name[name] = keywords self._keywords_by_mac[mac] = keywords self._names_by_mac[mac] = name - self.index_device(name, keywords, mac) + self._index_device(name, keywords, mac) - def index_device(self, name: str, keywords: str, mac: str) -> None: + def _index_device(self, name: str, keywords: str, mac: str) -> None: properties = [("name", name)] tags = set() for kw in keywords.split(): @@ -92,14 +94,14 @@ class SmartHomeRegistry(object): properties.append((key, value)) else: tags.add(kw) - device = logical_search.Document( + dev = logical_search.Document( docid=mac, tags=tags, properties=properties, reference=None, ) - logger.debug(f'Indexing document {device}') - self._corpus.add_doc(device) + logger.debug('Indexing document: %s', dev) + self._corpus.add_doc(dev) def __repr__(self) -> str: s = "Known devices:\n" @@ -108,10 +110,31 @@ class SmartHomeRegistry(object): s += f" {name} ({mac}) => {keywords}\n" return s - def get_keywords_by_name(self, name: str) -> Optional[device.Device]: + def get_keywords_by_name(self, name: str) -> Optional[str]: + """Given the name of a device, get its keywords. + + >>> reg = SmartHomeRegistry('/home/scott/bin/network_mac_addresses.txt') + >>> reg.get_keywords_by_name('near_kitchen_lamp') + 'wifi smart light goog meross test' + + >>> reg.get_keywords_by_name('unknown') is None + True + + """ return self._keywords_by_name.get(name, None) def get_macs_by_name(self, name: str) -> Set[str]: + """Given the name of a device, get its MAC address(es) + + >>> reg = SmartHomeRegistry('/home/scott/bin/network_mac_addresses.txt') + >>> reg.get_macs_by_name('near_kitchen_lamp') + {'34:29:8F:12:34:8E'} + + >>> reg.get_macs_by_name('unknown') + set() + + """ + retval = set() for (mac, lname) in self._names_by_mac.items(): if name in lname: @@ -119,6 +142,19 @@ class SmartHomeRegistry(object): return retval def get_macs_by_keyword(self, keyword: str) -> Set[str]: + """Given a keyword, return the set of MAC address(es) that have + that keyword. + + >>> reg = SmartHomeRegistry('/home/scott/bin/network_mac_addresses.txt') + >>> r = reg.get_macs_by_keyword('test') + >>> e = set(['34:29:8F:12:26:74' , '34:29:8F:12:34:8E']) + >>> r == e + True + + >>> reg.get_macs_by_keyword('unknown') + set() + + """ retval = set() for (mac, keywords) in self._keywords_by_mac.items(): if keyword in keywords: @@ -126,24 +162,30 @@ class SmartHomeRegistry(object): return retval def get_device_by_name(self, name: str) -> Optional[device.Device]: + """Given a name, return its Device object.""" + if name in self._macs_by_name: return self.get_device_by_mac(self._macs_by_name[name]) return None def get_all_devices(self) -> List[device.Device]: + """Return a list of all known devices.""" + retval = [] - for (mac, kws) in self._keywords_by_mac.items(): + for mac, _ in self._keywords_by_mac.items(): if mac is not None: - device = self.get_device_by_mac(mac) - if device is not None: - retval.append(device) + dev = self.get_device_by_mac(mac) + if dev is not None: + retval.append(dev) return retval def get_device_by_mac(self, mac: str) -> Optional[device.Device]: + """Given a MAC address, return its Device object.""" + if mac in self._keywords_by_mac: name = self._names_by_mac[mac] kws = self._keywords_by_mac[mac] - logger.debug(f'Found {name} -> {mac} ({kws})') + logger.debug('Found %s -> %s (%s)', name, mac, kws) try: if 'light' in kws.lower(): if 'tplink' in kws.lower(): @@ -183,25 +225,35 @@ class SmartHomeRegistry(object): logger.debug(' ...an unknown device (should this be here?)') return device.Device(name, mac, kws) except Exception as e: - logger.warning( - f'Got exception {e} while trying to communicate with device {name}/{mac}.' + logger.exception(e) + logger.debug( + 'Device %s at %s with %s confused me; returning a generic Device', + name, + mac, + kws, ) return device.Device(name, mac, kws) - logger.warning(f'{mac} is not a known smart home device, returning None') + logger.warning('%s is not a known smart home device, returning None', mac) return None def query(self, query: str) -> List[device.Device]: - """Evaluates a lighting query expression formed of keywords to search + """Evaluates a device query expression formed of keywords to search for, logical operators (and, or, not), and parenthesis. Returns a list of matching lights. """ retval = [] - logger.debug(f'Executing query {query}') + logger.debug('Executing query: %s', query) results = self._corpus.query(query) if results is not None: for mac in results: if mac is not None: - device = self.get_device_by_mac(mac) - if device is not None: - retval.append(device) + dev = self.get_device_by_mac(mac) + if dev is not None: + retval.append(dev) return retval + + +if __name__ == '__main__': + import doctest + + doctest.testmod()