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(f'{command} succeeded.')
65 class BaseOutlet(dev.Device):
66 def __init__(self, name: str, mac: str, keywords: str = "") -> None:
67 super().__init__(name.strip(), mac.strip(), keywords)
70 def turn_on(self) -> bool:
74 def turn_off(self) -> bool:
78 def is_on(self) -> bool:
82 def is_off(self) -> bool:
86 class TPLinkOutlet(BaseOutlet):
87 def __init__(self, name: str, mac: str, keywords: str = '') -> None:
88 super().__init__(name, mac, keywords)
89 self.info: Optional[Dict] = None
90 self.info_ts: Optional[datetime.datetime] = None
93 def get_tplink_name(self) -> Optional[str]:
94 self.info = self.get_info()
95 if self.info is not None:
96 return self.info["alias"]
99 def get_cmdline(self) -> str:
101 f"{config.config['smart_outlets_tplink_location']} -m {self.mac} "
102 f"--no_logging_console "
106 def command(self, cmd: str, extra_args: str = None, **kwargs) -> bool:
107 cmd = self.get_cmdline() + f"-c {cmd}"
108 if extra_args is not None:
109 cmd += f" {extra_args}"
110 return tplink_outlet_command(cmd)
113 def turn_on(self) -> bool:
114 return self.command('on')
117 def turn_off(self) -> bool:
118 return self.command('off')
121 def is_on(self) -> bool:
122 return self.get_on_duration_seconds() > 0
125 def is_off(self) -> bool:
126 return not self.is_on()
128 @timeout(10.0, use_signals=False, error_message="Timed out waiting for tplink.py")
129 def get_info(self) -> Optional[Dict]:
130 cmd = self.get_cmdline() + "-c info"
131 out = subprocess.getoutput(cmd)
132 out = re.sub("Sent:.*\n", "", out)
133 out = re.sub("Received: *", "", out)
135 self.info = json.loads(out)["system"]["get_sysinfo"]
136 self.info_ts = datetime.datetime.now()
138 except Exception as e:
140 print(out, file=sys.stderr)
145 def get_on_duration_seconds(self) -> int:
146 self.info = self.get_info()
147 if self.info is None:
149 return int(self.info.get("on_time", "0"))
152 class TPLinkOutletWithChildren(TPLinkOutlet):
153 def __init__(self, name: str, mac: str, keywords: str = "") -> None:
154 super().__init__(name, mac, keywords)
155 self.children: List[str] = []
156 self.info: Optional[Dict] = None
157 self.info_ts: Optional[datetime.datetime] = None
158 assert "children" in self.keywords
159 self.info = self.get_info()
160 if self.info is not None:
161 for child in self.info["children"]:
162 self.children.append(child["id"])
165 def get_cmdline(self, child: Optional[str] = None) -> str:
167 f"{config.config['smart_outlets_tplink_location']} -m {self.mac} "
168 f"--no_logging_console "
170 if child is not None:
171 cmd += f"-x {child} "
175 def command(self, cmd: str, extra_args: str = None, **kwargs) -> bool:
176 child: Optional[str] = kwargs.get('child', None)
177 cmd = self.get_cmdline(child) + f"-c {cmd}"
178 if extra_args is not None:
179 cmd += f" {extra_args}"
180 logger.debug(f'About to execute {cmd}')
181 return tplink_outlet_command(cmd)
183 def get_children(self) -> List[str]:
187 def turn_on(self, child: str = None) -> bool:
188 return self.command("on", child)
191 def turn_off(self, child: str = None) -> bool:
192 return self.command("off", child)
194 def get_on_duration_seconds(self, child: str = None) -> int:
195 self.info = self.get_info()
197 if self.info is None:
199 return int(self.info.get("on_time", "0"))
201 if self.info is None:
203 for chi in self.info.get("children", {}):
204 if chi["id"] == child:
205 return int(chi.get("on_time", "0"))
209 class GoogleOutlet(BaseOutlet):
210 def __init__(self, name: str, mac: str, keywords: str = "") -> None:
211 super().__init__(name.strip(), mac.strip(), keywords)
214 def goog_name(self) -> str:
215 name = self.get_name()
216 return name.replace('_', ' ')
219 def parse_google_response(response: GoogleResponse) -> bool:
220 return response.success
223 def turn_on(self) -> bool:
224 return GoogleOutlet.parse_google_response(ask_google(f'turn {self.goog_name()} on'))
227 def turn_off(self) -> bool:
228 return GoogleOutlet.parse_google_response(ask_google(f'turn {self.goog_name()} off'))
231 def is_on(self) -> bool:
232 r = ask_google(f'is {self.goog_name()} on?')
235 if r.audio_transcription is not None:
236 return 'is on' in r.audio_transcription
237 raise Exception('Can\'t talk to Google right now!?')
240 def is_off(self) -> bool:
241 return not self.is_on()
244 @decorator_utils.singleton
245 class MerossWrapper(object):
246 """Global singleton helper class for MerossOutlets. Note that
247 instantiating this class causes HTTP traffic with an external
248 Meross server. Meross blocks customers who hit their servers too
249 aggressively so MerossOutlet is lazy about creating instances of
255 self.loop = asyncio.get_event_loop()
256 self.email = os.environ.get('MEROSS_EMAIL') or scott_secrets.MEROSS_EMAIL
257 self.password = os.environ.get('MEROSS_PASSWORD') or scott_secrets.MEROSS_PASSWORD
258 self.devices = self.loop.run_until_complete(self.find_meross_devices())
259 atexit.register(self.loop.close)
261 async def find_meross_devices(self) -> List[Any]:
262 http_api_client = await MerossHttpClient.async_from_user_password(
263 email=self.email, password=self.password
266 # Setup and start the device manager
267 manager = MerossManager(http_client=http_api_client)
268 await manager.async_init()
271 await manager.async_device_discovery()
272 devices = manager.find_devices()
273 for device in devices:
274 await device.async_update()
277 def get_meross_device_by_name(self, name: str) -> Optional[Any]:
279 name = name.replace('_', ' ')
280 for device in self.devices:
281 if device.name.lower() == name:
286 class MerossOutlet(BaseOutlet):
287 def __init__(self, name: str, mac: str, keywords: str = '') -> None:
288 super().__init__(name, mac, keywords)
289 self.meross_wrapper: Optional[MerossWrapper] = None
290 self.device: Optional[Any] = None
292 def lazy_initialize_device(self):
293 """If we make too many calls to Meross they will block us; only talk
294 to them when someone actually wants to control a device."""
295 if self.meross_wrapper is None:
296 self.meross_wrapper = MerossWrapper()
297 self.device = self.meross_wrapper.get_meross_device_by_name(self.name)
298 if self.device is None:
299 raise Exception(f'{self.name} is not a known Meross device?!')
302 def turn_on(self) -> bool:
303 self.lazy_initialize_device()
304 assert self.meross_wrapper is not None
305 assert self.device is not None
306 self.meross_wrapper.loop.run_until_complete(self.device.async_turn_on())
310 def turn_off(self) -> bool:
311 self.lazy_initialize_device()
312 assert self.meross_wrapper is not None
313 assert self.device is not None
314 self.meross_wrapper.loop.run_until_complete(self.device.async_turn_off())
318 def is_on(self) -> bool:
319 self.lazy_initialize_device()
320 assert self.device is not None
321 return self.device.is_on()
324 def is_off(self) -> bool:
325 return not self.is_on()