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