3 # © Copyright 2021-2022, Scott Gasch
5 """Utilities for dealing with the smart outlets."""
12 from abc import abstractmethod
13 from typing import Any, Dict, List, Optional
15 from meross_iot.http_api import MerossHttpClient
16 from meross_iot.manager import MerossManager
17 from overrides import overrides
21 import decorator_utils
23 import smart_home.device as dev
24 import smart_home.tplink_utils as tplink
25 from decorator_utils import memoized
26 from google_assistant import GoogleResponse, ask_google
28 logger = logging.getLogger(__name__)
30 parser = config.add_commandline_args(
31 f"Outlet Utils ({__file__})",
32 "Args related to smart outlets.",
35 '--smart_outlets_tplink_location',
36 default='/home/scott/bin/tplink.py',
38 help='The location of the tplink.py helper',
39 type=argparse_utils.valid_filename,
43 class BaseOutlet(dev.Device):
44 """An abstract base class for smart outlets."""
46 def __init__(self, name: str, mac: str, keywords: str = "") -> None:
47 super().__init__(name.strip(), mac.strip(), keywords)
50 def turn_on(self) -> bool:
54 def turn_off(self) -> bool:
58 def is_on(self) -> bool:
62 def is_off(self) -> bool:
66 class TPLinkOutlet(BaseOutlet):
67 """A TPLink smart outlet."""
69 def __init__(self, name: str, mac: str, keywords: str = '') -> None:
70 super().__init__(name, mac, keywords)
71 self.info: Optional[Dict] = None
72 self.info_ts: Optional[datetime.datetime] = None
75 def get_tplink_name(self) -> Optional[str]:
76 self.info = self.get_info()
77 if self.info is not None:
78 return self.info["alias"]
81 def get_cmdline(self) -> str:
83 f"{config.config['smart_outlets_tplink_location']} -m {self.mac} "
84 f"--no_logging_console "
88 def command(self, cmd: str, extra_args: str = None, **kwargs) -> bool:
89 cmd = self.get_cmdline() + f"-c {cmd}"
90 if extra_args is not None:
91 cmd += f" {extra_args}"
92 return tplink.tplink_command_wrapper(cmd)
95 def turn_on(self) -> bool:
96 return self.command('on')
99 def turn_off(self) -> bool:
100 return self.command('off')
103 def is_on(self) -> bool:
104 return self.get_on_duration_seconds() > 0
107 def is_off(self) -> bool:
108 return not self.is_on()
110 def get_info(self) -> Optional[Dict]:
113 self.info = tplink.tplink_get_info(ip)
114 if self.info is not None:
115 self.info_ts = datetime.datetime.now()
121 def get_on_duration_seconds(self) -> int:
122 self.info = self.get_info()
123 if self.info is None:
125 return int(self.info.get("on_time", "0"))
128 class TPLinkOutletWithChildren(TPLinkOutlet):
129 """A TPLink outlet where the top and bottom plus are individually
132 def __init__(self, name: str, mac: str, keywords: str = "") -> None:
133 super().__init__(name, mac, keywords)
134 self.children: List[str] = []
135 self.info: Optional[Dict] = None
136 self.info_ts: Optional[datetime.datetime] = None
137 assert self.keywords is not None
138 assert "children" in self.keywords
139 self.info = self.get_info()
140 if self.info is not None:
141 for child in self.info["children"]:
142 self.children.append(child["id"])
144 def get_cmdline_with_child(self, child: Optional[str] = None) -> str:
145 cmd = super().get_cmdline()
146 if child is not None:
147 cmd += f"-x {child} "
151 def command(self, cmd: str, extra_args: str = None, **kwargs) -> bool:
152 child: Optional[str] = kwargs.get('child', None)
153 cmd = self.get_cmdline_with_child(child) + f"-c {cmd}"
154 if extra_args is not None:
155 cmd += f" {extra_args}"
156 logger.debug('About to execute: %s', cmd)
157 return tplink.tplink_command_wrapper(cmd)
159 def get_children(self) -> List[str]:
163 def turn_on(self) -> bool:
164 return self.command("on", None)
167 def turn_off(self) -> bool:
168 return self.command("off", None)
170 def turn_on_child(self, child: str = None) -> bool:
171 return self.command("on", child)
173 def turn_off_child(self, child: str = None) -> bool:
174 return self.command("off", child)
176 def get_child_on_duration_seconds(self, child: str = None) -> int:
178 return super().get_on_duration_seconds()
180 self.info = self.get_info()
181 if self.info is None:
183 for chi in self.info.get("children", {}):
184 if chi["id"] == child:
185 return int(chi.get("on_time", "0"))
189 def get_on_duration_seconds(self) -> int:
190 self.info = self.get_info()
191 if self.info is not None:
192 for child in self.info["children"]:
193 if int(child["on_time"]) > 0:
198 class GoogleOutlet(BaseOutlet):
199 """A smart outlet controlled via Google Assistant."""
201 def __init__(self, name: str, mac: str, keywords: str = "") -> None:
202 super().__init__(name.strip(), mac.strip(), keywords)
205 def goog_name(self) -> str:
206 name = self.get_name()
207 return name.replace('_', ' ')
210 def parse_google_response(response: GoogleResponse) -> bool:
211 return response.success
214 def turn_on(self) -> bool:
215 return GoogleOutlet.parse_google_response(ask_google(f'turn {self.goog_name()} on'))
218 def turn_off(self) -> bool:
219 return GoogleOutlet.parse_google_response(ask_google(f'turn {self.goog_name()} off'))
222 def is_on(self) -> bool:
223 r = ask_google(f'is {self.goog_name()} on?')
226 if r.audio_transcription is not None:
227 return 'is on' in r.audio_transcription
228 raise Exception('Can\'t talk to Google right now!?')
231 def is_off(self) -> bool:
232 return not self.is_on()
235 @decorator_utils.singleton
236 class MerossWrapper(object):
237 """Global singleton helper class for MerossOutlets. Note that
238 instantiating this class causes HTTP traffic with an external
239 Meross server. Meross blocks customers who hit their servers too
240 aggressively so MerossOutlet is lazy about creating instances of
246 self.loop = asyncio.get_event_loop()
247 self.email = os.environ.get('MEROSS_EMAIL') or scott_secrets.MEROSS_EMAIL
248 self.password = os.environ.get('MEROSS_PASSWORD') or scott_secrets.MEROSS_PASSWORD
249 self.devices = self.loop.run_until_complete(self.find_meross_devices())
250 atexit.register(self.loop.close)
252 async def find_meross_devices(self) -> List[Any]:
253 http_api_client = await MerossHttpClient.async_from_user_password(
254 email=self.email, password=self.password
257 # Setup and start the device manager
258 manager = MerossManager(http_client=http_api_client)
259 await manager.async_init()
262 await manager.async_device_discovery()
263 devices = manager.find_devices()
264 for device in devices:
265 await device.async_update()
268 def get_meross_device_by_name(self, name: str) -> Optional[Any]:
270 name = name.replace('_', ' ')
271 for device in self.devices:
272 if device.name.lower() == name:
277 class MerossOutlet(BaseOutlet):
278 """A Meross smart outlet class."""
280 def __init__(self, name: str, mac: str, keywords: str = '') -> None:
281 super().__init__(name, mac, keywords)
282 self.meross_wrapper: Optional[MerossWrapper] = None
283 self.device: Optional[Any] = None
285 def lazy_initialize_device(self):
286 """If we make too many calls to Meross they will block us; only talk
287 to them when someone actually wants to control a device."""
288 if self.meross_wrapper is None:
289 self.meross_wrapper = MerossWrapper()
290 self.device = self.meross_wrapper.get_meross_device_by_name(self.name)
291 if self.device is None:
292 raise Exception(f'{self.name} is not a known Meross device?!')
295 def turn_on(self) -> bool:
296 self.lazy_initialize_device()
297 assert self.meross_wrapper is not None
298 assert self.device is not None
299 self.meross_wrapper.loop.run_until_complete(self.device.async_turn_on())
303 def turn_off(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_off())
311 def is_on(self) -> bool:
312 self.lazy_initialize_device()
313 assert self.device is not None
314 return self.device.is_on()
317 def is_off(self) -> bool:
318 return not self.is_on()