X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=smart_home%2Fregistry.py;h=d63474da56f98bc44d78a221f47c3f4fd7d04971;hb=532df2c5b57c7517dfb3dddd8c1358fbadf8baf3;hp=20fb3f43807bf0dbbd45fa7f5c36ed8f4d1bc917;hpb=b454ad295eb3024a238d32bf2aef1ebc3c496b44;p=python_utils.git diff --git a/smart_home/registry.py b/smart_home/registry.py index 20fb3f4..d63474d 100644 --- a/smart_home/registry.py +++ b/smart_home/registry.py @@ -1,23 +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 -import warnings +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', @@ -28,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: @@ -60,13 +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: - msg = f'SH-CONFIG> {line} is malformed?!' - logger.warning(msg) - warnings.warn(msg) + logger.warning('SH-CONFIG> "%s" is malformed?! Skipping it.', line) continue mac = mac.strip() name = name.strip() @@ -76,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: @@ -95,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" @@ -111,7 +110,7 @@ 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]: return self._keywords_by_name.get(name, None) def get_macs_by_name(self, name: str) -> Set[str]: @@ -135,18 +134,18 @@ class SmartHomeRegistry(object): def get_all_devices(self) -> List[device.Device]: 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]: 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(): @@ -187,10 +186,14 @@ class SmartHomeRegistry(object): return device.Device(name, mac, kws) except Exception as e: 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) - msg = f'{mac} is not a known smart home device, returning None' - logger.warning(msg) - warnings.warn(msg) + logger.warning('%s is not a known smart home device, returning None', mac) return None def query(self, query: str) -> List[device.Device]: @@ -199,12 +202,12 @@ class SmartHomeRegistry(object): 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