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 logger.warning(f'{command} died with signal {signal}')
52 logging_utils.hlog("%s died with signal %d" % (command, signal))
55 exit_value = result >> 8
57 logger.warning(f'{command} failed, exited {exit_value}')
58 logging_utils.hlog("%s failed, exit %d" % (command, exit_value))
60 logger.debug(f'{command} succeeded.')
64 class BaseOutlet(dev.Device):
65 def __init__(self, name: str, mac: str, keywords: str = "") -> None:
66 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) -> 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)
112 def turn_on(self) -> bool:
113 return self.command('on')
115 def turn_off(self) -> bool:
116 return self.command('off')
118 def is_on(self) -> bool:
119 return self.get_on_duration_seconds() > 0
121 def is_off(self) -> bool:
122 return not self.is_on()
125 10.0, use_signals=False, error_message="Timed out waiting for tplink.py"
127 def get_info(self) -> Optional[Dict]:
128 cmd = self.get_cmdline() + "-c info"
129 out = subprocess.getoutput(cmd)
130 out = re.sub("Sent:.*\n", "", out)
131 out = re.sub("Received: *", "", out)
133 self.info = json.loads(out)["system"]["get_sysinfo"]
134 self.info_ts = datetime.datetime.now()
136 except Exception as e:
138 print(out, file=sys.stderr)
143 def get_on_duration_seconds(self) -> int:
144 self.info = self.get_info()
145 if self.info is None:
147 return int(self.info.get("on_time", "0"))
150 class TPLinkOutletWithChildren(TPLinkOutlet):
151 def __init__(self, name: str, mac: str, keywords: str = "") -> None:
152 super().__init__(name, mac, keywords)
153 self.children: List[str] = []
154 self.info: Optional[Dict] = None
155 self.info_ts: Optional[datetime.datetime] = None
156 assert "children" in self.keywords
157 self.info = self.get_info()
158 if self.info is not None:
159 for child in self.info["children"]:
160 self.children.append(child["id"])
163 def get_cmdline(self, child: Optional[str] = None) -> str:
165 f"{config.config['smart_outlets_tplink_location']} -m {self.mac} "
166 f"--no_logging_console "
168 if child is not None:
169 cmd += f"-x {child} "
174 self, cmd: str, child: str = None, extra_args: str = None
176 cmd = self.get_cmdline(child) + f"-c {cmd}"
177 if extra_args is not None:
178 cmd += f" {extra_args}"
179 logger.debug(f'About to execute {cmd}')
180 return tplink_outlet_command(cmd)
182 def get_children(self) -> List[str]:
185 def turn_on(self, child: str = None) -> bool:
186 return self.command("on", child)
188 def turn_off(self, child: str = None) -> bool:
189 return self.command("off", child)
191 def get_on_duration_seconds(self, child: str = None) -> int:
192 self.info = self.get_info()
194 if self.info is None:
196 return int(self.info.get("on_time", "0"))
198 if self.info is None:
200 for chi in self.info.get("children", {}):
201 if chi["id"] == child:
202 return int(chi.get("on_time", "0"))
206 class GoogleOutlet(BaseOutlet):
207 def __init__(self, name: str, mac: str, keywords: str = "") -> None:
208 super().__init__(name.strip(), mac.strip(), keywords)
211 def goog_name(self) -> str:
212 name = self.get_name()
213 return name.replace('_', ' ')
216 def parse_google_response(response: GoogleResponse) -> bool:
217 return response.success
219 def turn_on(self) -> bool:
220 return GoogleOutlet.parse_google_response(
221 ask_google('turn {self.goog_name()} on')
224 def turn_off(self) -> bool:
225 return GoogleOutlet.parse_google_response(
226 ask_google('turn {self.goog_name()} off')
229 def is_on(self) -> bool:
230 r = ask_google(f'is {self.goog_name()} on?')
233 return 'is on' in r.audio_transcription
235 def is_off(self) -> bool:
236 return not self.is_on()
239 @decorator_utils.singleton
240 class MerossWrapper(object):
241 """Note that instantiating this class causes HTTP traffic with an
242 external Meross server. Meross blocks customers who hit their
243 servers too aggressively so MerossOutlet is lazy about creating
244 instances of this class.
249 self.loop = asyncio.get_event_loop()
250 self.email = os.environ.get('MEROSS_EMAIL') or scott_secrets.MEROSS_EMAIL
251 self.password = os.environ.get('MEROSS_PASSWORD') or scott_secrets.MEROSS_PASSWORD
252 self.devices = self.loop.run_until_complete(self.find_meross_devices())
253 atexit.register(self.loop.close)
255 async def find_meross_devices(self) -> List[Any]:
256 http_api_client = await MerossHttpClient.async_from_user_password(
257 email=self.email, password=self.password
260 # Setup and start the device manager
261 manager = MerossManager(http_client=http_api_client)
262 await manager.async_init()
265 await manager.async_device_discovery()
266 devices = manager.find_devices()
267 for device in devices:
268 await device.async_update()
271 def get_meross_device_by_name(self, name: str) -> Optional[Any]:
273 name = name.replace('_', ' ')
274 for device in self.devices:
275 if device.name.lower() == name:
280 class MerossOutlet(BaseOutlet):
281 def __init__(self, name: str, mac: str, keywords: str = '') -> None:
282 super().__init__(name, mac, keywords)
283 self.meross_wrapper = None
286 def lazy_initialize_device(self):
287 """If we make too many calls to Meross they will block us; only talk
288 to them when someone actually wants to control a device."""
289 if self.meross_wrapper is None:
290 self.meross_wrapper = MerossWrapper()
291 self.device = self.meross_wrapper.get_meross_device_by_name(self.name)
292 if self.device is None:
293 raise Exception(f'{self.name} is not a known Meross device?!')
295 def turn_on(self) -> bool:
296 self.lazy_initialize_device()
297 self.meross_wrapper.loop.run_until_complete(
298 self.device.async_turn_on()
302 def turn_off(self) -> bool:
303 self.lazy_initialize_device()
304 self.meross_wrapper.loop.run_until_complete(
305 self.device.async_turn_off()
309 def is_on(self) -> bool:
310 self.lazy_initialize_device()
311 return self.device.is_on()
313 def is_off(self) -> bool:
314 return not self.is_on()