3 # © Copyright 2021-2022, Scott Gasch
5 """A searchable registry of known smart home devices and a factory for
6 constructing our wrappers around them."""
10 from typing import Dict, List, Optional, Set
16 from smart_home import cameras, chromecasts, device, lights, outlets
18 args = config.add_commandline_args(
19 f"Smart Home Registry ({__file__})",
20 "Args related to the smart home configuration registry.",
23 '--smart_home_registry_file_location',
24 default='/home/scott/bin/network_mac_addresses.txt',
26 help='The location of network_mac_addresses.txt',
27 type=argparse_utils.valid_filename,
31 logger = logging.getLogger(__name__)
34 class SmartHomeRegistry(object):
35 """A searchable registry of known smart home devices and a factory for
36 constructing our wrappers around them."""
40 registry_file: Optional[str] = None,
41 filters: List[str] = ['smart'],
43 self._macs_by_name: Dict[str, str] = {}
44 self._keywords_by_name: Dict[str, str] = {}
45 self._keywords_by_mac: Dict[str, str] = {}
46 self._names_by_mac: Dict[str, str] = {}
47 self._corpus: logical_search.Corpus = logical_search.Corpus()
49 # Read the disk config file...
50 if registry_file is None:
51 registry_file = config.config['smart_home_registry_file_location']
52 assert file_utils.does_file_exist(registry_file)
53 logger.debug('Reading %s', registry_file)
54 with open(registry_file, "r") as rf:
55 contents = rf.readlines()
57 # Parse the contents...
59 line = line.rstrip("\n")
60 line = re.sub(r"#.*$", r"", line)
64 logger.debug('SH-CONFIG> %s', line)
66 (mac, name, keywords) = line.split(",")
68 logger.warning('SH-CONFIG> "%s" is malformed?! Skipping it.', line)
72 keywords = keywords.strip()
75 if filters is not None:
78 logger.debug('Skipping this entry b/c of filter: %s', f)
82 self._macs_by_name[name] = mac
83 self._keywords_by_name[name] = keywords
84 self._keywords_by_mac[mac] = keywords
85 self._names_by_mac[mac] = name
86 self._index_device(name, keywords, mac)
88 def _index_device(self, name: str, keywords: str, mac: str) -> None:
89 properties = [("name", name)]
91 for kw in keywords.split():
93 key, value = kw.split(":")
94 properties.append((key, value))
97 dev = logical_search.Document(
100 properties=properties,
103 logger.debug('Indexing document: %s', dev)
104 self._corpus.add_doc(dev)
106 def __repr__(self) -> str:
107 s = "Known devices:\n"
108 for name, keywords in self._keywords_by_name.items():
109 mac = self._macs_by_name[name]
110 s += f" {name} ({mac}) => {keywords}\n"
113 def get_keywords_by_name(self, name: str) -> Optional[str]:
114 """Given the name of a device, get its keywords.
116 >>> reg = SmartHomeRegistry('/home/scott/bin/network_mac_addresses.txt')
117 >>> reg.get_keywords_by_name('near_kitchen_lamp')
118 'wifi smart light goog meross test'
120 >>> reg.get_keywords_by_name('unknown') is None
124 return self._keywords_by_name.get(name, None)
126 def get_macs_by_name(self, name: str) -> Set[str]:
127 """Given the name of a device, get its MAC address(es)
129 >>> reg = SmartHomeRegistry('/home/scott/bin/network_mac_addresses.txt')
130 >>> reg.get_macs_by_name('near_kitchen_lamp')
131 {'34:29:8F:12:34:8E'}
133 >>> reg.get_macs_by_name('unknown')
139 for (mac, lname) in self._names_by_mac.items():
144 def get_macs_by_keyword(self, keyword: str) -> Set[str]:
145 """Given a keyword, return the set of MAC address(es) that have
148 >>> reg = SmartHomeRegistry('/home/scott/bin/network_mac_addresses.txt')
149 >>> r = reg.get_macs_by_keyword('test')
150 >>> e = set(['34:29:8F:12:26:74' , '34:29:8F:12:34:8E'])
154 >>> reg.get_macs_by_keyword('unknown')
159 for (mac, keywords) in self._keywords_by_mac.items():
160 if keyword in keywords:
164 def get_device_by_name(self, name: str) -> Optional[device.Device]:
165 """Given a name, return its Device object."""
167 if name in self._macs_by_name:
168 return self.get_device_by_mac(self._macs_by_name[name])
171 def get_all_devices(self) -> List[device.Device]:
172 """Return a list of all known devices."""
175 for mac, _ in self._keywords_by_mac.items():
177 dev = self.get_device_by_mac(mac)
182 def get_device_by_mac(self, mac: str) -> Optional[device.Device]:
183 """Given a MAC address, return its Device object."""
185 if mac in self._keywords_by_mac:
186 name = self._names_by_mac[mac]
187 kws = self._keywords_by_mac[mac]
188 logger.debug('Found %s -> %s (%s)', name, mac, kws)
190 if 'light' in kws.lower():
191 if 'tplink' in kws.lower():
192 logger.debug(' ...a TPLinkLight')
193 return lights.TPLinkLight(name, mac, kws)
194 elif 'tuya' in kws.lower():
195 logger.debug(' ...a TuyaLight')
196 return lights.TuyaLight(name, mac, kws)
197 elif 'goog' in kws.lower():
198 logger.debug(' ...a GoogleLight')
199 return lights.GoogleLight(name, mac, kws)
201 raise Exception(f'Unknown light device: {name}, {mac}, {kws}')
202 elif 'outlet' in kws.lower():
203 if 'tplink' in kws.lower():
204 if 'children' in kws.lower():
205 logger.debug(' ...a TPLinkOutletWithChildren')
206 return outlets.TPLinkOutletWithChildren(name, mac, kws)
208 logger.debug(' ...a TPLinkOutlet')
209 return outlets.TPLinkOutlet(name, mac, kws)
210 elif 'meross' in kws.lower():
211 logger.debug(' ...a MerossOutlet')
212 return outlets.MerossOutlet(name, mac, kws)
213 elif 'goog' in kws.lower():
214 logger.debug(' ...a GoogleOutlet')
215 return outlets.GoogleOutlet(name, mac, kws)
217 raise Exception(f'Unknown outlet device: {name}, {mac}, {kws}')
218 elif 'camera' in kws.lower():
219 logger.debug(' ...a BaseCamera')
220 return cameras.BaseCamera(name, mac, kws)
221 elif 'ccast' in kws.lower():
222 logger.debug(' ...a Chromecast')
223 return chromecasts.BaseChromecast(name, mac, kws)
225 logger.debug(' ...an unknown device (should this be here?)')
226 return device.Device(name, mac, kws)
227 except Exception as e:
230 'Device %s at %s with %s confused me; returning a generic Device',
235 return device.Device(name, mac, kws)
236 logger.warning('%s is not a known smart home device, returning None', mac)
239 def query(self, query: str) -> List[device.Device]:
240 """Evaluates a device query expression formed of keywords to search
241 for, logical operators (and, or, not), and parenthesis.
242 Returns a list of matching lights.
245 logger.debug('Executing query: %s', query)
246 results = self._corpus.query(query)
247 if results is not None:
250 dev = self.get_device_by_mac(mac)
256 if __name__ == '__main__':