3 """Utilities for dealing with the smart outlets."""
5 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
22 import decorator_utils
25 import smart_home.device as dev
26 from google_assistant import ask_google, GoogleResponse
27 from decorator_utils import timeout, memoized
29 logger = logging.getLogger(__name__)
31 parser = config.add_commandline_args(
32 f"Outlet Utils ({__file__})",
33 "Args related to smart outlets.",
36 '--smart_outlets_tplink_location',
37 default='/home/scott/bin/tplink.py',
39 help='The location of the tplink.py helper',
40 type=argparse_utils.valid_filename,
45 5.0, use_signals=False, error_message="Timed out waiting for tplink.py"
47 def tplink_outlet_command(command: str) -> bool:
48 result = os.system(command)
49 signal = result & 0xFF
51 msg = f'{command} died with signal {signal}'
53 logging_utils.hlog(msg)
56 exit_value = result >> 8
58 msg = f'{command} failed, exited {exit_value}'
60 logging_utils.hlog(msg)
62 logger.debug(f'{command} succeeded.')
66 class BaseOutlet(dev.Device):
67 def __init__(self, name: str, mac: str, keywords: str = "") -> None:
68 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 def __init__(self, name: str, mac: str, keywords: str = '') -> None:
90 super().__init__(name, mac, keywords)
91 self.info: Optional[Dict] = None
92 self.info_ts: Optional[datetime.datetime] = None
95 def get_tplink_name(self) -> Optional[str]:
96 self.info = self.get_info()
97 if self.info is not None:
98 return self.info["alias"]
101 def get_cmdline(self) -> str:
103 f"{config.config['smart_outlets_tplink_location']} -m {self.mac} "
104 f"--no_logging_console "
108 def command(self, cmd: str, extra_args: str = None) -> bool:
109 cmd = self.get_cmdline() + f"-c {cmd}"
110 if extra_args is not None:
111 cmd += f" {extra_args}"
112 return tplink_outlet_command(cmd)
114 def turn_on(self) -> bool:
115 return self.command('on')
117 def turn_off(self) -> bool:
118 return self.command('off')
120 def is_on(self) -> bool:
121 return self.get_on_duration_seconds() > 0
123 def is_off(self) -> bool:
124 return not self.is_on()
127 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} "
176 self, cmd: str, child: str = None, extra_args: str = 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]:
187 def turn_on(self, child: str = None) -> bool:
188 return self.command("on", child)
190 def turn_off(self, child: str = None) -> bool:
191 return self.command("off", child)
193 def get_on_duration_seconds(self, child: str = None) -> int:
194 self.info = self.get_info()
196 if self.info is None:
198 return int(self.info.get("on_time", "0"))
200 if self.info is None:
202 for chi in self.info.get("children", {}):
203 if chi["id"] == child:
204 return int(chi.get("on_time", "0"))
208 class GoogleOutlet(BaseOutlet):
209 def __init__(self, name: str, mac: str, keywords: str = "") -> None:
210 super().__init__(name.strip(), mac.strip(), keywords)
213 def goog_name(self) -> str:
214 name = self.get_name()
215 return name.replace('_', ' ')
218 def parse_google_response(response: GoogleResponse) -> bool:
219 return response.success
221 def turn_on(self) -> bool:
222 return GoogleOutlet.parse_google_response(
223 ask_google('turn {self.goog_name()} on')
226 def turn_off(self) -> bool:
227 return GoogleOutlet.parse_google_response(
228 ask_google('turn {self.goog_name()} off')
231 def is_on(self) -> bool:
232 r = ask_google(f'is {self.goog_name()} on?')
235 return 'is on' in r.audio_transcription
237 def is_off(self) -> bool:
238 return not self.is_on()
241 @decorator_utils.singleton
242 class MerossWrapper(object):
243 """Global singleton helper class for MerossOutlets. Note that
244 instantiating this class causes HTTP traffic with an external
245 Meross server. Meross blocks customers who hit their servers too
246 aggressively so MerossOutlet is lazy about creating instances of
251 self.loop = asyncio.get_event_loop()
252 self.email = os.environ.get('MEROSS_EMAIL') or scott_secrets.MEROSS_EMAIL
253 self.password = os.environ.get('MEROSS_PASSWORD') or scott_secrets.MEROSS_PASSWORD
254 self.devices = self.loop.run_until_complete(self.find_meross_devices())
255 atexit.register(self.loop.close)
257 async def find_meross_devices(self) -> List[Any]:
258 http_api_client = await MerossHttpClient.async_from_user_password(
259 email=self.email, password=self.password
262 # Setup and start the device manager
263 manager = MerossManager(http_client=http_api_client)
264 await manager.async_init()
267 await manager.async_device_discovery()
268 devices = manager.find_devices()
269 for device in devices:
270 await device.async_update()
273 def get_meross_device_by_name(self, name: str) -> Optional[Any]:
275 name = name.replace('_', ' ')
276 for device in self.devices:
277 if device.name.lower() == name:
282 class MerossOutlet(BaseOutlet):
283 def __init__(self, name: str, mac: str, keywords: str = '') -> None:
284 super().__init__(name, mac, keywords)
285 self.meross_wrapper = None
288 def lazy_initialize_device(self):
289 """If we make too many calls to Meross they will block us; only talk
290 to them when someone actually wants to control a device."""
291 if self.meross_wrapper is None:
292 self.meross_wrapper = MerossWrapper()
293 self.device = self.meross_wrapper.get_meross_device_by_name(self.name)
294 if self.device is None:
295 raise Exception(f'{self.name} is not a known Meross device?!')
297 def turn_on(self) -> bool:
298 self.lazy_initialize_device()
299 self.meross_wrapper.loop.run_until_complete(
300 self.device.async_turn_on()
304 def turn_off(self) -> bool:
305 self.lazy_initialize_device()
306 self.meross_wrapper.loop.run_until_complete(
307 self.device.async_turn_off()
311 def is_on(self) -> bool:
312 self.lazy_initialize_device()
313 return self.device.is_on()
315 def is_off(self) -> bool:
316 return not self.is_on()