summaryrefslogtreecommitdiff
path: root/smart_home
diff options
context:
space:
mode:
Diffstat (limited to 'smart_home')
-rw-r--r--smart_home/chromecasts.py29
-rw-r--r--smart_home/device.py10
-rw-r--r--smart_home/lights.py39
-rw-r--r--smart_home/outlets.py71
-rw-r--r--smart_home/registry.py18
5 files changed, 89 insertions, 78 deletions
diff --git a/smart_home/chromecasts.py b/smart_home/chromecasts.py
index a5db86f..bd2a80c 100644
--- a/smart_home/chromecasts.py
+++ b/smart_home/chromecasts.py
@@ -6,17 +6,18 @@ import atexit
import datetime
import logging
import threading
+from typing import Any, List
import pychromecast
-from decorator_utils import memoized
import smart_home.device as dev
+from decorator_utils import memoized
logger = logging.getLogger(__name__)
class BaseChromecast(dev.Device):
- ccasts = []
+ ccasts: List[Any] = []
refresh_ts = None
browser = None
lock = threading.Lock()
@@ -25,31 +26,32 @@ class BaseChromecast(dev.Device):
super().__init__(name.strip(), mac.strip(), keywords)
ip = self.get_ip()
now = datetime.datetime.now()
- with BaseChromecast.lock as l:
+ with BaseChromecast.lock:
if (
- BaseChromecast.refresh_ts is None
- or (now - BaseChromecast.refresh_ts).total_seconds() > 60
+ BaseChromecast.refresh_ts is None
+ or (now - BaseChromecast.refresh_ts).total_seconds() > 60
):
logger.debug('Refreshing the shared chromecast info list')
if BaseChromecast.browser is not None:
BaseChromecast.browser.stop_discovery()
- BaseChromecast.ccasts, BaseChromecast.browser = pychromecast.get_chromecasts(
- timeout=15.0
- )
+ (
+ BaseChromecast.ccasts,
+ BaseChromecast.browser,
+ ) = pychromecast.get_chromecasts(timeout=15.0)
+ assert BaseChromecast.browser
atexit.register(BaseChromecast.browser.stop_discovery)
BaseChromecast.refresh_ts = now
self.cast = None
for cc in BaseChromecast.ccasts:
- if (
- cc.cast_info.host == ip
- and cc.cast_info.cast_type != 'group'
- ):
+ if cc.cast_info.host == ip and cc.cast_info.cast_type != 'group':
logger.debug(f'Found chromecast at {ip}: {cc}')
self.cast = cc
self.cast.wait(timeout=1.0)
if self.cast is None:
- raise Exception(f'Can\'t find ccast device at {ip}, is that really a ccast device?')
+ raise Exception(
+ f'Can\'t find ccast device at {ip}, is that really a ccast device?'
+ )
def is_idle(self):
return self.cast.is_idle
@@ -116,4 +118,3 @@ class BaseChromecast(dev.Device):
f"Chromecast({self.cast.socket_client.host!r}, port={self.cast.socket_client.port!r}, "
f"device={self.cast.cast_info.friendly_name!r})"
)
-
diff --git a/smart_home/device.py b/smart_home/device.py
index 9675b7c..02717a3 100644
--- a/smart_home/device.py
+++ b/smart_home/device.py
@@ -8,17 +8,17 @@ import arper
class Device(object):
def __init__(
- self,
- name: str,
- mac: str,
- keywords: Optional[List[str]],
+ self,
+ name: str,
+ mac: str,
+ keywords: Optional[List[str]],
):
self.name = name
self.mac = mac
self.keywords = keywords
self.arper = arper.Arper()
if keywords is not None:
- self.kws = keywords.split()
+ self.kws = keywords
else:
self.kws = []
diff --git a/smart_home/lights.py b/smart_home/lights.py
index 64f2105..240e7da 100644
--- a/smart_home/lights.py
+++ b/smart_home/lights.py
@@ -2,7 +2,6 @@
"""Utilities for dealing with the smart lights."""
-from abc import abstractmethod
import datetime
import json
import logging
@@ -10,10 +9,11 @@ import os
import re
import subprocess
import sys
+from abc import abstractmethod
from typing import Any, Dict, List, Optional, Tuple
-from overrides import overrides
import tinytuya as tt
+from overrides import overrides
import ansi
import argparse_utils
@@ -21,8 +21,8 @@ import arper
import config
import logging_utils
import smart_home.device as dev
-from google_assistant import ask_google, GoogleResponse
-from decorator_utils import timeout, memoized
+from decorator_utils import memoized, timeout
+from google_assistant import GoogleResponse, ask_google
logger = logging.getLogger(__name__)
@@ -39,9 +39,7 @@ args.add_argument(
)
-@timeout(
- 5.0, use_signals=False, error_message="Timed out waiting for tplink.py"
-)
+@timeout(5.0, use_signals=False, error_message="Timed out waiting for tplink.py")
def tplink_light_command(command: str) -> bool:
result = os.system(command)
signal = result & 0xFF
@@ -69,9 +67,9 @@ class BaseLight(dev.Device):
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
+ color,
)
- if m is not None and len(m.group) == 3:
+ if m is not None and len(m.groups()) == 3:
red = int(m.group(0), 16)
green = int(m.group(1), 16)
blue = int(m.group(2), 16)
@@ -147,7 +145,9 @@ class GoogleLight(BaseLight):
r = ask_google(f"is {self.goog_name()} on?")
if not r.success:
return False
- return 'is on' in r.audio_transcription
+ if r.audio_transcription is not None:
+ return 'is on' in r.audio_transcription
+ raise Exception("Can't reach Google?!")
@overrides
def is_off(self) -> bool:
@@ -163,11 +163,12 @@ class GoogleLight(BaseLight):
# the bookcase one is set to 40% bright
txt = r.audio_transcription
- m = re.search(r"(\d+)% bright", txt)
- if m is not None:
- return int(m.group(1))
- if "is off" in txt:
- return 0
+ if txt is not None:
+ m = re.search(r"(\d+)% bright", txt)
+ if m is not None:
+ return int(m.group(1))
+ if "is off" in txt:
+ return 0
return None
@overrides
@@ -301,9 +302,7 @@ class TPLinkLight(BaseLight):
def get_children(self) -> List[str]:
return self.children
- def command(
- self, cmd: str, child: str = None, extra_args: str = None
- ) -> bool:
+ 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}"
@@ -333,9 +332,7 @@ class TPLinkLight(BaseLight):
def make_color(self, color: str) -> bool:
raise NotImplementedError
- @timeout(
- 10.0, use_signals=False, error_message="Timed out waiting for tplink.py"
- )
+ @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)
diff --git a/smart_home/outlets.py b/smart_home/outlets.py
index c079cfd..d4a4886 100644
--- a/smart_home/outlets.py
+++ b/smart_home/outlets.py
@@ -2,7 +2,6 @@
"""Utilities for dealing with the smart outlets."""
-from abc import abstractmethod
import asyncio
import atexit
import datetime
@@ -12,10 +11,12 @@ import os
import re
import subprocess
import sys
+from abc import abstractmethod
from typing import Any, Dict, List, Optional
from meross_iot.http_api import MerossHttpClient
from meross_iot.manager import MerossManager
+from overrides import overrides
import argparse_utils
import config
@@ -23,8 +24,8 @@ import decorator_utils
import logging_utils
import scott_secrets
import smart_home.device as dev
-from google_assistant import ask_google, GoogleResponse
-from decorator_utils import timeout, memoized
+from decorator_utils import memoized, timeout
+from google_assistant import GoogleResponse, ask_google
logger = logging.getLogger(__name__)
@@ -41,9 +42,7 @@ parser.add_argument(
)
-@timeout(
- 5.0, use_signals=False, error_message="Timed out waiting for tplink.py"
-)
+@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
@@ -66,7 +65,6 @@ def tplink_outlet_command(command: str) -> bool:
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:
@@ -105,27 +103,29 @@ class TPLinkOutlet(BaseOutlet):
)
return cmd
- def command(self, cmd: str, extra_args: str = None) -> bool:
+ def command(self, cmd: str, extra_args: str = None, **kwargs) -> bool:
cmd = self.get_cmdline() + f"-c {cmd}"
if extra_args is not None:
cmd += f" {extra_args}"
return tplink_outlet_command(cmd)
+ @overrides
def turn_on(self) -> bool:
return self.command('on')
+ @overrides
def turn_off(self) -> bool:
return self.command('off')
+ @overrides
def is_on(self) -> bool:
return self.get_on_duration_seconds() > 0
+ @overrides
def is_off(self) -> bool:
return not self.is_on()
- @timeout(
- 10.0, use_signals=False, error_message="Timed out waiting for tplink.py"
- )
+ @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)
@@ -161,7 +161,7 @@ class TPLinkOutletWithChildren(TPLinkOutlet):
for child in self.info["children"]:
self.children.append(child["id"])
- # override
+ @overrides
def get_cmdline(self, child: Optional[str] = None) -> str:
cmd = (
f"{config.config['smart_outlets_tplink_location']} -m {self.mac} "
@@ -171,10 +171,9 @@ class TPLinkOutletWithChildren(TPLinkOutlet):
cmd += f"-x {child} "
return cmd
- # override
- def command(
- self, cmd: str, child: str = None, extra_args: str = None
- ) -> bool:
+ @overrides
+ def command(self, cmd: str, extra_args: str = None, **kwargs) -> bool:
+ child: Optional[str] = kwargs.get('child', None)
cmd = self.get_cmdline(child) + f"-c {cmd}"
if extra_args is not None:
cmd += f" {extra_args}"
@@ -184,9 +183,11 @@ class TPLinkOutletWithChildren(TPLinkOutlet):
def get_children(self) -> List[str]:
return self.children
+ @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)
@@ -218,22 +219,28 @@ class GoogleOutlet(BaseOutlet):
def parse_google_response(response: GoogleResponse) -> bool:
return response.success
+ @overrides
def turn_on(self) -> bool:
return GoogleOutlet.parse_google_response(
- ask_google('turn {self.goog_name()} on')
+ ask_google(f'turn {self.goog_name()} on')
)
+ @overrides
def turn_off(self) -> bool:
return GoogleOutlet.parse_google_response(
- ask_google('turn {self.goog_name()} off')
+ 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
+ if r.audio_transcription is not None:
+ return 'is on' in r.audio_transcription
+ raise Exception('Can\'t talk to Google right now!?')
+ @overrides
def is_off(self) -> bool:
return not self.is_on()
@@ -247,10 +254,13 @@ class MerossWrapper(object):
this class.
"""
+
def __init__(self):
self.loop = asyncio.get_event_loop()
self.email = os.environ.get('MEROSS_EMAIL') or scott_secrets.MEROSS_EMAIL
- self.password = os.environ.get('MEROSS_PASSWORD') or scott_secrets.MEROSS_PASSWORD
+ self.password = (
+ os.environ.get('MEROSS_PASSWORD') or scott_secrets.MEROSS_PASSWORD
+ )
self.devices = self.loop.run_until_complete(self.find_meross_devices())
atexit.register(self.loop.close)
@@ -282,8 +292,8 @@ class MerossWrapper(object):
class MerossOutlet(BaseOutlet):
def __init__(self, name: str, mac: str, keywords: str = '') -> None:
super().__init__(name, mac, keywords)
- self.meross_wrapper = None
- self.device = None
+ self.meross_wrapper: Optional[MerossWrapper] = None
+ self.device: Optional[Any] = None
def lazy_initialize_device(self):
"""If we make too many calls to Meross they will block us; only talk
@@ -294,23 +304,28 @@ class MerossOutlet(BaseOutlet):
if self.device is None:
raise Exception(f'{self.name} is not a known Meross device?!')
+ @overrides
def turn_on(self) -> bool:
self.lazy_initialize_device()
- self.meross_wrapper.loop.run_until_complete(
- self.device.async_turn_on()
- )
+ assert self.meross_wrapper
+ assert self.device
+ self.meross_wrapper.loop.run_until_complete(self.device.async_turn_on())
return True
+ @overrides
def turn_off(self) -> bool:
self.lazy_initialize_device()
- self.meross_wrapper.loop.run_until_complete(
- self.device.async_turn_off()
- )
+ assert self.meross_wrapper
+ assert self.device
+ self.meross_wrapper.loop.run_until_complete(self.device.async_turn_off())
return True
+ @overrides
def is_on(self) -> bool:
self.lazy_initialize_device()
+ assert self.device
return self.device.is_on()
+ @overrides
def is_off(self) -> bool:
return not self.is_on()
diff --git a/smart_home/registry.py b/smart_home/registry.py
index 7349081..16e18ba 100644
--- a/smart_home/registry.py
+++ b/smart_home/registry.py
@@ -8,15 +8,15 @@ 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.device as device
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 related to the smart home configuration registry.",
)
args.add_argument(
'--smart_home_registry_file_location',
@@ -32,9 +32,9 @@ logger = logging.getLogger(__file__)
class SmartHomeRegistry(object):
def __init__(
- self,
- registry_file: Optional[str] = None,
- filters: List[str] = ['smart'],
+ self,
+ registry_file: Optional[str] = None,
+ filters: List[str] = ['smart'],
) -> None:
self._macs_by_name = {}
self._keywords_by_name = {}
@@ -44,13 +44,11 @@ class SmartHomeRegistry(object):
# Read the disk config file...
if registry_file is None:
- registry_file = config.config[
- 'smart_home_registry_file_location'
- ]
+ 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()
+ with open(registry_file, "r") as rf:
+ contents = rf.readlines()
# Parse the contents...
for line in contents:
generated by cgit v1.3 (git 2.53.0) at 2026-07-17 17:05:05 +0000