3 """Utilities for dealing with the smart outlets."""
14 from abc import abstractmethod
15 from typing import Any, Dict, List, Optional
17 from meross_iot.http_api import MerossHttpClient
18 from meross_iot.manager import MerossManager
19 from overrides import overrides
23 import decorator_utils
26 import smart_home.device as dev
27 from decorator_utils import memoized, timeout
28 from google_assistant import GoogleResponse, ask_google
30 logger = logging.getLogger(__name__)
32 parser = config.add_commandline_args(
33 f"Outlet Utils ({__file__})",
34 "Args related to smart outlets.",
37 '--smart_outlets_tplink_location',
38 default='/home/scott/bin/tplink.py',
40 help='The location of the tplink.py helper',
41 type=argparse_utils.valid_filename,
45 @timeout(5.0, use_signals=False, error_message="Timed out waiting for tplink.py")
46 def tplink_outlet_command(command: str) -> bool:
47 result = os.system(command)
48 signal = result & 0xFF
50 msg = f'{command} died with signal {signal}'
52 logging_utils.hlog(msg)
55 exit_value = result >> 8
57 msg = f'{command} failed, exited {exit_value}'
59 logging_utils.hlog(msg)
61 logger.debug('%s succeeded.', command)
65 class BaseOutlet(dev.Device):
66 """An abstract base class for smart outlets."""
68 def __init__(self, name: str, mac: str, keywords: str = "") -> None:
69 super().__init__(name.strip(), mac.strip(), keywords)
72 def turn_on(self) -> bool:
76 def turn_off(self) -> bool:
80 def is_on(self) -> bool:
84 def is_off(self) -> bool:
88 class TPLinkOutlet(BaseOutlet):
89 """A TPLink smart outlet."""
91 def __init__(self, name: str, mac: str, keywords: str = '') -> None:
92 super().__init__(name, mac, keywords)
93 self.info: Optional[Dict] = None
94 self.info_ts: Optional[datetime.datetime] = None
97 def get_tplink_name(self) -> Optional[str]:
98 self.info = self.get_info()
99 if self.info is not None:
100 return self.info["alias"]
103 def get_cmdline(self) -> str:
105 f"{config.config['smart_outlets_tplink_location']} -m {self.mac} "
106 f"--no_logging_console "
110 def command(self, cmd: str, extra_args: str = None, **kwargs) -> bool:
111 cmd = self.get_cmdline() + f"-c {cmd}"
112 if extra_args is not None:
113 cmd += f" {extra_args}"
114 return tplink_outlet_command(cmd)
117 def turn_on(self) -> bool:
118 return self.command('on')
121 def turn_off(self) -> bool:
122 return self.command('off')
125 def is_on(self) -> bool:
126 return self.get_on_duration_seconds() > 0
129 def is_off(self) -> bool:
130 return not self.is_on()
132 @timeout(10.0, use_signals=False, error_message="Timed out waiting for tplink.py")
133 def get_info(self) -> Optional[Dict]:
134 cmd = self.get_cmdline() + "-c info"
135 out = subprocess.getoutput(cmd)
136 out = re.sub("Sent:.*\n", "", out)
137 out = re.sub("Received: *", "", out)
139 self.info = json.loads(out)["system"]["get_sysinfo"]
140 self.info_ts = datetime.datetime.now()
142 except Exception as e:
144 print(out, file=sys.stderr)
149 def get_on_duration_seconds(self) -> int:
150 self.info = self.get_info()
151 if self.info is None:
153 return int(self.info.get("on_time", "0"))
156 class TPLinkOutletWithChildren(TPLinkOutlet):
157 """A TPLink outlet where the top and bottom plus are individually
160 def __init__(self, name: str, mac: str, keywords: str = "") -> None:
161 super().__init__(name, mac, keywords)
162 self.children: List[str] = []
163 self.info: Optional[Dict] = None
164 self.info_ts: Optional[datetime.datetime] = None
165 assert self.keywords is not None
166 assert "children" in self.keywords
167 self.info = self.get_info()
168 if self.info is not None:
169 for child in self.info["children"]:
170 self.children.append(child["id"])
173 def get_cmdline(self, child: Optional[str] = None) -> str:
175 f"{config.config['smart_outlets_tplink_location']} -m {self.mac} "
176 f"--no_logging_console "
178 if child is not None:
179 cmd += f"-x {child} "
183 def command(self, cmd: str, extra_args: str = None, **kwargs) -> bool:
184 child: Optional[str] = kwargs.get('child', None)
185 cmd = self.get_cmdline(child) + f"-c {cmd}"
186 if extra_args is not None:
187 cmd += f" {extra_args}"
188 logger.debug('About to execute: %s', cmd)
189 return tplink_outlet_command(cmd)
191 def get_children(self) -> List[str]:
195 def turn_on(self, child: str = None) -> bool:
196 return self.command("on", child)
199 def turn_off(self, child: str = None) -> bool:
200 return self.command("off", child)
202 def get_on_duration_seconds(self, child: str = None) -> int:
203 self.info = self.get_info()
205 if self.info is None:
207 return int(self.info.get("on_time", "0"))
209 if self.info is None:
211 for chi in self.info.get("children", {}):
212 if chi["id"] == child:
213 return int(chi.get("on_time", "0"))
217 class GoogleOutlet(BaseOutlet):
218 """A smart outlet controlled via Google Assistant."""
220 def __init__(self, name: str, mac: str, keywords: str = "") -> None:
221 super().__init__(name.strip(), mac.strip(), keywords)
224 def goog_name(self) -> str:
225 name = self.get_name()
226 return name.replace('_', ' ')
229 def parse_google_response(response: GoogleResponse) -> bool:
230 return response.success
233 def turn_on(self) -> bool:
234 return GoogleOutlet.parse_google_response(ask_google(f'turn {self.goog_name()} on'))
237 def turn_off(self) -> bool:
238 return GoogleOutlet.parse_google_response(ask_google(f'turn {self.goog_name()} off'))
241 def is_on(self) -> bool:
242 r = ask_google(f'is {self.goog_name()} on?')
245 if r.audio_transcription is not None:
246 return 'is on' in r.audio_transcription
247 raise Exception('Can\'t talk to Google right now!?')
250 def is_off(self) -> bool:
251 return not self.is_on()
254 @decorator_utils.singleton
255 class MerossWrapper(object):
256 """Global singleton helper class for MerossOutlets. Note that
257 instantiating this class causes HTTP traffic with an external
258 Meross server. Meross blocks customers who hit their servers too
259 aggressively so MerossOutlet is lazy about creating instances of
265 self.loop = asyncio.get_event_loop()
266 self.email = os.environ.get('MEROSS_EMAIL') or scott_secrets.MEROSS_EMAIL
267 self.password = os.environ.get('MEROSS_PASSWORD') or scott_secrets.MEROSS_PASSWORD
268 self.devices = self.loop.run_until_complete(self.find_meross_devices())
269 atexit.register(self.loop.close)
271 async def find_meross_devices(self) -> List[Any]:
272 http_api_client = await MerossHttpClient.async_from_user_password(
273 email=self.email, password=self.password
276 # Setup and start the device manager
277 manager = MerossManager(http_client=http_api_client)
278 await manager.async_init()
281 await manager.async_device_discovery()
282 devices = manager.find_devices()
283 for device in devices:
284 await device.async_update()
287 def get_meross_device_by_name(self, name: str) -> Optional[Any]:
289 name = name.replace('_', ' ')
290 for device in self.devices:
291 if device.name.lower() == name:
296 class MerossOutlet(BaseOutlet):
297 """A Meross smart outlet class."""
299 def __init__(self, name: str, mac: str, keywords: str = '') -> None:
300 super().__init__(name, mac, keywords)
301 self.meross_wrapper: Optional[MerossWrapper] = None
302 self.device: Optional[Any] = None
304 def lazy_initialize_device(self):
305 """If we make too many calls to Meross they will block us; only talk
306 to them when someone actually wants to control a device."""
307 if self.meross_wrapper is None:
308 self.meross_wrapper = MerossWrapper()
309 self.device = self.meross_wrapper.get_meross_device_by_name(self.name)
310 if self.device is None:
311 raise Exception(f'{self.name} is not a known Meross device?!')
314 def turn_on(self) -> bool:
315 self.lazy_initialize_device()
316 assert self.meross_wrapper is not None
317 assert self.device is not None
318 self.meross_wrapper.loop.run_until_complete(self.device.async_turn_on())
322 def turn_off(self) -> bool:
323 self.lazy_initialize_device()
324 assert self.meross_wrapper is not None
325 assert self.device is not None
326 self.meross_wrapper.loop.run_until_complete(self.device.async_turn_off())
330 def is_on(self) -> bool:
331 self.lazy_initialize_device()
332 assert self.device is not None
333 return self.device.is_on()
336 def is_off(self) -> bool:
337 return not self.is_on()