Smart outlets

1 files changed, 164 insertions, 0 deletions

diff --git a/smart_home/config.py b/smart_home/config.py
new file mode 100644
index 0000000..723e097
--- /dev/null
+++ b/smart_home/config.py
@@ -0,0 +1,164 @@
+#!/usr/bin/env python3
+
+import re
+from typing import List, Optional, Set
+
+import argparse_utils
+import config
+import file_utils
+import logical_search
+import smart_home.device as device
+import smart_home.lights as lights
+import smart_home.outlets as outlets
+
+parser = config.add_commandline_args(
+    f"Smart Home Config ({__file__})",
+    "Args related to the smart home config."
+)
+parser.add_argument(
+    '--smart_home_config_file_location',
+    default='/home/scott/bin/network_mac_addresses.txt',
+    metavar='FILENAME',
+    help='The location of network_mac_addresses.txt',
+    type=argparse_utils.valid_filename,
+)
+
+
+class SmartHomeConfig(object):
+    def __init__(
+            self,
+            config_file: Optional[str] = None,
+            filters: List[str] = ['smart'],
+    ) -> None:
+        if config_file is None:
+            config_file = config.config[
+                'smart_home_config_file_location'
+            ]
+        assert file_utils.does_file_exist(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()
+
+        for line in contents:
+            line = line.rstrip("\n")
+            line = re.sub(r"#.*$", r"", line)
+            line = line.strip()
+            if line == "":
+                continue
+            (mac, name, keywords) = line.split(",")
+            mac = mac.strip()
+            name = name.strip()
+            keywords = keywords.strip()
+
+            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)
+
+    def index_device(self, name: str, keywords: str, mac: str) -> None:
+        properties = [("name", name)]
+        tags = set()
+        for kw in keywords.split():
+            if ":" in kw:
+                key, value = kw.split(":")
+                properties.append((key, value))
+            else:
+                tags.add(kw)
+        device = logical_search.Document(
+            docid=mac,
+            tags=tags,
+            properties=properties,
+            reference=None,
+        )
+        self._corpus.add_doc(device)
+
+    def __repr__(self) -> str:
+        s = "Known devices:\n"
+        for name, keywords in self._keywords_by_name.items():
+            mac = self._macs_by_name[name]
+            s += f"  {name} ({mac}) => {keywords}\n"
+        return s
+
+    def get_keywords_by_name(self, name: str) -> Optional[device.Device]:
+        return self._keywords_by_name.get(name, None)
+
+    def get_macs_by_name(self, name: str) -> Set[str]:
+        retval = set()
+        for (mac, lname) in self._names_by_mac.items():
+            if name in lname:
+                retval.add(mac)
+        return retval
+
+    def get_macs_by_keyword(self, keyword: str) -> Set[str]:
+        retval = set()
+        for (mac, keywords) in self._keywords_by_mac.items():
+            if keyword in keywords:
+                retval.add(mac)
+        return retval
+
+    def get_device_by_name(self, name: str) -> Optional[device.Device]:
+        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]:
+        retval = []
+        for (mac, kws) 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)
+        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]
+            if 'light' in kws.lower():
+                if 'tplink' in kws.lower():
+                    return lights.TPLinkLight(name, mac, kws)
+                elif 'tuya' in kws.lower():
+                    return lights.TuyaLight(name, mac, kws)
+                elif 'goog' in kws.lower():
+                    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():
+                        return outlets.TPLinkOutletWithChildren(name, mac, kws)
+                    else:
+                        return outlets.TPLinkOutlet(name, mac, kws)
+                elif 'goog' in kws.lower():
+                    return outlets.GoogleOutlet(name, mac, kws)
+                else:
+                    raise Exception(f'Unknown outlet device: {name}, {mac}, {kws}')
+            else:
+                return device.Device(name, mac, kws)
+        return None
+
+    def query(self, query: str) -> List[device.Device]:
+        """Evaluates a lighting query expression formed of keywords to search
+        for, logical operators (and, or, not), and parenthesis.
+        Returns a list of matching lights.
+        """
+        retval = []
+        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)
+        return retval