5 from typing import List, Optional, Set
12 import smart_home.device as device
13 import smart_home.cameras as cameras
14 import smart_home.chromecasts as chromecasts
15 import smart_home.lights as lights
16 import smart_home.outlets as 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(__file__)
34 class SmartHomeRegistry(object):
37 registry_file: Optional[str] = None,
38 filters: List[str] = ['smart'],
40 self._macs_by_name = {}
41 self._keywords_by_name = {}
42 self._keywords_by_mac = {}
43 self._names_by_mac = {}
44 self._corpus = logical_search.Corpus()
46 # Read the disk config file...
47 if registry_file is None:
48 registry_file = config.config[
49 'smart_home_registry_file_location'
51 assert file_utils.does_file_exist(registry_file)
52 logger.debug(f'Reading {registry_file}')
53 with open(registry_file, "r") as f:
54 contents = f.readlines()
56 # Parse the contents...
58 line = line.rstrip("\n")
59 line = re.sub(r"#.*$", r"", line)
63 logger.debug(f'SH-CONFIG> {line}')
65 (mac, name, keywords) = line.split(",")
67 msg = f'SH-CONFIG> {line} is malformed?!'
73 keywords = keywords.strip()
76 if filters is not None:
79 logger.debug(f'Skipping this entry b/c of filter {f}')
83 self._macs_by_name[name] = mac
84 self._keywords_by_name[name] = keywords
85 self._keywords_by_mac[mac] = keywords
86 self._names_by_mac[mac] = name
87 self.index_device(name, keywords, mac)
89 def index_device(self, name: str, keywords: str, mac: str) -> None:
90 properties = [("name", name)]
92 for kw in keywords.split():
94 key, value = kw.split(":")
95 properties.append((key, value))
98 device = logical_search.Document(
101 properties=properties,
104 logger.debug(f'Indexing document {device}')
105 self._corpus.add_doc(device)
107 def __repr__(self) -> str:
108 s = "Known devices:\n"
109 for name, keywords in self._keywords_by_name.items():
110 mac = self._macs_by_name[name]
111 s += f" {name} ({mac}) => {keywords}\n"
114 def get_keywords_by_name(self, name: str) -> Optional[device.Device]:
115 return self._keywords_by_name.get(name, None)
117 def get_macs_by_name(self, name: str) -> Set[str]:
119 for (mac, lname) in self._names_by_mac.items():
124 def get_macs_by_keyword(self, keyword: str) -> Set[str]:
126 for (mac, keywords) in self._keywords_by_mac.items():
127 if keyword in keywords:
131 def get_device_by_name(self, name: str) -> Optional[device.Device]:
132 if name in self._macs_by_name:
133 return self.get_device_by_mac(self._macs_by_name[name])
136 def get_all_devices(self) -> List[device.Device]:
138 for (mac, kws) in self._keywords_by_mac.items():
140 device = self.get_device_by_mac(mac)
141 if device is not None:
142 retval.append(device)
145 def get_device_by_mac(self, mac: str) -> Optional[device.Device]:
146 if mac in self._keywords_by_mac:
147 name = self._names_by_mac[mac]
148 kws = self._keywords_by_mac[mac]
149 logger.debug(f'Found {name} -> {mac} ({kws})')
151 if 'light' in kws.lower():
152 if 'tplink' in kws.lower():
153 logger.debug(' ...a TPLinkLight')
154 return lights.TPLinkLight(name, mac, kws)
155 elif 'tuya' in kws.lower():
156 logger.debug(' ...a TuyaLight')
157 return lights.TuyaLight(name, mac, kws)
158 elif 'goog' in kws.lower():
159 logger.debug(' ...a GoogleLight')
160 return lights.GoogleLight(name, mac, kws)
162 raise Exception(f'Unknown light device: {name}, {mac}, {kws}')
163 elif 'outlet' in kws.lower():
164 if 'tplink' in kws.lower():
165 if 'children' in kws.lower():
166 logger.debug(' ...a TPLinkOutletWithChildren')
167 return outlets.TPLinkOutletWithChildren(name, mac, kws)
169 logger.debug(' ...a TPLinkOutlet')
170 return outlets.TPLinkOutlet(name, mac, kws)
171 elif 'meross' in kws.lower():
172 logger.debug(' ...a MerossOutlet')
173 return outlets.MerossOutlet(name, mac, kws)
174 elif 'goog' in kws.lower():
175 logger.debug(' ...a GoogleOutlet')
176 return outlets.GoogleOutlet(name, mac, kws)
178 raise Exception(f'Unknown outlet device: {name}, {mac}, {kws}')
179 elif 'camera' in kws.lower():
180 logger.debug(' ...a BaseCamera')
181 return cameras.BaseCamera(name, mac, kws)
182 elif 'ccast' in kws.lower():
183 logger.debug(' ...a Chromecast')
184 return chromecasts.BaseChromecast(name, mac, kws)
186 logger.debug(' ...an unknown device (should this be here?)')
187 return device.Device(name, mac, kws)
188 except Exception as e:
190 return device.Device(name, mac, kws)
191 msg = f'{mac} is not a known smart home device, returning None'
196 def query(self, query: str) -> List[device.Device]:
197 """Evaluates a lighting query expression formed of keywords to search
198 for, logical operators (and, or, not), and parenthesis.
199 Returns a list of matching lights.
202 logger.debug(f'Executing query {query}')
203 results = self._corpus.query(query)
204 if results is not None:
207 device = self.get_device_by_mac(mac)
208 if device is not None:
209 retval.append(device)