X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=smart_home%2Fregistry.py;fp=smart_home%2Fregistry.py;h=f79ae90c15f5f7c881ae1acecaee434681aaf2aa;hb=cce8d58af187c0a7fb7585eab5bda9fed731b719;hp=16e18ba11bcc5fa19443245f02df546c68d54787;hpb=14b42faebd598dc14cec6eaef77f06845e500b4b;p=python_utils.git diff --git a/smart_home/registry.py b/smart_home/registry.py index 16e18ba..f79ae90 100644 --- a/smart_home/registry.py +++ b/smart_home/registry.py @@ -1,18 +1,17 @@ #!/usr/bin/env python3 +"""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.cameras as cameras -import smart_home.chromecasts as chromecasts -import smart_home.device as device -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__})", @@ -27,26 +26,29 @@ 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'], ) -> 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'] assert file_utils.does_file_exist(registry_file) - logger.debug(f'Reading {registry_file}') + logger.debug('Reading %s', registry_file) with open(registry_file, "r") as rf: contents = rf.readlines() @@ -57,12 +59,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?! Skipping it.' - logger.warning(msg) + logger.warning('SH-CONFIG> "%s" is malformed?! Skipping it.', line) continue mac = mac.strip() name = name.strip() @@ -72,7 +73,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: @@ -91,14 +92,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" @@ -107,7 +108,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]: @@ -131,18 +132,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(): @@ -184,11 +185,13 @@ class SmartHomeRegistry(object): except Exception as e: logger.exception(e) logger.debug( - f'Device {name} at {mac} with {kws} confused me, returning a generic Device' + '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) + logger.warning('%s is not a known smart home device, returning None', mac) return None def query(self, query: str) -> List[device.Device]: @@ -197,12 +200,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