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 self.keywords is not None
159 assert "children" in self.keywords
160 self.info = self.get_info()
161 if self.info is not None:
162 for child in self.info["children"]:
163 self.children.append(child["id"])
166 def get_cmdline(self, child: Optional[str] = None) -> str:
168 f"{config.config['smart_outlets_tplink_location']} -m {self.mac} "
169 f"--no_logging_console "
171 if child is not None:
172 cmd += f"-x {child} "
176 def command(self, cmd: str, extra_args: str = None, **kwargs) -> bool:
177 child: Optional[str] = kwargs.get('child', None)
178 cmd = self.get_cmdline(child) + f"-c {cmd}"
179 if extra_args is not None:
180 cmd += f" {extra_args}"
181 logger.debug(f'About to execute {cmd}')
182 return tplink_outlet_command(cmd)
184 def get_children(self) -> List[str]:
188 def turn_on(self, child: str = None) -> bool:
189 return self.command("on", child)
192 def turn_off(self, child: str = None) -> bool:
193 return self.command("off", child)
195 def get_on_duration_seconds(self, child: str = None) -> int:
196 self.info = self.get_info()
198 if self.info is None:
200 return int(self.info.get("on_time", "0"))
202 if self.info is None:
204 for chi in self.info.get("children", {}):
205 if chi["id"] == child:
206 return int(chi.get("on_time", "0"))
210 class GoogleOutlet(BaseOutlet):
211 def __init__(self, name: str, mac: str, keywords: str = "") -> None:
212 super().__init__(name.strip(), mac.strip(), keywords)
215 def goog_name(self) -> str:
216 name = self.get_name()
217 return name.replace('_', ' ')
220 def parse_google_response(response: GoogleResponse) -> bool:
221 return response.success
224 def turn_on(self) -> bool:
225 return GoogleOutlet.parse_google_response(ask_google(f'turn {self.goog_name()} on'))
228 def turn_off(self) -> bool:
229 return GoogleOutlet.parse_google_response(ask_google(f'turn {self.goog_name()} off'))
232 def is_on(self) -> bool:
233 r = ask_google(f'is {self.goog_name()} on?')
236 if r.audio_transcription is not None:
237 return 'is on' in r.audio_transcription
238 raise Exception('Can\'t talk to Google right now!?')
241 def is_off(self) -> bool:
242 return not self.is_on()
245 @decorator_utils.singleton
246 class MerossWrapper(object):
247 """Global singleton helper class for MerossOutlets. Note that
248 instantiating this class causes HTTP traffic with an external
249 Meross server. Meross blocks customers who hit their servers too
250 aggressively so MerossOutlet is lazy about creating instances of
256 self.loop = asyncio.get_event_loop()
257 self.email = os.environ.get('MEROSS_EMAIL') or scott_secrets.MEROSS_EMAIL
258 self.password = os.environ.get('MEROSS_PASSWORD') or scott_secrets.MEROSS_PASSWORD
259 self.devices = self.loop.run_until_complete(self.find_meross_devices())
260 atexit.register(self.loop.close)
262 async def find_meross_devices(self) -> List[Any]:
263 http_api_client = await MerossHttpClient.async_from_user_password(
264 email=self.email, password=self.password
267 # Setup and start the device manager
268 manager = MerossManager(http_client=http_api_client)
269 await manager.async_init()
272 await manager.async_device_discovery()
273 devices = manager.find_devices()
274 for device in devices:
275 await device.async_update()
278 def get_meross_device_by_name(self, name: str) -> Optional[Any]:
280 name = name.replace('_', ' ')
281 for device in self.devices:
282 if device.name.lower() == name:
287 class MerossOutlet(BaseOutlet):
288 def __init__(self, name: str, mac: str, keywords: str = '') -> None:
289 super().__init__(name, mac, keywords)
290 self.meross_wrapper: Optional[MerossWrapper] = None
291 self.device: Optional[Any] = None
293 def lazy_initialize_device(self):
294 """If we make too many calls to Meross they will block us; only talk
295 to them when someone actually wants to control a device."""
296 if self.meross_wrapper is None:
297 self.meross_wrapper = MerossWrapper()
298 self.device = self.meross_wrapper.get_meross_device_by_name(self.name)
299 if self.device is None:
300 raise Exception(f'{self.name} is not a known Meross device?!')
303 def turn_on(self) -> bool:
304 self.lazy_initialize_device()
305 assert self.meross_wrapper is not None
306 assert self.device is not None
307 self.meross_wrapper.loop.run_until_complete(self.device.async_turn_on())
311 def turn_off(self) -> bool:
312 self.lazy_initialize_device()
313 assert self.meross_wrapper is not None
314 assert self.device is not None
315 self.meross_wrapper.loop.run_until_complete(self.device.async_turn_off())
319 def is_on(self) -> bool:
320 self.lazy_initialize_device()
321 assert self.device is not None
322 return self.device.is_on()
325 def is_off(self) -> bool:
326 return not self.is_on()