--- /dev/null
+../pyproject.toml
\ No newline at end of file
--- /dev/null
+../pyproject.toml
\ No newline at end of file
from contextlib import contextmanager
from functools import wraps
from multiprocessing import RLock, shared_memory
-from typing import Any, Dict, Generator, ItemsView, Iterator, KeysView, Optional, ValuesView
+from typing import (
+ Any,
+ Dict,
+ Generator,
+ ItemsView,
+ Iterator,
+ KeysView,
+ Optional,
+ ValuesView,
+)
from decorator_utils import synchronized
from dateparse.dateparse_utilsLexer import dateparse_utilsLexer # type: ignore
from dateparse.dateparse_utilsListener import dateparse_utilsListener # type: ignore
from dateparse.dateparse_utilsParser import dateparse_utilsParser # type: ignore
-from datetime_utils import TimeUnit, date_to_datetime, datetime_to_date, n_timeunits_from_base
+from datetime_utils import (
+ TimeUnit,
+ date_to_datetime,
+ datetime_to_date,
+ n_timeunits_from_base,
+)
logger = logging.getLogger(__name__)
--- /dev/null
+../pyproject.toml
\ No newline at end of file
global executors / worker pools with automatic shutdown semantics."""
from __future__ import annotations
-
import concurrent.futures as fut
import logging
import os
"""File-based locking helper."""
from __future__ import annotations
-
import contextlib
import datetime
import json
searching."""
from __future__ import annotations
-
import enum
import sys
from collections import defaultdict
--- /dev/null
+../pyproject.toml
\ No newline at end of file
import glob
import logging
import os
-from typing import Callable, List, NamedTuple, Optional, Set
import warnings
+from typing import Callable, List, NamedTuple, Optional, Set
import argparse_utils
import config
elif in_spec.image_file_prepopulated_list is not None:
images += in_spec.image_file_prepopulated_list
else:
- raise ValueError(
- 'One of image_file_glob or image_file_prepopulated_list is required'
- )
+ raise ValueError('One of image_file_glob or image_file_prepopulated_list is required')
skip_list = read_skip_list()
for image in images:
profile="black"
sections="FUTURE,STDLIB,THIRDPARTY,FIRSTPARTY,LOCALFOLDER"
no_lines_before="STDLIB"
+known_first_party = [
+ "acl", "ansi", "argparse_utils", "arper", "base_presence", "bootstrap", "camera_utils",
+ "config", "constants", "conversion_utils", "datetime_utils", "decorator_utils", "deferred_operand",
+ "dict_utils", "directory_filter", "exceptions", "exec_utils", "executors", "file_utils",
+ "function_utils", "google_assistant", "histogram", "id_generator", "input_utils", "letter_compress",
+ "list_utils", "lockfile", "logging_utils", "logical_search", "math_utils", "misc_utils",
+ "orb_utils", "parallelize", "persistent", "profanity_filter", "remote_worker", "scott_secrets",
+ "site_config", "smart_future", "state_tracker", "stopwatch", "string_utils", "text_utils",
+ "thread_utils", "type_utils", "unittest_utils", "unscrambler", "waitable_presence",
+ "smart_home.cameras", "smart_home.chromecasts", "smart_home.device_utils", "smart_home.device",
+ "smart_home.lights", "smart_home.outlets", "smart_home.registry", "smart_home.thermometers",
+ "cached.weather_data", "cached.weather_forecast", "collect.bidict", "collect.bst",
+ "collect.shared_dict", "collect.trie", "dateparse.dateparse_utils", "ml.model_trainer",
+ "ml.quick_label", "type.centcount", "type.locations", "type.money", "type.people",
+ "type.rate",
+]
[tool.pylint]
[tool.pylint.MASTER]
# --enable=similarities". If you want to run only the classes checker, but have
# no Warning level messages displayed, use "--disable=all --enable=classes
# --disable=W".
-disable="invalid-name, missing-function-docstring, useless-object-inheritance, unspecified-encoding, import-outside-toplevel, raw-checker-failed, bad-inline-option, locally-disabled, line-too-long, too-few-public-methods, file-ignored, suppressed-message, useless-suppression, no-else-return, no-else-raise, no-else-break, deprecated-pragma, unnecessary-pass, use-symbolic-message-instead, broad-except"
+disable="invalid-name, missing-function-docstring, useless-object-inheritance, unspecified-encoding, import-outside-toplevel, raw-checker-failed, bad-inline-option, locally-disabled, line-too-long, too-few-public-methods, file-ignored, suppressed-message, useless-suppression, no-else-return, no-else-raise, no-else-break, deprecated-pragma, unnecessary-pass, use-symbolic-message-instead, broad-except, wrong-import-order"
# Enable the message, report, category or checker with the given id(s). You can
# either give multiple identifier separated by comma (,) or put this option
available."""
from __future__ import annotations
-
import concurrent
import concurrent.futures as fut
import logging
class BaseCamera(dev.Device):
+ """A base class for a webcam device."""
+
camera_mapping = {
'cabin_drivewaycam': 'cabin_driveway',
'outside_backyard_camera': 'backyard',
if name == 'driveway':
return 'http://10.0.0.226:8080/Umtxxf1uKMBniFblqeQ9KRbb6DDzN4/mjpeg/GKlT2FfiSQ/driveway'
else:
- return f'http://10.0.0.226:8080/Umtxxf1uKMBniFblqeQ9KRbb6DDzN4/mp4/GKlT2FfiSQ/{name}/s.mp4'
+ return (
+ f'http://10.0.0.226:8080/Umtxxf1uKMBniFblqeQ9KRbb6DDzN4/mp4/GKlT2FfiSQ/{name}/s.mp4'
+ )
class BaseChromecast(dev.Device):
+ """A base class to represent a Google Chromecase device."""
+
ccasts: List[Any] = []
refresh_ts = None
browser = None
self.cast = None
for cc in BaseChromecast.ccasts:
if cc.cast_info.host == ip and cc.cast_info.cast_type != 'group':
- logger.debug(f'Found chromecast at {ip}: {cc}')
+ logger.debug('Found chromecast at %s: %s', 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
#!/usr/bin/env python3
+"""General utility functions involving smart home devices."""
+
import logging
from typing import Any
-import smart_home.cameras as cameras
-import smart_home.chromecasts as chromecasts
-import smart_home.lights as lights
-import smart_home.outlets as outlets
+from smart_home import cameras, chromecasts, lights, outlets
logger = logging.getLogger(__name__)
logger.warning(msg)
logging_utils.hlog(msg)
return False
- logger.debug(f'{command} succeeded.')
+ logger.debug('%s succeeded.', command)
return True
class BaseOutlet(dev.Device):
+ """An abstract base class for smart outlets."""
+
def __init__(self, name: str, mac: str, keywords: str = "") -> None:
super().__init__(name.strip(), mac.strip(), keywords)
class TPLinkOutlet(BaseOutlet):
+ """A TPLink smart outlet."""
+
def __init__(self, name: str, mac: str, keywords: str = '') -> None:
super().__init__(name, mac, keywords)
self.info: Optional[Dict] = None
class TPLinkOutletWithChildren(TPLinkOutlet):
+ """A TPLink outlet where the top and bottom plus are individually
+ controllable."""
+
def __init__(self, name: str, mac: str, keywords: str = "") -> None:
super().__init__(name, mac, keywords)
self.children: List[str] = []
cmd = self.get_cmdline(child) + f"-c {cmd}"
if extra_args is not None:
cmd += f" {extra_args}"
- logger.debug(f'About to execute {cmd}')
+ logger.debug('About to execute: %s', cmd)
return tplink_outlet_command(cmd)
def get_children(self) -> List[str]:
class GoogleOutlet(BaseOutlet):
+ """A smart outlet controlled via Google Assistant."""
+
def __init__(self, name: str, mac: str, keywords: str = "") -> None:
super().__init__(name.strip(), mac.strip(), keywords)
self.info = None
class MerossOutlet(BaseOutlet):
+ """A Meross smart outlet class."""
+
def __init__(self, name: str, mac: str, keywords: str = '') -> None:
super().__init__(name, mac, keywords)
self.meross_wrapper: Optional[MerossWrapper] = None
--- /dev/null
+../pyproject.toml
\ No newline at end of file
#!/usr/bin/env python3
+"""Code involving querying various smart home thermometers."""
+
import logging
import urllib.request
from typing import Optional
class ThermometerRegistry(object):
+ """A registry of thermometer hosts / names and how to talk with them."""
+
def __init__(self):
self.thermometers = {
'house_outside': ('10.0.0.75', 'outside_temp'),
record = self.thermometers.get(location, None)
if record is None:
logger.error(
- f'Location {location} is not known. Valid locations are {self.thermometers.keys()}.'
+ 'Location %s is not known. Valid locations are %s.',
+ location,
+ self.thermometers.keys(),
)
return None
url = f'http://{record[0]}/~pi/{record[1]}'
- logger.debug(f'Constructed URL: {url}')
+ logger.debug('Constructed URL: %s', url)
try:
www = urllib.request.urlopen(url, timeout=3)
temp = www.read().decode('utf-8')
temp = round(temp)
except Exception as e:
logger.exception(e)
- logger.error(f'Failed to read temperature at URL: {url}')
+ logger.error('Failed to read temperature at URL: %s', url)
temp = None
finally:
if www is not None:
import unicodedata
import warnings
from itertools import zip_longest
-from typing import Any, Callable, Dict, Iterable, List, Literal, Optional, Sequence, Tuple
+from typing import (
+ Any,
+ Callable,
+ Dict,
+ Iterable,
+ List,
+ Literal,
+ Optional,
+ Sequence,
+ Tuple,
+)
from uuid import uuid4
import list_utils
--- /dev/null
+../pyproject.toml
\ No newline at end of file
import unittest
-from collect.shared_dict import SharedDict
import parallelize as p
import smart_future
import unittest_utils
+from collect.shared_dict import SharedDict
class SharedDictTest(unittest.TestCase):
--- /dev/null
+../pyproject.toml
\ No newline at end of file
projects / python_utils.git / commitdiff