3 """Utilities for dealing with the smart outlets."""
5 from abc import abstractmethod
13 from typing import Dict, List, Optional
18 import smart_home.device as dev
19 from google_assistant import ask_google, GoogleResponse
20 from decorator_utils import timeout, memoized
22 logger = logging.getLogger(__name__)
24 parser = config.add_commandline_args(
25 f"Outlet Utils ({__file__})",
26 "Args related to smart outlets.",
29 '--smart_outlets_tplink_location',
30 default='/home/scott/bin/tplink.py',
32 help='The location of the tplink.py helper',
33 type=argparse_utils.valid_filename,
38 5.0, use_signals=False, error_message="Timed out waiting for tplink.py"
40 def tplink_outlet_command(command: str) -> bool:
41 result = os.system(command)
42 signal = result & 0xFF
44 logger.warning(f'{command} died with signal {signal}')
45 logging_utils.hlog("%s died with signal %d" % (command, signal))
48 exit_value = result >> 8
50 logger.warning(f'{command} failed, exited {exit_value}')
51 logging_utils.hlog("%s failed, exit %d" % (command, exit_value))
53 logger.debug(f'{command} succeeded.')
57 class BaseOutlet(dev.Device):
58 def __init__(self, name: str, mac: str, keywords: str = "") -> None:
59 super().__init__(name.strip(), mac.strip(), keywords)
63 def turn_on(self) -> bool:
67 def turn_off(self) -> bool:
71 def is_on(self) -> bool:
75 def is_off(self) -> bool:
79 class TPLinkOutlet(BaseOutlet):
80 def __init__(self, name: str, mac: str, keywords: str = '') -> None:
81 super().__init__(name, mac, keywords)
82 self.info: Optional[Dict] = None
83 self.info_ts: Optional[datetime.datetime] = None
86 def get_tplink_name(self) -> Optional[str]:
87 self.info = self.get_info()
88 if self.info is not None:
89 return self.info["alias"]
92 def get_cmdline(self) -> str:
94 f"{config.config['smart_outlets_tplink_location']} -m {self.mac} "
95 f"--no_logging_console "
99 def command(self, cmd: str, extra_args: str = None) -> bool:
100 cmd = self.get_cmdline() + f"-c {cmd}"
101 if extra_args is not None:
102 cmd += f" {extra_args}"
103 return tplink_outlet_command(cmd)
105 def turn_on(self) -> bool:
106 return self.command('on')
108 def turn_off(self) -> bool:
109 return self.command('off')
111 def is_on(self) -> bool:
112 return self.get_on_duration_seconds() > 0
114 def is_off(self) -> bool:
115 return not self.is_on()
118 10.0, use_signals=False, error_message="Timed out waiting for tplink.py"
120 def get_info(self) -> Optional[Dict]:
121 cmd = self.get_cmdline() + "-c info"
122 out = subprocess.getoutput(cmd)
123 out = re.sub("Sent:.*\n", "", out)
124 out = re.sub("Received: *", "", out)
126 self.info = json.loads(out)["system"]["get_sysinfo"]
127 self.info_ts = datetime.datetime.now()
129 except Exception as e:
131 print(out, file=sys.stderr)
136 def get_on_duration_seconds(self) -> int:
137 self.info = self.get_info()
138 if self.info is None:
140 return int(self.info.get("on_time", "0"))
143 class TPLinkOutletWithChildren(TPLinkOutlet):
144 def __init__(self, name: str, mac: str, keywords: str = "") -> None:
145 super().__init__(name, mac, keywords)
146 self.children: List[str] = []
147 self.info: Optional[Dict] = None
148 self.info_ts: Optional[datetime.datetime] = None
149 assert "children" in self.keywords
150 self.info = self.get_info()
151 if self.info is not None:
152 for child in self.info["children"]:
153 self.children.append(child["id"])
156 def get_cmdline(self, child: Optional[str] = None) -> str:
158 f"{config.config['smart_outlets_tplink_location']} -m {self.mac} "
159 f"--no_logging_console "
161 if child is not None:
162 cmd += f"-x {child} "
167 self, cmd: str, child: str = None, extra_args: str = None
169 cmd = self.get_cmdline(child) + f"-c {cmd}"
170 if extra_args is not None:
171 cmd += f" {extra_args}"
172 logger.debug(f'About to execute {cmd}')
173 return tplink_outlet_command(cmd)
175 def get_children(self) -> List[str]:
178 def turn_on(self, child: str = None) -> bool:
179 return self.command("on", child)
181 def turn_off(self, child: str = None) -> bool:
182 return self.command("off", child)
184 def get_on_duration_seconds(self, child: str = None) -> int:
185 self.info = self.get_info()
187 if self.info is None:
189 return int(self.info.get("on_time", "0"))
191 if self.info is None:
193 for chi in self.info.get("children", {}):
194 if chi["id"] == child:
195 return int(chi.get("on_time", "0"))
199 class GoogleOutlet(BaseOutlet):
200 def __init__(self, name: str, mac: str, keywords: str = "") -> None:
201 super().__init__(name.strip(), mac.strip(), keywords)
204 def goog_name(self) -> str:
205 name = self.get_name()
206 return name.replace('_', ' ')
209 def parse_google_response(response: GoogleResponse) -> bool:
210 return response.success
212 def turn_on(self) -> bool:
213 return GoogleOutlet.parse_google_response(
214 ask_google('turn {self.goog_name()} on')
217 def turn_off(self) -> bool:
218 return GoogleOutlet.parse_google_response(
219 ask_google('turn {self.goog_name()} off')
222 def is_on(self) -> bool:
223 r = ask_google(f'is {self.goog_name()} on?')
226 return 'is on' in r.audio_transcription
228 def is_off(self) -> bool:
229 return not self.is_on()