3 """Utilities for dealing with the smart outlets."""
10 from abc import abstractmethod
11 from typing import Any, Dict, List, Optional
13 from meross_iot.http_api import MerossHttpClient
14 from meross_iot.manager import MerossManager
15 from overrides import overrides
19 import decorator_utils
21 import smart_home.device as dev
22 import smart_home.tplink_utils as tplink
23 from decorator_utils import memoized
24 from google_assistant import GoogleResponse, ask_google
26 logger = logging.getLogger(__name__)
28 parser = config.add_commandline_args(
29 f"Outlet Utils ({__file__})",
30 "Args related to smart outlets.",
33 '--smart_outlets_tplink_location',
34 default='/home/scott/bin/tplink.py',
36 help='The location of the tplink.py helper',
37 type=argparse_utils.valid_filename,
41 class BaseOutlet(dev.Device):
42 """An abstract base class for smart outlets."""
44 def __init__(self, name: str, mac: str, keywords: str = "") -> None:
45 super().__init__(name.strip(), mac.strip(), keywords)
48 def turn_on(self) -> bool:
52 def turn_off(self) -> bool:
56 def is_on(self) -> bool:
60 def is_off(self) -> bool:
64 class TPLinkOutlet(BaseOutlet):
65 """A TPLink smart outlet."""
67 def __init__(self, name: str, mac: str, keywords: str = '') -> None:
68 super().__init__(name, mac, keywords)
69 self.info: Optional[Dict] = None
70 self.info_ts: Optional[datetime.datetime] = None
73 def get_tplink_name(self) -> Optional[str]:
74 self.info = self.get_info()
75 if self.info is not None:
76 return self.info["alias"]
79 def get_cmdline(self) -> str:
81 f"{config.config['smart_outlets_tplink_location']} -m {self.mac} "
82 f"--no_logging_console "
86 def command(self, cmd: str, extra_args: str = None, **kwargs) -> bool:
87 cmd = self.get_cmdline() + f"-c {cmd}"
88 if extra_args is not None:
89 cmd += f" {extra_args}"
90 return tplink.tplink_command_wrapper(cmd)
93 def turn_on(self) -> bool:
94 return self.command('on')
97 def turn_off(self) -> bool:
98 return self.command('off')
101 def is_on(self) -> bool:
102 return self.get_on_duration_seconds() > 0
105 def is_off(self) -> bool:
106 return not self.is_on()
108 def get_info(self) -> Optional[Dict]:
111 self.info = tplink.tplink_get_info(ip)
112 if self.info is not None:
113 self.info_ts = datetime.datetime.now()
119 def get_on_duration_seconds(self) -> int:
120 self.info = self.get_info()
121 if self.info is None:
123 return int(self.info.get("on_time", "0"))
126 class TPLinkOutletWithChildren(TPLinkOutlet):
127 """A TPLink outlet where the top and bottom plus are individually
130 def __init__(self, name: str, mac: str, keywords: str = "") -> None:
131 super().__init__(name, mac, keywords)
132 self.children: List[str] = []
133 self.info: Optional[Dict] = None
134 self.info_ts: Optional[datetime.datetime] = None
135 assert self.keywords is not None
136 assert "children" in self.keywords
137 self.info = self.get_info()
138 if self.info is not None:
139 for child in self.info["children"]:
140 self.children.append(child["id"])
142 def get_cmdline_with_child(self, child: Optional[str] = None) -> str:
143 cmd = super().get_cmdline()
144 if child is not None:
145 cmd += f"-x {child} "
149 def command(self, cmd: str, extra_args: str = None, **kwargs) -> bool:
150 child: Optional[str] = kwargs.get('child', None)
151 cmd = self.get_cmdline_with_child(child) + f"-c {cmd}"
152 if extra_args is not None:
153 cmd += f" {extra_args}"
154 logger.debug('About to execute: %s', cmd)
155 return tplink.tplink_command_wrapper(cmd)
157 def get_children(self) -> List[str]:
161 def turn_on(self) -> bool:
162 return self.command("on", None)
165 def turn_off(self) -> bool:
166 return self.command("off", None)
168 def turn_on_child(self, child: str = None) -> bool:
169 return self.command("on", child)
171 def turn_off_child(self, child: str = None) -> bool:
172 return self.command("off", child)
174 def get_child_on_duration_seconds(self, child: str = None) -> int:
176 return super().get_on_duration_seconds()
178 self.info = self.get_info()
179 if self.info is None:
181 for chi in self.info.get("children", {}):
182 if chi["id"] == child:
183 return int(chi.get("on_time", "0"))
187 class GoogleOutlet(BaseOutlet):
188 """A smart outlet controlled via Google Assistant."""
190 def __init__(self, name: str, mac: str, keywords: str = "") -> None:
191 super().__init__(name.strip(), mac.strip(), keywords)
194 def goog_name(self) -> str:
195 name = self.get_name()
196 return name.replace('_', ' ')
199 def parse_google_response(response: GoogleResponse) -> bool:
200 return response.success
203 def turn_on(self) -> bool:
204 return GoogleOutlet.parse_google_response(ask_google(f'turn {self.goog_name()} on'))
207 def turn_off(self) -> bool:
208 return GoogleOutlet.parse_google_response(ask_google(f'turn {self.goog_name()} off'))
211 def is_on(self) -> bool:
212 r = ask_google(f'is {self.goog_name()} on?')
215 if r.audio_transcription is not None:
216 return 'is on' in r.audio_transcription
217 raise Exception('Can\'t talk to Google right now!?')
220 def is_off(self) -> bool:
221 return not self.is_on()
224 @decorator_utils.singleton
225 class MerossWrapper(object):
226 """Global singleton helper class for MerossOutlets. Note that
227 instantiating this class causes HTTP traffic with an external
228 Meross server. Meross blocks customers who hit their servers too
229 aggressively so MerossOutlet is lazy about creating instances of
235 self.loop = asyncio.get_event_loop()
236 self.email = os.environ.get('MEROSS_EMAIL') or scott_secrets.MEROSS_EMAIL
237 self.password = os.environ.get('MEROSS_PASSWORD') or scott_secrets.MEROSS_PASSWORD
238 self.devices = self.loop.run_until_complete(self.find_meross_devices())
239 atexit.register(self.loop.close)
241 async def find_meross_devices(self) -> List[Any]:
242 http_api_client = await MerossHttpClient.async_from_user_password(
243 email=self.email, password=self.password
246 # Setup and start the device manager
247 manager = MerossManager(http_client=http_api_client)
248 await manager.async_init()
251 await manager.async_device_discovery()
252 devices = manager.find_devices()
253 for device in devices:
254 await device.async_update()
257 def get_meross_device_by_name(self, name: str) -> Optional[Any]:
259 name = name.replace('_', ' ')
260 for device in self.devices:
261 if device.name.lower() == name:
266 class MerossOutlet(BaseOutlet):
267 """A Meross smart outlet class."""
269 def __init__(self, name: str, mac: str, keywords: str = '') -> None:
270 super().__init__(name, mac, keywords)
271 self.meross_wrapper: Optional[MerossWrapper] = None
272 self.device: Optional[Any] = None
274 def lazy_initialize_device(self):
275 """If we make too many calls to Meross they will block us; only talk
276 to them when someone actually wants to control a device."""
277 if self.meross_wrapper is None:
278 self.meross_wrapper = MerossWrapper()
279 self.device = self.meross_wrapper.get_meross_device_by_name(self.name)
280 if self.device is None:
281 raise Exception(f'{self.name} is not a known Meross device?!')
284 def turn_on(self) -> bool:
285 self.lazy_initialize_device()
286 assert self.meross_wrapper is not None
287 assert self.device is not None
288 self.meross_wrapper.loop.run_until_complete(self.device.async_turn_on())
292 def turn_off(self) -> bool:
293 self.lazy_initialize_device()
294 assert self.meross_wrapper is not None
295 assert self.device is not None
296 self.meross_wrapper.loop.run_until_complete(self.device.async_turn_off())
300 def is_on(self) -> bool:
301 self.lazy_initialize_device()
302 assert self.device is not None
303 return self.device.is_on()
306 def is_off(self) -> bool:
307 return not self.is_on()