1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
|
#!/usr/bin/env python3
"""Utilities for dealing with the smart outlets."""
from abc import abstractmethod
import datetime
import json
import logging
import os
import re
import subprocess
import sys
from typing import Any, Dict, List, Optional, Set
import argparse_utils
import config
import logging_utils
import smart_home.device as dev
from google_assistant import ask_google, GoogleResponse
from decorator_utils import timeout, memoized
logger = logging.getLogger(__name__)
parser = config.add_commandline_args(
f"Outlet Utils ({__file__})",
"Args related to smart outlets.",
)
parser.add_argument(
'--smart_outlets_tplink_location',
default='/home/scott/bin/tplink.py',
metavar='FILENAME',
help='The location of the tplink.py helper',
type=argparse_utils.valid_filename,
)
@timeout(
5.0, use_signals=False, error_message="Timed out waiting for tplink.py"
)
def tplink_outlet_command(command: str) -> bool:
result = os.system(command)
signal = result & 0xFF
if signal != 0:
logger.warning(f'{command} died with signal {signal}')
logging_utils.hlog("%s died with signal %d" % (command, signal))
return False
else:
exit_value = result >> 8
if exit_value != 0:
logger.warning(f'{command} failed, exited {exit_value}')
logging_utils.hlog("%s failed, exit %d" % (command, exit_value))
return False
logger.debug(f'{command} succeeded.')
return True
class BaseOutlet(dev.Device):
def __init__(self, name: str, mac: str, keywords: str = "") -> None:
super().__init__(name.strip(), mac.strip(), keywords)
self.info = None
@abstractmethod
def turn_on(self) -> bool:
pass
@abstractmethod
def turn_off(self) -> bool:
pass
@abstractmethod
def is_on(self) -> bool:
pass
@abstractmethod
def is_off(self) -> bool:
pass
class TPLinkOutlet(BaseOutlet):
def __init__(self, name: str, mac: str, keywords: str = '') -> None:
super().__init__(name, mac, keywords)
self.info: Optional[Dict] = None
self.info_ts: Optional[datetime.datetime] = None
@memoized
def get_tplink_name(self) -> Optional[str]:
self.info = self.get_info()
if self.info is not None:
return self.info["alias"]
return None
def get_cmdline(self) -> str:
cmd = (
f"{config.config['smart_outlets_tplink_location']} -m {self.mac} "
f"--no_logging_console "
)
return cmd
def command(self, cmd: str, extra_args: str = None) -> bool:
cmd = self.get_cmdline() + f"-c {cmd}"
if extra_args is not None:
cmd += f" {extra_args}"
return tplink_outlet_command(cmd)
def turn_on(self) -> bool:
return self.command('on')
def turn_off(self) -> bool:
return self.command('off')
def is_on(self) -> bool:
return self.get_on_duration_seconds() > 0
def is_off(self) -> bool:
return not self.is_on()
@timeout(
10.0, use_signals=False, error_message="Timed out waiting for tplink.py"
)
def get_info(self) -> Optional[Dict]:
cmd = self.get_cmdline() + "-c info"
out = subprocess.getoutput(cmd)
out = re.sub("Sent:.*\n", "", out)
out = re.sub("Received: *", "", out)
try:
self.info = json.loads(out)["system"]["get_sysinfo"]
self.info_ts = datetime.datetime.now()
return self.info
except Exception as e:
logger.exception(e)
print(out, file=sys.stderr)
self.info = None
self.info_ts = None
return None
def get_on_duration_seconds(self) -> int:
self.info = self.get_info()
if self.info is None:
return 0
return int(self.info.get("on_time", "0"))
class TPLinkOutletWithChildren(TPLinkOutlet):
def __init__(self, name: str, mac: str, keywords: str = "") -> None:
super().__init__(name, mac, keywords)
self.children: List[str] = []
self.info: Optional[Dict] = None
self.info_ts: Optional[datetime.datetime] = None
assert "children" in self.keywords
self.info = self.get_info()
if self.info is not None:
for child in self.info["children"]:
self.children.append(child["id"])
# override
def get_cmdline(self, child: Optional[str] = None) -> str:
cmd = (
f"{config.config['smart_outlets_tplink_location']} -m {self.mac} "
f"--no_logging_console "
)
if child is not None:
cmd += f"-x {child} "
return cmd
# override
def command(
self, cmd: str, child: str = None, extra_args: str = None
) -> bool:
cmd = self.get_cmdline(child) + f"-c {cmd}"
if extra_args is not None:
cmd += f" {extra_args}"
logger.debug(f'About to execute {cmd}')
return tplink_light_command(cmd)
def get_children(self) -> List[str]:
return self.children
def turn_on(self, child: str = None) -> bool:
return self.command("on", child)
def turn_off(self, child: str = None) -> bool:
return self.command("off", child)
def get_on_duration_seconds(self, child: str = None) -> int:
self.info = self.get_info()
if child is None:
if self.info is None:
return 0
return int(self.info.get("on_time", "0"))
else:
if self.info is None:
return 0
for chi in self.info.get("children", {}):
if chi["id"] == child:
return int(chi.get("on_time", "0"))
return 0
class GoogleOutlet(BaseOutlet):
def __init__(self, name: str, mac: str, keywords: str = "") -> None:
super().__init__(name.strip(), mac.strip(), keywords)
self.info = None
def goog_name(self) -> str:
name = self.get_name()
return name.replace('_', ' ')
@staticmethod
def parse_google_response(response: GoogleResponse) -> bool:
return response.success
def turn_on(self) -> bool:
return GoogleOutlet.parse_google_response(
ask_google('turn {self.goog_name()} on')
)
def turn_off(self) -> bool:
return GoogleOutlet.parse_google_response(
ask_google('turn {self.goog_name()} off')
)
def is_on(self) -> bool:
r = ask_google(f'is {self.goog_name()} on?')
if not r.success:
return False
return 'is on' in r.audio_transcription
def is_off(self) -> bool:
return not self.is_on()
|
