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(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]:
109 cmd = self.get_cmdline() + "-c info"
110 self.info = tplink.tplink_get_info(cmd)
111 if self.info is not None:
112 self.info_ts = datetime.datetime.now()
117 def get_on_duration_seconds(self) -> int:
118 self.info = self.get_info()
119 if self.info is None:
121 return int(self.info.get("on_time", "0"))
124 class TPLinkOutletWithChildren(TPLinkOutlet):
125 """A TPLink outlet where the top and bottom plus are individually
128 def __init__(self, name: str, mac: str, keywords: str = "") -> None:
129 super().__init__(name, mac, keywords)
130 self.children: List[str] = []
131 self.info: Optional[Dict] = None
132 self.info_ts: Optional[datetime.datetime] = None
133 assert self.keywords is not None
134 assert "children" in self.keywords
135 self.info = self.get_info()
136 if self.info is not None:
137 for child in self.info["children"]:
138 self.children.append(child["id"])
140 def get_cmdline_with_child(self, child: Optional[str] = None) -> str:
141 cmd = super().get_cmdline()
142 if child is not None:
143 cmd += f"-x {child} "
147 def command(self, cmd: str, extra_args: str = None, **kwargs) -> bool:
148 child: Optional[str] = kwargs.get('child', None)
149 cmd = self.get_cmdline_with_child(child) + f"-c {cmd}"
150 if extra_args is not None:
151 cmd += f" {extra_args}"
152 logger.debug('About to execute: %s', cmd)
153 return tplink.tplink_command(cmd)
155 def get_children(self) -> List[str]:
159 def turn_on(self) -> bool:
160 return self.command("on", None)
163 def turn_off(self) -> bool:
164 return self.command("off", None)
166 def turn_on_child(self, child: str = None) -> bool:
167 return self.command("on", child)
169 def turn_off_child(self, child: str = None) -> bool:
170 return self.command("off", child)
172 def get_child_on_duration_seconds(self, child: str = None) -> int:
174 return super().get_on_duration_seconds()
176 self.info = self.get_info()
177 if self.info is None:
179 for chi in self.info.get("children", {}):
180 if chi["id"] == child:
181 return int(chi.get("on_time", "0"))
185 class GoogleOutlet(BaseOutlet):
186 """A smart outlet controlled via Google Assistant."""
188 def __init__(self, name: str, mac: str, keywords: str = "") -> None:
189 super().__init__(name.strip(), mac.strip(), keywords)
192 def goog_name(self) -> str:
193 name = self.get_name()
194 return name.replace('_', ' ')
197 def parse_google_response(response: GoogleResponse) -> bool:
198 return response.success
201 def turn_on(self) -> bool:
202 return GoogleOutlet.parse_google_response(ask_google(f'turn {self.goog_name()} on'))
205 def turn_off(self) -> bool:
206 return GoogleOutlet.parse_google_response(ask_google(f'turn {self.goog_name()} off'))
209 def is_on(self) -> bool:
210 r = ask_google(f'is {self.goog_name()} on?')
213 if r.audio_transcription is not None:
214 return 'is on' in r.audio_transcription
215 raise Exception('Can\'t talk to Google right now!?')
218 def is_off(self) -> bool:
219 return not self.is_on()
222 @decorator_utils.singleton
223 class MerossWrapper(object):
224 """Global singleton helper class for MerossOutlets. Note that
225 instantiating this class causes HTTP traffic with an external
226 Meross server. Meross blocks customers who hit their servers too
227 aggressively so MerossOutlet is lazy about creating instances of
233 self.loop = asyncio.get_event_loop()
234 self.email = os.environ.get('MEROSS_EMAIL') or scott_secrets.MEROSS_EMAIL
235 self.password = os.environ.get('MEROSS_PASSWORD') or scott_secrets.MEROSS_PASSWORD
236 self.devices = self.loop.run_until_complete(self.find_meross_devices())
237 atexit.register(self.loop.close)
239 async def find_meross_devices(self) -> List[Any]:
240 http_api_client = await MerossHttpClient.async_from_user_password(
241 email=self.email, password=self.password
244 # Setup and start the device manager
245 manager = MerossManager(http_client=http_api_client)
246 await manager.async_init()
249 await manager.async_device_discovery()
250 devices = manager.find_devices()
251 for device in devices:
252 await device.async_update()
255 def get_meross_device_by_name(self, name: str) -> Optional[Any]:
257 name = name.replace('_', ' ')
258 for device in self.devices:
259 if device.name.lower() == name:
264 class MerossOutlet(BaseOutlet):
265 """A Meross smart outlet class."""
267 def __init__(self, name: str, mac: str, keywords: str = '') -> None:
268 super().__init__(name, mac, keywords)
269 self.meross_wrapper: Optional[MerossWrapper] = None
270 self.device: Optional[Any] = None
272 def lazy_initialize_device(self):
273 """If we make too many calls to Meross they will block us; only talk
274 to them when someone actually wants to control a device."""
275 if self.meross_wrapper is None:
276 self.meross_wrapper = MerossWrapper()
277 self.device = self.meross_wrapper.get_meross_device_by_name(self.name)
278 if self.device is None:
279 raise Exception(f'{self.name} is not a known Meross device?!')
282 def turn_on(self) -> bool:
283 self.lazy_initialize_device()
284 assert self.meross_wrapper is not None
285 assert self.device is not None
286 self.meross_wrapper.loop.run_until_complete(self.device.async_turn_on())
290 def turn_off(self) -> bool:
291 self.lazy_initialize_device()
292 assert self.meross_wrapper is not None
293 assert self.device is not None
294 self.meross_wrapper.loop.run_until_complete(self.device.async_turn_off())
298 def is_on(self) -> bool:
299 self.lazy_initialize_device()
300 assert self.device is not None
301 return self.device.is_on()
304 def is_off(self) -> bool:
305 return not self.is_on()