smart_home/lights.py

   1 #!/usr/bin/env python3
   2
   3 """Utilities for dealing with the smart lights."""
   4
   5 from abc import abstractmethod
   6 import datetime
   7 import json
   8 import logging
   9 import os
  10 import re
  11 import subprocess
  12 import sys
  13 from typing import Any, Dict, List, Optional, Tuple
  14
  15 from overrides import overrides
  16 import tinytuya as tt
  17
  18 import ansi
  19 import argparse_utils
  20 import arper
  21 import config
  22 import logging_utils
  23 import smart_home.device as dev
  24 from google_assistant import ask_google, GoogleResponse
  25 from decorator_utils import timeout, memoized
  26
  27 logger = logging.getLogger(__name__)
  28
  29 args = config.add_commandline_args(
  30     f"Smart Lights ({__file__})",
  31     "Args related to smart lights.",
  32 )
  33 args.add_argument(
  34     '--smart_lights_tplink_location',
  35     default='/home/scott/bin/tplink.py',
  36     metavar='FILENAME',
  37     help='The location of the tplink.py helper',
  38     type=argparse_utils.valid_filename,
  39 )
  40
  41
  42 @timeout(
  43     5.0, use_signals=False, error_message="Timed out waiting for tplink.py"
  44 )
  45 def tplink_light_command(command: str) -> bool:
  46     result = os.system(command)
  47     signal = result & 0xFF
  48     if signal != 0:
  49         logger.warning(f'{command} died with signal {signal}')
  50         logging_utils.hlog("%s died with signal %d" % (command, signal))
  51         return False
  52     else:
  53         exit_value = result >> 8
  54         if exit_value != 0:
  55             logger.warning(f'{command} failed, exited {exit_value}')
  56             logging_utils.hlog("%s failed, exit %d" % (command, exit_value))
  57             return False
  58     logger.debug(f'{command} succeeded.')
  59     return True
  60
  61
  62 class BaseLight(dev.Device):
  63     def __init__(self, name: str, mac: str, keywords: str = "") -> None:
  64         super().__init__(name.strip(), mac.strip(), keywords)
  65
  66     @staticmethod
  67     def parse_color_string(color: str) -> Optional[Tuple[int, int, int]]:
  68         m = re.match(
  69             'r#?([0-9a-fA-F][0-9a-fA-F])([0-9a-fA-F][0-9a-fA-F])([0-9a-fA-F][0-9a-fA-F])',
  70             color
  71         )
  72         if m is not None and len(m.group) == 3:
  73             red = int(m.group(0), 16)
  74             green = int(m.group(1), 16)
  75             blue = int(m.group(2), 16)
  76             return (red, green, blue)
  77         color = color.lower()
  78         return ansi.COLOR_NAMES_TO_RGB.get(color, None)
  79
  80     @abstractmethod
  81     def turn_on(self) -> bool:
  82         pass
  83
  84     @abstractmethod
  85     def turn_off(self) -> bool:
  86         pass
  87
  88     @abstractmethod
  89     def is_on(self) -> bool:
  90         pass
  91
  92     @abstractmethod
  93     def is_off(self) -> bool:
  94         pass
  95
  96     @abstractmethod
  97     def get_dimmer_level(self) -> Optional[int]:
  98         pass
  99
 100     @abstractmethod
 101     def set_dimmer_level(self, level: int) -> bool:
 102         pass
 103
 104     @abstractmethod
 105     def make_color(self, color: str) -> bool:
 106         pass
 107
 108
 109 class GoogleLight(BaseLight):
 110     def __init__(self, name: str, mac: str, keywords: str = "") -> None:
 111         super().__init__(name, mac, keywords)
 112
 113     def goog_name(self) -> str:
 114         name = self.get_name()
 115         return name.replace("_", " ")
 116
 117     @staticmethod
 118     def parse_google_response(response: GoogleResponse) -> bool:
 119         return response.success
 120
 121     @overrides
 122     def turn_on(self) -> bool:
 123         return GoogleLight.parse_google_response(
 124             ask_google(f"turn {self.goog_name()} on")
 125         )
 126
 127     @overrides
 128     def turn_off(self) -> bool:
 129         return GoogleLight.parse_google_response(
 130             ask_google(f"turn {self.goog_name()} off")
 131         )
 132
 133     @overrides
 134     def is_on(self) -> bool:
 135         r = ask_google(f"is {self.goog_name()} on?")
 136         if not r.success:
 137             return False
 138         return 'is on' in r.audio_transcription
 139
 140     @overrides
 141     def is_off(self) -> bool:
 142         return not self.is_on()
 143
 144     @overrides
 145     def get_dimmer_level(self) -> Optional[int]:
 146         if not self.has_keyword("dimmer"):
 147             return False
 148         r = ask_google(f'how bright is {self.goog_name()}?')
 149         if not r.success:
 150             return None
 151
 152         # the bookcase one is set to 40% bright
 153         txt = r.audio_transcription
 154         m = re.search(r"(\d+)% bright", txt)
 155         if m is not None:
 156             return int(m.group(1))
 157         if "is off" in txt:
 158             return 0
 159         return None
 160
 161     @overrides
 162     def set_dimmer_level(self, level: int) -> bool:
 163         if not self.has_keyword("dimmer"):
 164             return False
 165         if 0 <= level <= 100:
 166             was_on = self.is_on()
 167             r = ask_google(f"set {self.goog_name()} to {level} percent")
 168             if not r.success:
 169                 return False
 170             if not was_on:
 171                 self.turn_off()
 172             return True
 173         return False
 174
 175     @overrides
 176     def make_color(self, color: str) -> bool:
 177         return GoogleLight.parse_google_response(
 178             ask_google(f"make {self.goog_name()} {color}")
 179         )
 180
 181
 182 class TuyaLight(BaseLight):
 183     ids_by_mac = {
 184         '68:C6:3A:DE:1A:94': '8844664268c63ade1a94',
 185         '68:C6:3A:DE:27:1A': '8844664268c63ade271a',
 186         '68:C6:3A:DE:1D:95': '8844664268c63ade1d95',
 187         '68:C6:3A:DE:19:B3': '8844664268c63ade19b3',
 188         '80:7D:3A:77:3B:F5': '07445340807d3a773bf5',
 189         '80:7D:3A:58:37:02': '07445340807d3a583702',
 190     }
 191     keys_by_mac = {
 192         '68:C6:3A:DE:1A:94': '237f19b1b3d49c36',
 193         '68:C6:3A:DE:27:1A': '237f19b1b3d49c36',
 194         '68:C6:3A:DE:1D:95': '04b90fc5cd7625d8',
 195         '68:C6:3A:DE:19:B3': '2d601f2892f1aefd',
 196         '80:7D:3A:77:3B:F5': '27ab921fe4633519',
 197         '80:7D:3A:58:37:02': '8559b5416bfa0c05',
 198     }
 199
 200     def __init__(self, name: str, mac: str, keywords: str = "") -> None:
 201         super().__init__(name, mac, keywords)
 202         mac = mac.upper()
 203         if mac not in TuyaLight.ids_by_mac or mac not in TuyaLight.keys_by_mac:
 204             raise Exception(f'{mac} is unknown; add it to ids_by_mac and keys_by_mac')
 205         self.devid = TuyaLight.ids_by_mac[mac]
 206         self.key = TuyaLight.keys_by_mac[mac]
 207         self.arper = arper.Arper()
 208         ip = self.get_ip()
 209         self.bulb = tt.BulbDevice(self.devid, ip, local_key=self.key)
 210
 211     def get_status(self) -> Dict[str, Any]:
 212         return self.bulb.status()
 213
 214     @overrides
 215     def turn_on(self) -> bool:
 216         self.bulb.turn_on()
 217         return True
 218
 219     @overrides
 220     def turn_off(self) -> bool:
 221         self.bulb.turn_off()
 222         return True
 223
 224     @overrides
 225     def is_on(self) -> bool:
 226         s = self.get_status()
 227         return s['dps']['1']
 228
 229     @overrides
 230     def is_off(self) -> bool:
 231         return not self.is_on()
 232
 233     @overrides
 234     def get_dimmer_level(self) -> Optional[int]:
 235         s = self.get_status()
 236         return s['dps']['3']
 237
 238     @overrides
 239     def set_dimmer_level(self, level: int) -> bool:
 240         self.bulb.set_brightness(level)
 241         return True
 242
 243     @overrides
 244     def make_color(self, color: str) -> bool:
 245         rgb = BaseLight.parse_color_string(color)
 246         if rgb is not None:
 247             self.bulb.set_colour(rgb[0], rgb[1], rgb[2])
 248             return True
 249         return False
 250
 251
 252 class TPLinkLight(BaseLight):
 253     def __init__(self, name: str, mac: str, keywords: str = "") -> None:
 254         super().__init__(name, mac, keywords)
 255         self.children: List[str] = []
 256         self.info: Optional[Dict] = None
 257         self.info_ts: Optional[datetime.datetime] = None
 258         if "children" in self.keywords:
 259             self.info = self.get_info()
 260             if self.info is not None:
 261                 for child in self.info["children"]:
 262                     self.children.append(child["id"])
 263
 264     @memoized
 265     def get_tplink_name(self) -> Optional[str]:
 266         self.info = self.get_info()
 267         if self.info is not None:
 268             return self.info["alias"]
 269         return None
 270
 271     def get_cmdline(self, child: str = None) -> str:
 272         cmd = (
 273             f"{config.config['smart_lights_tplink_location']} -m {self.mac} "
 274             f"--no_logging_console "
 275         )
 276         if child is not None:
 277             cmd += f"-x {child} "
 278         return cmd
 279
 280     def get_children(self) -> List[str]:
 281         return self.children
 282
 283     def command(
 284         self, cmd: str, child: str = None, extra_args: str = None
 285     ) -> bool:
 286         cmd = self.get_cmdline(child) + f"-c {cmd}"
 287         if extra_args is not None:
 288             cmd += f" {extra_args}"
 289         logger.debug(f'About to execute {cmd}')
 290         return tplink_light_command(cmd)
 291
 292     @overrides
 293     def turn_on(self, child: str = None) -> bool:
 294         return self.command("on", child)
 295
 296     @overrides
 297     def turn_off(self, child: str = None) -> bool:
 298         return self.command("off", child)
 299
 300     @overrides
 301     def is_on(self) -> bool:
 302         return self.get_on_duration_seconds() > 0
 303
 304     @overrides
 305     def is_off(self) -> bool:
 306         return not self.is_on()
 307
 308     @overrides
 309     def make_color(self, color: str) -> bool:
 310         raise NotImplementedError
 311
 312     @timeout(
 313         10.0, use_signals=False, error_message="Timed out waiting for tplink.py"
 314     )
 315     def get_info(self) -> Optional[Dict]:
 316         cmd = self.get_cmdline() + "-c info"
 317         out = subprocess.getoutput(cmd)
 318         out = re.sub("Sent:.*\n", "", out)
 319         out = re.sub("Received: *", "", out)
 320         try:
 321             self.info = json.loads(out)["system"]["get_sysinfo"]
 322             self.info_ts = datetime.datetime.now()
 323             return self.info
 324         except Exception as e:
 325             logger.exception(e)
 326             print(out, file=sys.stderr)
 327             self.info = None
 328             self.info_ts = None
 329             return None
 330
 331     def get_on_duration_seconds(self, child: str = None) -> int:
 332         self.info = self.get_info()
 333         if child is None:
 334             if self.info is None:
 335                 return 0
 336             return int(self.info.get("on_time", "0"))
 337         else:
 338             if self.info is None:
 339                 return 0
 340             for chi in self.info.get("children", {}):
 341                 if chi["id"] == child:
 342                     return int(chi.get("on_time", "0"))
 343         return 0
 344
 345     @overrides
 346     def get_dimmer_level(self) -> Optional[int]:
 347         if not self.has_keyword("dimmer"):
 348             return False
 349         self.info = self.get_info()
 350         if self.info is None:
 351             return None
 352         return int(self.info.get("brightness", "0"))
 353
 354     @overrides
 355     def set_dimmer_level(self, level: int) -> bool:
 356         if not self.has_keyword("dimmer"):
 357             return False
 358         cmd = (
 359             self.get_cmdline()
 360             + f'-j \'{{"smartlife.iot.dimmer":{{"set_brightness":{{"brightness":{level} }} }} }}\''
 361         )
 362         return tplink_light_command(cmd)
 363
 364
 365 # class GoogleLightGroup(GoogleLight):
 366 #     def __init__(self, name: str, members: List[GoogleLight], keywords: str = "") -> None:
 367 #         if len(members) < 1:
 368 #             raise Exception("There must be at least one light in the group.")
 369 #         self.members = members
 370 #         mac = GoogleLightGroup.make_up_mac(members)
 371 #         super().__init__(name, mac, keywords)
 372
 373 #     @staticmethod
 374 #     def make_up_mac(members: List[GoogleLight]):
 375 #         mac = members[0].get_mac()
 376 #         b = mac.split(':')
 377 #         b[5] = int(b[5], 16) + 1
 378 #         if b[5] > 255:
 379 #             b[5] = 0
 380 #         b[5] = str(b[5])
 381 #         return ":".join(b)
 382
 383 #     def is_on(self) -> bool:
 384 #         r = ask_google(f"are {self.goog_name()} on?")
 385 #         if not r.success:
 386 #             return False
 387 #         return 'is on' in r.audio_transcription
 388
 389 #     def get_dimmer_level(self) -> Optional[int]:
 390 #         if not self.has_keyword("dimmer"):
 391 #             return False
 392 #         r = ask_google(f'how bright are {self.goog_name()}?')
 393 #         if not r.success:
 394 #             return None
 395
 396 #         # four lights are set to 100% brightness
 397 #         txt = r.audio_transcription
 398 #         m = re.search(r"(\d+)% bright", txt)
 399 #         if m is not None:
 400 #             return int(m.group(1))
 401 #         if "is off" in txt:
 402 #             return 0
 403 #         return None
 404
 405 #     def set_dimmer_level(self, level: int) -> bool:
 406 #         if not self.has_keyword("dimmer"):
 407 #             return False
 408 #         if 0 <= level <= 100:
 409 #             was_on = self.is_on()
 410 #             r = ask_google(f"set {self.goog_name()} to {level} percent")
 411 #             if not r.success:
 412 #                 return False
 413 #             if not was_on:
 414 #                 self.turn_off()
 415 #             return True
 416 #         return False
 417
 418 #     def make_color(self, color: str) -> bool:
 419 #         return GoogleLight.parse_google_response(
 420 #             ask_google(f"make {self.goog_name()} {color}")
 421 #         )