#!/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',
)
-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:
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()
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:
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"
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]:
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():
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]:
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
projects / python_utils.git / blobdiff