5 from typing import List, Optional, Set
11 import smart_home.device as device
12 import smart_home.cameras as cameras
13 import smart_home.lights as lights
14 import smart_home.outlets as outlets
16 parser = config.add_commandline_args(
17 f"Smart Home Config ({__file__})",
18 "Args related to the smart home config."
21 '--smart_home_config_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(__file__)
32 class SmartHomeConfig(object):
35 config_file: Optional[str] = None,
36 filters: List[str] = ['smart'],
38 self._macs_by_name = {}
39 self._keywords_by_name = {}
40 self._keywords_by_mac = {}
41 self._names_by_mac = {}
42 self._corpus = logical_search.Corpus()
44 # Read the disk config file...
45 if config_file is None:
46 config_file = config.config[
47 'smart_home_config_file_location'
49 assert file_utils.does_file_exist(config_file)
50 logger.debug(f'Reading {config_file}')
51 with open(config_file, "r") as f:
52 contents = f.readlines()
54 # Parse the contents...
56 line = line.rstrip("\n")
57 line = re.sub(r"#.*$", r"", line)
61 logger.debug(f'> {line}')
62 (mac, name, keywords) = line.split(",")
65 keywords = keywords.strip()
68 if filters is not None:
71 logger.debug(f'Skipping this entry b/c of filter {f}')
75 self._macs_by_name[name] = mac
76 self._keywords_by_name[name] = keywords
77 self._keywords_by_mac[mac] = keywords
78 self._names_by_mac[mac] = name
79 self.index_device(name, keywords, mac)
81 def index_device(self, name: str, keywords: str, mac: str) -> None:
82 properties = [("name", name)]
84 for kw in keywords.split():
86 key, value = kw.split(":")
87 properties.append((key, value))
90 device = logical_search.Document(
93 properties=properties,
96 logger.debug(f'Indexing document {device}')
97 self._corpus.add_doc(device)
99 def __repr__(self) -> str:
100 s = "Known devices:\n"
101 for name, keywords in self._keywords_by_name.items():
102 mac = self._macs_by_name[name]
103 s += f" {name} ({mac}) => {keywords}\n"
106 def get_keywords_by_name(self, name: str) -> Optional[device.Device]:
107 return self._keywords_by_name.get(name, None)
109 def get_macs_by_name(self, name: str) -> Set[str]:
111 for (mac, lname) in self._names_by_mac.items():
116 def get_macs_by_keyword(self, keyword: str) -> Set[str]:
118 for (mac, keywords) in self._keywords_by_mac.items():
119 if keyword in keywords:
123 def get_device_by_name(self, name: str) -> Optional[device.Device]:
124 if name in self._macs_by_name:
125 return self.get_device_by_mac(self._macs_by_name[name])
128 def get_all_devices(self) -> List[device.Device]:
130 for (mac, kws) in self._keywords_by_mac.items():
132 device = self.get_device_by_mac(mac)
133 if device is not None:
134 retval.append(device)
137 def get_device_by_mac(self, mac: str) -> Optional[device.Device]:
138 if mac in self._keywords_by_mac:
139 name = self._names_by_mac[mac]
140 kws = self._keywords_by_mac[mac]
141 logger.debug(f'Found {name} -> {mac} ({kws})')
142 if 'light' in kws.lower():
143 if 'tplink' in kws.lower():
144 logger.debug(' ...a TPLinkLight')
145 return lights.TPLinkLight(name, mac, kws)
146 elif 'tuya' in kws.lower():
147 logger.debug(' ...a TuyaLight')
148 return lights.TuyaLight(name, mac, kws)
149 elif 'goog' in kws.lower():
150 logger.debug(' ...a GoogleLight')
151 return lights.GoogleLight(name, mac, kws)
153 raise Exception(f'Unknown light device: {name}, {mac}, {kws}')
154 elif 'outlet' in kws.lower():
155 if 'tplink' in kws.lower():
156 if 'children' in kws.lower():
157 logger.debug(' ...a TPLinkOutletWithChildren')
158 return outlets.TPLinkOutletWithChildren(name, mac, kws)
160 logger.debug(' ...a TPLinkOutlet')
161 return outlets.TPLinkOutlet(name, mac, kws)
162 elif 'goog' in kws.lower():
163 logger.debug(' ...a GoogleOutlet')
164 return outlets.GoogleOutlet(name, mac, kws)
166 raise Exception(f'Unknown outlet device: {name}, {mac}, {kws}')
167 elif 'camera' in kws.lower():
168 logger.debug(' ...a BaseCamera')
169 return cameras.BaseCamera(name, mac, kws)
171 logger.debug(' ...an unknown device (should this be here?)')
172 return device.Device(name, mac, kws)
173 logger.warning(f'{mac} is not known, returning None')
176 def query(self, query: str) -> List[device.Device]:
177 """Evaluates a lighting query expression formed of keywords to search
178 for, logical operators (and, or, not), and parenthesis.
179 Returns a list of matching lights.
182 logger.debug(f'Executing query {query}')
183 results = self._corpus.query(query)
184 if results is not None:
187 device = self.get_device_by_mac(mac)
188 if device is not None:
189 retval.append(device)