diff options
Diffstat (limited to 'smart_home')
| -rw-r--r-- | smart_home/chromecasts.py | 88 | ||||
| -rw-r--r-- | smart_home/device.py | 13 | ||||
| -rw-r--r-- | smart_home/lights.py | 73 | ||||
| -rw-r--r-- | smart_home/registry.py (renamed from smart_home/config.py) | 96 |
4 files changed, 204 insertions, 66 deletions
diff --git a/smart_home/chromecasts.py b/smart_home/chromecasts.py new file mode 100644 index 0000000..08290e5 --- /dev/null +++ b/smart_home/chromecasts.py @@ -0,0 +1,88 @@ +#!/usr/bin/env python3 + +"""Utilities for dealing with the webcams.""" + +import logging +import time + +import pychromecast + +from decorator_utils import memoized +import smart_home.device as dev + +logger = logging.getLogger(__name__) + + +class BaseChromecast(dev.Device): + def __init__(self, name: str, mac: str, keywords: str = "") -> None: + super().__init__(name.strip(), mac.strip(), keywords) + ip = self.get_ip() + self.cast = pychromecast.Chromecast(ip) + self.cast.wait() + time.sleep(0.1) + + def is_idle(self): + return self.cast.is_idle + + @memoized + def get_uuid(self): + return self.cast.uuid + + @memoized + def get_friendly_name(self): + return self.cast.name + + def get_uri(self): + return self.cast.url + + @memoized + def get_model_name(self): + return self.cast.model_name + + @memoized + def get_cast_type(self): + return self.cast.cast_type + + @memoized + def app_id(self): + return self.cast.app_id + + def get_app_display_name(self): + return self.cast.app_display_name + + def get_media_controller(self): + return self.cast.media_controller + + def status(self): + if self.is_idle(): + return 'idle' + app = self.get_app_display_name() + mc = self.get_media_controller() + status = mc.status + return f'{app} / {status.title}' + + def start_app(self, app_id, force_launch=False): + """Start an app on the Chromecast.""" + self.cast.start_app(app_id, force_launch) + + def quit_app(self): + """Tells the Chromecast to quit current app_id.""" + self.cast.quit_app() + + def volume_up(self, delta=0.1): + """Increment volume by 0.1 (or delta) unless it is already maxed. + Returns the new volume. + """ + return self.cast.volume_up(delta) + + def volume_down(self, delta=0.1): + """Decrement the volume by 0.1 (or delta) unless it is already 0. + Returns the new volume. + """ + return self.cast.volume_down(delta) + + def __repr__(self): + return ( + f"Chromecast({self.cast.socket_client.host!r}, port={self.cast.socket_client.port!r}, " + f"device={self.cast.device!r})" + ) diff --git a/smart_home/device.py b/smart_home/device.py index 04b0bfe..9675b7c 100644 --- a/smart_home/device.py +++ b/smart_home/device.py @@ -28,9 +28,20 @@ class Device(object): def get_mac(self) -> str: return self.mac - def get_ip(self) -> str: + def get_ip(self) -> Optional[str]: return self.arper.get_ip_by_mac(self.mac) + def has_static_ip(self) -> bool: + for kw in self.kws: + m = re.search(r'static:([\d\.]+)', kw) + if m is not None: + ip = m.group(1) + assert self.get_ip() == ip + return True + return False + + # Add command -> URL logic here. + def get_keywords(self) -> Optional[List[str]]: return self.kws diff --git a/smart_home/lights.py b/smart_home/lights.py index 5446722..76b1500 100644 --- a/smart_home/lights.py +++ b/smart_home/lights.py @@ -10,11 +10,14 @@ import os import re import subprocess import sys -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple +from overrides import overrides import tinytuya as tt +import ansi import argparse_utils +import arper import config import logging_utils import smart_home.device as dev @@ -23,11 +26,11 @@ from decorator_utils import timeout, memoized logger = logging.getLogger(__name__) -parser = config.add_commandline_args( +args = config.add_commandline_args( f"Smart Lights ({__file__})", "Args related to smart lights.", ) -parser.add_argument( +args.add_argument( '--smart_lights_tplink_location', default='/home/scott/bin/tplink.py', metavar='FILENAME', @@ -60,6 +63,20 @@ class BaseLight(dev.Device): def __init__(self, name: str, mac: str, keywords: str = "") -> None: super().__init__(name.strip(), mac.strip(), keywords) + @staticmethod + def parse_color_string(color: str) -> Optional[Tuple[int, int, int]]: + m = re.match( + '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])', + color + ) + if m is not None and len(m.group) == 3: + red = int(m.group(0), 16) + green = int(m.group(1), 16) + blue = int(m.group(2), 16) + return (red, green, blue) + color = color.lower() + return ansi.COLOR_NAMES_TO_RGB.get(color, None) + @abstractmethod def turn_on(self) -> bool: pass @@ -101,25 +118,30 @@ class GoogleLight(BaseLight): def parse_google_response(response: GoogleResponse) -> bool: return response.success + @overrides def turn_on(self) -> bool: return GoogleLight.parse_google_response( ask_google(f"turn {self.goog_name()} on") ) + @overrides def turn_off(self) -> bool: return GoogleLight.parse_google_response( ask_google(f"turn {self.goog_name()} off") ) + @overrides 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 + @overrides def is_off(self) -> bool: return not self.is_on() + @overrides def get_dimmer_level(self) -> Optional[int]: if not self.has_keyword("dimmer"): return False @@ -136,6 +158,7 @@ class GoogleLight(BaseLight): return 0 return None + @overrides def set_dimmer_level(self, level: int) -> bool: if not self.has_keyword("dimmer"): return False @@ -149,6 +172,7 @@ class GoogleLight(BaseLight): return True return False + @overrides def make_color(self, color: str) -> bool: return GoogleLight.parse_google_response( ask_google(f"make {self.goog_name()} {color}") @@ -174,50 +198,55 @@ class TuyaLight(BaseLight): } def __init__(self, name: str, mac: str, keywords: str = "") -> None: - from subprocess import Popen, PIPE super().__init__(name, mac, keywords) mac = mac.upper() if mac not in TuyaLight.ids_by_mac or mac not in TuyaLight.keys_by_mac: raise Exception(f'{mac} is unknown; add it to ids_by_mac and keys_by_mac') self.devid = TuyaLight.ids_by_mac[mac] self.key = TuyaLight.keys_by_mac[mac] - try: - pid = Popen(['maclookup', mac], stdout=PIPE) - ip = pid.communicate()[0] - ip = ip[:-1] - except Exception: - ip = '0.0.0.0' + self.arper = arper.Arper() + ip = self.get_ip() self.bulb = tt.BulbDevice(self.devid, ip, local_key=self.key) + def get_status(self) -> Dict[str, Any]: + return self.bulb.status() + + @overrides def turn_on(self) -> bool: self.bulb.turn_on() return True + @overrides def turn_off(self) -> bool: self.bulb.turn_off() return True - def get_status(self) -> Dict[str, Any]: - return self.bulb.status() - + @overrides def is_on(self) -> bool: s = self.get_status() return s['dps']['1'] + @overrides def is_off(self) -> bool: return not self.is_on() + @overrides def get_dimmer_level(self) -> Optional[int]: s = self.get_status() return s['dps']['3'] + @overrides def set_dimmer_level(self, level: int) -> bool: self.bulb.set_brightness(level) return True + @overrides def make_color(self, color: str) -> bool: - self.bulb.set_colour(255,0,0) - return True + rgb = BaseLight.parse_color_string(color) + if rgb is not None: + self.bulb.set_colour(rgb[0], rgb[1], rgb[2]) + return True + return False class TPLinkLight(BaseLight): @@ -260,18 +289,23 @@ class TPLinkLight(BaseLight): logger.debug(f'About to execute {cmd}') return tplink_light_command(cmd) + @overrides def turn_on(self, child: str = None) -> bool: return self.command("on", child) + @overrides def turn_off(self, child: str = None) -> bool: return self.command("off", child) + @overrides def is_on(self) -> bool: return self.get_on_duration_seconds() > 0 + @overrides def is_off(self) -> bool: return not self.is_on() + @overrides def make_color(self, color: str) -> bool: raise NotImplementedError @@ -308,13 +342,7 @@ class TPLinkLight(BaseLight): return int(chi.get("on_time", "0")) return 0 - def get_on_limit_seconds(self) -> Optional[int]: - for kw in self.kws: - m = re.search(r"timeout:(\d+)", kw) - if m is not None: - return int(m.group(1)) * 60 - return None - + @overrides def get_dimmer_level(self) -> Optional[int]: if not self.has_keyword("dimmer"): return False @@ -323,6 +351,7 @@ class TPLinkLight(BaseLight): return None return int(self.info.get("brightness", "0")) + @overrides def set_dimmer_level(self, level: int) -> bool: if not self.has_keyword("dimmer"): return False diff --git a/smart_home/config.py b/smart_home/registry.py index a28caa7..2d23981 100644 --- a/smart_home/config.py +++ b/smart_home/registry.py @@ -10,15 +10,16 @@ import file_utils import logical_search import smart_home.device as device import smart_home.cameras as cameras +import smart_home.chromecasts as chromecasts import smart_home.lights as lights import smart_home.outlets as outlets -parser = config.add_commandline_args( - f"Smart Home Config ({__file__})", - "Args related to the smart home config." +args = config.add_commandline_args( + f"Smart Home Registry ({__file__})", + "Args related to the smart home configuration registry." ) -parser.add_argument( - '--smart_home_config_file_location', +args.add_argument( + '--smart_home_registry_file_location', default='/home/scott/bin/network_mac_addresses.txt', metavar='FILENAME', help='The location of network_mac_addresses.txt', @@ -29,10 +30,10 @@ parser.add_argument( logger = logging.getLogger(__file__) -class SmartHomeConfig(object): +class SmartHomeRegistry(object): def __init__( self, - config_file: Optional[str] = None, + registry_file: Optional[str] = None, filters: List[str] = ['smart'], ) -> None: self._macs_by_name = {} @@ -42,13 +43,13 @@ class SmartHomeConfig(object): self._corpus = logical_search.Corpus() # Read the disk config file... - if config_file is None: - config_file = config.config[ - 'smart_home_config_file_location' + if registry_file is None: + registry_file = config.config[ + 'smart_home_registry_file_location' ] - assert file_utils.does_file_exist(config_file) - logger.debug(f'Reading {config_file}') - with open(config_file, "r") as f: + assert file_utils.does_file_exist(registry_file) + logger.debug(f'Reading {registry_file}') + with open(registry_file, "r") as f: contents = f.readlines() # Parse the contents... @@ -58,7 +59,7 @@ class SmartHomeConfig(object): line = line.strip() if line == "": continue - logger.debug(f'> {line}') + logger.debug(f'SH-CONFIG> {line}') (mac, name, keywords) = line.split(",") mac = mac.strip() name = name.strip() @@ -139,38 +140,47 @@ class SmartHomeConfig(object): name = self._names_by_mac[mac] kws = self._keywords_by_mac[mac] logger.debug(f'Found {name} -> {mac} ({kws})') - if 'light' in kws.lower(): - if 'tplink' in kws.lower(): - logger.debug(' ...a TPLinkLight') - return lights.TPLinkLight(name, mac, kws) - elif 'tuya' in kws.lower(): - logger.debug(' ...a TuyaLight') - return lights.TuyaLight(name, mac, kws) - elif 'goog' in kws.lower(): - logger.debug(' ...a GoogleLight') - return lights.GoogleLight(name, mac, kws) - else: - raise Exception(f'Unknown light device: {name}, {mac}, {kws}') - elif 'outlet' in kws.lower(): - if 'tplink' in kws.lower(): - if 'children' in kws.lower(): - logger.debug(' ...a TPLinkOutletWithChildren') - return outlets.TPLinkOutletWithChildren(name, mac, kws) + try: + if 'light' in kws.lower(): + if 'tplink' in kws.lower(): + logger.debug(' ...a TPLinkLight') + return lights.TPLinkLight(name, mac, kws) + elif 'tuya' in kws.lower(): + logger.debug(' ...a TuyaLight') + return lights.TuyaLight(name, mac, kws) + elif 'goog' in kws.lower(): + logger.debug(' ...a GoogleLight') + return lights.GoogleLight(name, mac, kws) + else: + raise Exception(f'Unknown light device: {name}, {mac}, {kws}') + elif 'outlet' in kws.lower(): + if 'tplink' in kws.lower(): + if 'children' in kws.lower(): + logger.debug(' ...a TPLinkOutletWithChildren') + return outlets.TPLinkOutletWithChildren(name, mac, kws) + else: + logger.debug(' ...a TPLinkOutlet') + return outlets.TPLinkOutlet(name, mac, kws) + elif 'goog' in kws.lower(): + logger.debug(' ...a GoogleOutlet') + return outlets.GoogleOutlet(name, mac, kws) else: - logger.debug(' ...a TPLinkOutlet') - return outlets.TPLinkOutlet(name, mac, kws) - elif 'goog' in kws.lower(): - logger.debug(' ...a GoogleOutlet') - return outlets.GoogleOutlet(name, mac, kws) + raise Exception(f'Unknown outlet device: {name}, {mac}, {kws}') + elif 'camera' in kws.lower(): + logger.debug(' ...a BaseCamera') + return cameras.BaseCamera(name, mac, kws) + elif 'ccast' in kws.lower(): + logger.debug(' ...a Chromecast') + return chromecasts.BaseChromecast(name, mac, kws) else: - raise Exception(f'Unknown outlet device: {name}, {mac}, {kws}') - elif 'camera' in kws.lower(): - logger.debug(' ...a BaseCamera') - return cameras.BaseCamera(name, mac, kws) - else: - logger.debug(' ...an unknown device (should this be here?)') + logger.debug(' ...an unknown device (should this be here?)') + return device.Device(name, mac, kws) + except Exception as e: + logger.warning( + f'Got exception {e} while trying to communicate with device {name}/{mac}.' + ) return device.Device(name, mac, kws) - logger.warning(f'{mac} is not known, returning None') + logger.warning(f'{mac} is not a known smart home device, returning None') return None def query(self, query: str) -> List[device.Device]: |