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 class GoogleOutlet(BaseOutlet):
190 """A smart outlet controlled via Google Assistant."""
192 def __init__(self, name: str, mac: str, keywords: str = "") -> None:
193 super().__init__(name.strip(), mac.strip(), keywords)
196 def goog_name(self) -> str:
197 name = self.get_name()
198 return name.replace('_', ' ')
201 def parse_google_response(response: GoogleResponse) -> bool:
202 return response.success
205 def turn_on(self) -> bool:
206 return GoogleOutlet.parse_google_response(ask_google(f'turn {self.goog_name()} on'))
209 def turn_off(self) -> bool:
210 return GoogleOutlet.parse_google_response(ask_google(f'turn {self.goog_name()} off'))
213 def is_on(self) -> bool:
214 r = ask_google(f'is {self.goog_name()} on?')
217 if r.audio_transcription is not None:
218 return 'is on' in r.audio_transcription
219 raise Exception('Can\'t talk to Google right now!?')
222 def is_off(self) -> bool:
223 return not self.is_on()
226 @decorator_utils.singleton
227 class MerossWrapper(object):
228 """Global singleton helper class for MerossOutlets. Note that
229 instantiating this class causes HTTP traffic with an external
230 Meross server. Meross blocks customers who hit their servers too
231 aggressively so MerossOutlet is lazy about creating instances of
237 self.loop = asyncio.get_event_loop()
238 self.email = os.environ.get('MEROSS_EMAIL') or scott_secrets.MEROSS_EMAIL
239 self.password = os.environ.get('MEROSS_PASSWORD') or scott_secrets.MEROSS_PASSWORD
240 self.devices = self.loop.run_until_complete(self.find_meross_devices())
241 atexit.register(self.loop.close)
243 async def find_meross_devices(self) -> List[Any]:
244 http_api_client = await MerossHttpClient.async_from_user_password(
245 email=self.email, password=self.password
248 # Setup and start the device manager
249 manager = MerossManager(http_client=http_api_client)
250 await manager.async_init()
253 await manager.async_device_discovery()
254 devices = manager.find_devices()
255 for device in devices:
256 await device.async_update()
259 def get_meross_device_by_name(self, name: str) -> Optional[Any]:
261 name = name.replace('_', ' ')
262 for device in self.devices:
263 if device.name.lower() == name:
268 class MerossOutlet(BaseOutlet):
269 """A Meross smart outlet class."""
271 def __init__(self, name: str, mac: str, keywords: str = '') -> None:
272 super().__init__(name, mac, keywords)
273 self.meross_wrapper: Optional[MerossWrapper] = None
274 self.device: Optional[Any] = None
276 def lazy_initialize_device(self):
277 """If we make too many calls to Meross they will block us; only talk
278 to them when someone actually wants to control a device."""
279 if self.meross_wrapper is None:
280 self.meross_wrapper = MerossWrapper()
281 self.device = self.meross_wrapper.get_meross_device_by_name(self.name)
282 if self.device is None:
283 raise Exception(f'{self.name} is not a known Meross device?!')
286 def turn_on(self) -> bool:
287 self.lazy_initialize_device()
288 assert self.meross_wrapper is not None
289 assert self.device is not None
290 self.meross_wrapper.loop.run_until_complete(self.device.async_turn_on())
294 def turn_off(self) -> bool:
295 self.lazy_initialize_device()
296 assert self.meross_wrapper is not None
297 assert self.device is not None
298 self.meross_wrapper.loop.run_until_complete(self.device.async_turn_off())
302 def is_on(self) -> bool:
303 self.lazy_initialize_device()
304 assert self.device is not None
305 return self.device.is_on()
308 def is_off(self) -> bool:
309 return not self.is_on()