diff options
Diffstat (limited to 'smart_home/config.py')
| -rw-r--r-- | smart_home/config.py | 54 |
1 files changed, 40 insertions, 14 deletions
diff --git a/smart_home/config.py b/smart_home/config.py index 723e097..a28caa7 100644 --- a/smart_home/config.py +++ b/smart_home/config.py @@ -1,5 +1,6 @@ #!/usr/bin/env python3 +import logging import re from typing import List, Optional, Set @@ -8,6 +9,7 @@ import config import file_utils import logical_search import smart_home.device as device +import smart_home.cameras as cameras import smart_home.lights as lights import smart_home.outlets as outlets @@ -24,47 +26,57 @@ parser.add_argument( ) +logger = logging.getLogger(__file__) + + class SmartHomeConfig(object): def __init__( self, config_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() + + # Read the disk config file... if config_file is None: config_file = config.config[ 'smart_home_config_file_location' ] assert file_utils.does_file_exist(config_file) + logger.debug(f'Reading {config_file}') with open(config_file, "r") as f: contents = f.readlines() - self._macs_by_name = {} - self._keywords_by_name = {} - self._keywords_by_mac = {} - self._names_by_mac = {} - self._corpus = logical_search.Corpus() - + # Parse the contents... for line in contents: line = line.rstrip("\n") line = re.sub(r"#.*$", r"", line) line = line.strip() if line == "": continue + logger.debug(f'> {line}') (mac, name, keywords) = line.split(",") mac = mac.strip() name = name.strip() keywords = keywords.strip() + skip = False if filters is not None: for f in filters: - if not f in keywords: - continue - - self._macs_by_name[name] = mac - self._keywords_by_name[name] = keywords - self._keywords_by_mac[mac] = keywords - self._names_by_mac[mac] = name - self.index_device(name, keywords, mac) + if f not in keywords: + logger.debug(f'Skipping this entry b/c of filter {f}') + skip = True + break + if not skip: + self._macs_by_name[name] = mac + self._keywords_by_name[name] = keywords + self._keywords_by_mac[mac] = keywords + self._names_by_mac[mac] = name + self.index_device(name, keywords, mac) def index_device(self, name: str, keywords: str, mac: str) -> None: properties = [("name", name)] @@ -81,6 +93,7 @@ class SmartHomeConfig(object): properties=properties, reference=None, ) + logger.debug(f'Indexing document {device}') self._corpus.add_doc(device) def __repr__(self) -> str: @@ -125,27 +138,39 @@ class SmartHomeConfig(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})') if 'light' in kws.lower(): if 'tplink' in kws.lower(): + logger.debug(' ...a TPLinkLight') return lights.TPLinkLight(name, mac, kws) elif 'tuya' in kws.lower(): + logger.debug(' ...a TuyaLight') return lights.TuyaLight(name, mac, kws) elif 'goog' in kws.lower(): + logger.debug(' ...a GoogleLight') return lights.GoogleLight(name, mac, kws) else: raise Exception(f'Unknown light device: {name}, {mac}, {kws}') elif 'outlet' in kws.lower(): if 'tplink' in kws.lower(): if 'children' in kws.lower(): + logger.debug(' ...a TPLinkOutletWithChildren') return outlets.TPLinkOutletWithChildren(name, mac, kws) else: + logger.debug(' ...a TPLinkOutlet') return outlets.TPLinkOutlet(name, mac, kws) elif 'goog' in kws.lower(): + logger.debug(' ...a GoogleOutlet') return outlets.GoogleOutlet(name, mac, kws) else: raise Exception(f'Unknown outlet device: {name}, {mac}, {kws}') + elif 'camera' in kws.lower(): + logger.debug(' ...a BaseCamera') + return cameras.BaseCamera(name, mac, kws) else: + logger.debug(' ...an unknown device (should this be here?)') return device.Device(name, mac, kws) + logger.warning(f'{mac} is not known, returning None') return None def query(self, query: str) -> List[device.Device]: @@ -154,6 +179,7 @@ class SmartHomeConfig(object): Returns a list of matching lights. """ retval = [] + logger.debug(f'Executing query {query}') results = self._corpus.query(query) if results is not None: for mac in results: |