From eb9e6df32ed696158bf34dba6464277b648f5c74 Mon Sep 17 00:00:00 2001 From: Scott Gasch Date: Sun, 31 Oct 2021 13:08:51 -0700 Subject: Ugh, a bunch of things. @overrides. --lmodule. Chromecasts. etc... --- smart_home/chromecasts.py | 88 ++++++++++++++++++++ smart_home/config.py | 190 ------------------------------------------- smart_home/device.py | 13 ++- smart_home/lights.py | 73 ++++++++++++----- smart_home/registry.py | 200 ++++++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 351 insertions(+), 213 deletions(-) create mode 100644 smart_home/chromecasts.py delete mode 100644 smart_home/config.py create mode 100644 smart_home/registry.py (limited to 'smart_home') 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/config.py b/smart_home/config.py deleted file mode 100644 index a28caa7..0000000 --- a/smart_home/config.py +++ /dev/null @@ -1,190 +0,0 @@ -#!/usr/bin/env python3 - -import logging -import re -from typing import List, Optional, Set - -import argparse_utils -import config -import file_utils -import logical_search -import smart_home.device as device -import smart_home.cameras as cameras -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." -) -parser.add_argument( - '--smart_home_config_file_location', - default='/home/scott/bin/network_mac_addresses.txt', - metavar='FILENAME', - help='The location of network_mac_addresses.txt', - type=argparse_utils.valid_filename, -) - - -logger = logging.getLogger(__file__) - - -class SmartHomeConfig(object): - def __init__( - self, - config_file: Optional[str] = None, - filters: List[str] = ['smart'], - ) -> None: - self._macs_by_name = {} - self._keywords_by_name = {} - self._keywords_by_mac = {} - self._names_by_mac = {} - self._corpus = logical_search.Corpus() - - # Read the disk config file... - if config_file is None: - config_file = config.config[ - 'smart_home_config_file_location' - ] - assert file_utils.does_file_exist(config_file) - logger.debug(f'Reading {config_file}') - with open(config_file, "r") as f: - contents = f.readlines() - - # Parse the contents... - for line in contents: - line = line.rstrip("\n") - line = re.sub(r"#.*$", r"", line) - line = line.strip() - if line == "": - continue - logger.debug(f'> {line}') - (mac, name, keywords) = line.split(",") - mac = mac.strip() - name = name.strip() - keywords = keywords.strip() - - skip = False - if filters is not None: - for f in filters: - if f not in keywords: - logger.debug(f'Skipping this entry b/c of filter {f}') - skip = True - break - if not skip: - self._macs_by_name[name] = mac - self._keywords_by_name[name] = keywords - self._keywords_by_mac[mac] = keywords - self._names_by_mac[mac] = name - self.index_device(name, keywords, mac) - - def index_device(self, name: str, keywords: str, mac: str) -> None: - properties = [("name", name)] - tags = set() - for kw in keywords.split(): - if ":" in kw: - key, value = kw.split(":") - properties.append((key, value)) - else: - tags.add(kw) - device = logical_search.Document( - docid=mac, - tags=tags, - properties=properties, - reference=None, - ) - logger.debug(f'Indexing document {device}') - self._corpus.add_doc(device) - - def __repr__(self) -> str: - s = "Known devices:\n" - for name, keywords in self._keywords_by_name.items(): - mac = self._macs_by_name[name] - s += f" {name} ({mac}) => {keywords}\n" - return s - - def get_keywords_by_name(self, name: str) -> Optional[device.Device]: - return self._keywords_by_name.get(name, None) - - def get_macs_by_name(self, name: str) -> Set[str]: - retval = set() - for (mac, lname) in self._names_by_mac.items(): - if name in lname: - retval.add(mac) - return retval - - def get_macs_by_keyword(self, keyword: str) -> Set[str]: - retval = set() - for (mac, keywords) in self._keywords_by_mac.items(): - if keyword in keywords: - retval.add(mac) - return retval - - def get_device_by_name(self, name: str) -> Optional[device.Device]: - if name in self._macs_by_name: - return self.get_device_by_mac(self._macs_by_name[name]) - return None - - def get_all_devices(self) -> List[device.Device]: - retval = [] - for (mac, kws) in self._keywords_by_mac.items(): - if mac is not None: - device = self.get_device_by_mac(mac) - if device is not None: - retval.append(device) - return retval - - def get_device_by_mac(self, mac: str) -> Optional[device.Device]: - if mac in self._keywords_by_mac: - 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) - 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: - 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?)') - return device.Device(name, mac, kws) - logger.warning(f'{mac} is not known, returning None') - return None - - def query(self, query: str) -> List[device.Device]: - """Evaluates a lighting query expression formed of keywords to search - for, logical operators (and, or, not), and parenthesis. - Returns a list of matching lights. - """ - retval = [] - logger.debug(f'Executing query {query}') - results = self._corpus.query(query) - if results is not None: - for mac in results: - if mac is not None: - device = self.get_device_by_mac(mac) - if device is not None: - retval.append(device) - return retval 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/registry.py b/smart_home/registry.py new file mode 100644 index 0000000..2d23981 --- /dev/null +++ b/smart_home/registry.py @@ -0,0 +1,200 @@ +#!/usr/bin/env python3 + +import logging +import re +from typing import List, Optional, Set + +import argparse_utils +import config +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 + +args = config.add_commandline_args( + f"Smart Home Registry ({__file__})", + "Args related to the smart home configuration registry." +) +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', + type=argparse_utils.valid_filename, +) + + +logger = logging.getLogger(__file__) + + +class SmartHomeRegistry(object): + def __init__( + self, + registry_file: Optional[str] = None, + filters: List[str] = ['smart'], + ) -> None: + self._macs_by_name = {} + self._keywords_by_name = {} + self._keywords_by_mac = {} + self._names_by_mac = {} + self._corpus = logical_search.Corpus() + + # Read the disk config file... + if registry_file is None: + registry_file = config.config[ + 'smart_home_registry_file_location' + ] + 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... + for line in contents: + line = line.rstrip("\n") + line = re.sub(r"#.*$", r"", line) + line = line.strip() + if line == "": + continue + logger.debug(f'SH-CONFIG> {line}') + (mac, name, keywords) = line.split(",") + mac = mac.strip() + name = name.strip() + keywords = keywords.strip() + + skip = False + if filters is not None: + for f in filters: + if f not in keywords: + logger.debug(f'Skipping this entry b/c of filter {f}') + skip = True + break + if not skip: + self._macs_by_name[name] = mac + self._keywords_by_name[name] = keywords + self._keywords_by_mac[mac] = keywords + self._names_by_mac[mac] = name + self.index_device(name, keywords, mac) + + def index_device(self, name: str, keywords: str, mac: str) -> None: + properties = [("name", name)] + tags = set() + for kw in keywords.split(): + if ":" in kw: + key, value = kw.split(":") + properties.append((key, value)) + else: + tags.add(kw) + device = logical_search.Document( + docid=mac, + tags=tags, + properties=properties, + reference=None, + ) + logger.debug(f'Indexing document {device}') + self._corpus.add_doc(device) + + def __repr__(self) -> str: + s = "Known devices:\n" + for name, keywords in self._keywords_by_name.items(): + mac = self._macs_by_name[name] + s += f" {name} ({mac}) => {keywords}\n" + return s + + def get_keywords_by_name(self, name: str) -> Optional[device.Device]: + return self._keywords_by_name.get(name, None) + + def get_macs_by_name(self, name: str) -> Set[str]: + retval = set() + for (mac, lname) in self._names_by_mac.items(): + if name in lname: + retval.add(mac) + return retval + + def get_macs_by_keyword(self, keyword: str) -> Set[str]: + retval = set() + for (mac, keywords) in self._keywords_by_mac.items(): + if keyword in keywords: + retval.add(mac) + return retval + + def get_device_by_name(self, name: str) -> Optional[device.Device]: + if name in self._macs_by_name: + return self.get_device_by_mac(self._macs_by_name[name]) + return None + + def get_all_devices(self) -> List[device.Device]: + retval = [] + for (mac, kws) in self._keywords_by_mac.items(): + if mac is not None: + device = self.get_device_by_mac(mac) + if device is not None: + retval.append(device) + return retval + + def get_device_by_mac(self, mac: str) -> Optional[device.Device]: + if mac in self._keywords_by_mac: + name = self._names_by_mac[mac] + kws = self._keywords_by_mac[mac] + logger.debug(f'Found {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: + 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: + 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 a known smart home device, returning None') + return None + + def query(self, query: str) -> List[device.Device]: + """Evaluates a lighting query expression formed of keywords to search + for, logical operators (and, or, not), and parenthesis. + Returns a list of matching lights. + """ + retval = [] + logger.debug(f'Executing query {query}') + results = self._corpus.query(query) + if results is not None: + for mac in results: + if mac is not None: + device = self.get_device_by_mac(mac) + if device is not None: + retval.append(device) + return retval -- cgit v1.3