3 """A searchable registry of known smart home devices and a factory for
4 constructing our wrappers around them."""
8 from typing import Dict, List, Optional, Set
14 from smart_home import cameras, chromecasts, device, lights, outlets
16 args = config.add_commandline_args(
17 f"Smart Home Registry ({__file__})",
18 "Args related to the smart home configuration registry.",
21 '--smart_home_registry_file_location',
22 default='/home/scott/bin/network_mac_addresses.txt',
24 help='The location of network_mac_addresses.txt',
25 type=argparse_utils.valid_filename,
29 logger = logging.getLogger(__name__)
32 class SmartHomeRegistry(object):
33 """A searchable registry of known smart home devices and a factory for
34 constructing our wrappers around them."""
38 registry_file: Optional[str] = None,
39 filters: List[str] = ['smart'],
41 self._macs_by_name: Dict[str, str] = {}
42 self._keywords_by_name: Dict[str, str] = {}
43 self._keywords_by_mac: Dict[str, str] = {}
44 self._names_by_mac: Dict[str, str] = {}
45 self._corpus: logical_search.Corpus = logical_search.Corpus()
47 # Read the disk config file...
48 if registry_file is None:
49 registry_file = config.config['smart_home_registry_file_location']
50 assert file_utils.does_file_exist(registry_file)
51 logger.debug('Reading %s', registry_file)
52 with open(registry_file, "r") as rf:
53 contents = rf.readlines()
55 # Parse the contents...
57 line = line.rstrip("\n")
58 line = re.sub(r"#.*$", r"", line)
62 logger.debug('SH-CONFIG> %s', line)
64 (mac, name, keywords) = line.split(",")
66 logger.warning('SH-CONFIG> "%s" is malformed?! Skipping it.', line)
70 keywords = keywords.strip()
73 if filters is not None:
76 logger.debug('Skipping this entry b/c of filter: %s', f)
80 self._macs_by_name[name] = mac
81 self._keywords_by_name[name] = keywords
82 self._keywords_by_mac[mac] = keywords
83 self._names_by_mac[mac] = name
84 self.index_device(name, keywords, mac)
86 def index_device(self, name: str, keywords: str, mac: str) -> None:
87 properties = [("name", name)]
89 for kw in keywords.split():
91 key, value = kw.split(":")
92 properties.append((key, value))
95 dev = logical_search.Document(
98 properties=properties,
101 logger.debug('Indexing document: %s', dev)
102 self._corpus.add_doc(dev)
104 def __repr__(self) -> str:
105 s = "Known devices:\n"
106 for name, keywords in self._keywords_by_name.items():
107 mac = self._macs_by_name[name]
108 s += f" {name} ({mac}) => {keywords}\n"
111 def get_keywords_by_name(self, name: str) -> Optional[str]:
112 return self._keywords_by_name.get(name, None)
114 def get_macs_by_name(self, name: str) -> Set[str]:
116 for (mac, lname) in self._names_by_mac.items():
121 def get_macs_by_keyword(self, keyword: str) -> Set[str]:
123 for (mac, keywords) in self._keywords_by_mac.items():
124 if keyword in keywords:
128 def get_device_by_name(self, name: str) -> Optional[device.Device]:
129 if name in self._macs_by_name:
130 return self.get_device_by_mac(self._macs_by_name[name])
133 def get_all_devices(self) -> List[device.Device]:
135 for mac, _ in self._keywords_by_mac.items():
137 dev = self.get_device_by_mac(mac)
142 def get_device_by_mac(self, mac: str) -> Optional[device.Device]:
143 if mac in self._keywords_by_mac:
144 name = self._names_by_mac[mac]
145 kws = self._keywords_by_mac[mac]
146 logger.debug('Found %s -> %s (%s)', name, mac, kws)
148 if 'light' in kws.lower():
149 if 'tplink' in kws.lower():
150 logger.debug(' ...a TPLinkLight')
151 return lights.TPLinkLight(name, mac, kws)
152 elif 'tuya' in kws.lower():
153 logger.debug(' ...a TuyaLight')
154 return lights.TuyaLight(name, mac, kws)
155 elif 'goog' in kws.lower():
156 logger.debug(' ...a GoogleLight')
157 return lights.GoogleLight(name, mac, kws)
159 raise Exception(f'Unknown light device: {name}, {mac}, {kws}')
160 elif 'outlet' in kws.lower():
161 if 'tplink' in kws.lower():
162 if 'children' in kws.lower():
163 logger.debug(' ...a TPLinkOutletWithChildren')
164 return outlets.TPLinkOutletWithChildren(name, mac, kws)
166 logger.debug(' ...a TPLinkOutlet')
167 return outlets.TPLinkOutlet(name, mac, kws)
168 elif 'meross' in kws.lower():
169 logger.debug(' ...a MerossOutlet')
170 return outlets.MerossOutlet(name, mac, kws)
171 elif 'goog' in kws.lower():
172 logger.debug(' ...a GoogleOutlet')
173 return outlets.GoogleOutlet(name, mac, kws)
175 raise Exception(f'Unknown outlet device: {name}, {mac}, {kws}')
176 elif 'camera' in kws.lower():
177 logger.debug(' ...a BaseCamera')
178 return cameras.BaseCamera(name, mac, kws)
179 elif 'ccast' in kws.lower():
180 logger.debug(' ...a Chromecast')
181 return chromecasts.BaseChromecast(name, mac, kws)
183 logger.debug(' ...an unknown device (should this be here?)')
184 return device.Device(name, mac, kws)
185 except Exception as e:
188 'Device %s at %s with %s confused me; returning a generic Device',
193 return device.Device(name, mac, kws)
194 logger.warning('%s is not a known smart home device, returning None', mac)
197 def query(self, query: str) -> List[device.Device]:
198 """Evaluates a lighting query expression formed of keywords to search
199 for, logical operators (and, or, not), and parenthesis.
200 Returns a list of matching lights.
203 logger.debug('Executing query: %s', query)
204 results = self._corpus.query(query)
205 if results is not None:
208 dev = self.get_device_by_mac(mac)