From 889f57512986d1b799582c58d712b304b706391c Mon Sep 17 00:00:00 2001 From: Scott Gasch Date: Thu, 28 Oct 2021 21:23:55 -0700 Subject: [PATCH] Add cameras, fix bugs --- smart_home/cameras.py | 39 +++++++++++++++++++++++++++++++ smart_home/config.py | 54 ++++++++++++++++++++++++++++++++----------- 2 files changed, 79 insertions(+), 14 deletions(-) create mode 100644 smart_home/cameras.py diff --git a/smart_home/cameras.py b/smart_home/cameras.py new file mode 100644 index 0000000..963f54e --- /dev/null +++ b/smart_home/cameras.py @@ -0,0 +1,39 @@ +#!/usr/bin/env python3 + +"""Utilities for dealing with the webcams.""" + +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__) + +class BaseCamera(dev.Device): + camera_mapping = { + 'cabin_drivewaycam': 'cabin_driveway', + 'outside_backyard_camera': 'backyard', + 'outside_driveway_camera': 'driveway', + 'outside_doorbell_camera': 'doorbell', + 'outside_front_door_camera': 'frontdoor', + } + + def __init__(self, name: str, mac: str, keywords: str = "") -> None: + super().__init__(name.strip(), mac.strip(), keywords) + self.camera_name = BaseCamera.camera_mapping.get(name, None) + + def get_stream_url(self) -> str: + assert self.camera_name is not None + return f'http://10.0.0.56:81/mjpg/{self.camera_name}/video.mjpg?h=1024&q=99' diff --git a/smart_home/config.py b/smart_home/config.py index 723e097..a28caa7 100644 --- a/smart_home/config.py +++ b/smart_home/config.py @@ -1,5 +1,6 @@ #!/usr/bin/env python3 +import logging import re from typing import List, Optional, Set @@ -8,6 +9,7 @@ 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 @@ -24,47 +26,57 @@ parser.add_argument( ) +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() - self._macs_by_name = {} - self._keywords_by_name = {} - self._keywords_by_mac = {} - self._names_by_mac = {} - self._corpus = logical_search.Corpus() - + # 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 not f in keywords: - continue - - 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) + 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)] @@ -81,6 +93,7 @@ class SmartHomeConfig(object): properties=properties, reference=None, ) + logger.debug(f'Indexing document {device}') self._corpus.add_doc(device) def __repr__(self) -> str: @@ -125,27 +138,39 @@ class SmartHomeConfig(object): 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]: @@ -154,6 +179,7 @@ class SmartHomeConfig(object): 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: -- 2.45.2