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(
225 ask_google(f'turn {self.goog_name()} on')
229 def turn_off(self) -> bool:
230 return GoogleOutlet.parse_google_response(
231 ask_google(f'turn {self.goog_name()} off')
235 def is_on(self) -> bool:
236 r = ask_google(f'is {self.goog_name()} on?')
239 if r.audio_transcription is not None:
240 return 'is on' in r.audio_transcription
241 raise Exception('Can\'t talk to Google right now!?')
244 def is_off(self) -> bool:
245 return not self.is_on()
248 @decorator_utils.singleton
249 class MerossWrapper(object):
250 """Global singleton helper class for MerossOutlets. Note that
251 instantiating this class causes HTTP traffic with an external
252 Meross server. Meross blocks customers who hit their servers too
253 aggressively so MerossOutlet is lazy about creating instances of
259 self.loop = asyncio.get_event_loop()
260 self.email = os.environ.get('MEROSS_EMAIL') or scott_secrets.MEROSS_EMAIL
262 os.environ.get('MEROSS_PASSWORD') or scott_secrets.MEROSS_PASSWORD
264 self.devices = self.loop.run_until_complete(self.find_meross_devices())
265 atexit.register(self.loop.close)
267 async def find_meross_devices(self) -> List[Any]:
268 http_api_client = await MerossHttpClient.async_from_user_password(
269 email=self.email, password=self.password
272 # Setup and start the device manager
273 manager = MerossManager(http_client=http_api_client)
274 await manager.async_init()
277 await manager.async_device_discovery()
278 devices = manager.find_devices()
279 for device in devices:
280 await device.async_update()
283 def get_meross_device_by_name(self, name: str) -> Optional[Any]:
285 name = name.replace('_', ' ')
286 for device in self.devices:
287 if device.name.lower() == name:
292 class MerossOutlet(BaseOutlet):
293 def __init__(self, name: str, mac: str, keywords: str = '') -> None:
294 super().__init__(name, mac, keywords)
295 self.meross_wrapper: Optional[MerossWrapper] = None
296 self.device: Optional[Any] = None
298 def lazy_initialize_device(self):
299 """If we make too many calls to Meross they will block us; only talk
300 to them when someone actually wants to control a device."""
301 if self.meross_wrapper is None:
302 self.meross_wrapper = MerossWrapper()
303 self.device = self.meross_wrapper.get_meross_device_by_name(self.name)
304 if self.device is None:
305 raise Exception(f'{self.name} is not a known Meross device?!')
308 def turn_on(self) -> bool:
309 self.lazy_initialize_device()
310 assert self.meross_wrapper
312 self.meross_wrapper.loop.run_until_complete(self.device.async_turn_on())
316 def turn_off(self) -> bool:
317 self.lazy_initialize_device()
318 assert self.meross_wrapper
320 self.meross_wrapper.loop.run_until_complete(self.device.async_turn_off())
324 def is_on(self) -> bool:
325 self.lazy_initialize_device()
327 return self.device.is_on()
330 def is_off(self) -> bool:
331 return not self.is_on()