commit c45bf0cdd0865168699997e4f8e493da59fd29e9
parent 7c3d854058c815df87f1db01e94ee5a3f60e8a37
Author: Brian C. Lane <bcl@brianlane.com>
Date: Sun, 20 Mar 2022 12:21:06 -0700
mypy: Drop attempts to use mypy types
I'd rather focus on unit testing and coverage than fighting with a
bolted-on type system.
This removes the types, removes mypy from tox, and also makes tox
install the test dependencies from setup.cfg
Diffstat:
9 files changed, 31 insertions(+), 50 deletions(-)
diff --git a/setup.cfg b/setup.cfg
@@ -15,10 +15,8 @@ packages = find:
install_requires =
bottle
gevent
- mypy
- Pillow
- rfc3339
structlog
+ Pillow
[options.packages.find]
where = src
@@ -29,7 +27,6 @@ strix = ui/*
[options.extras_require]
testing =
coverage
- mypy
nose
pylint
pytest
diff --git a/src/strix/__init__.py b/src/strix/__init__.py
@@ -31,16 +31,14 @@ from . import motion
## Start the bottle/API thread
## Wait for a signal to shutdown
-from typing import List, Match, Optional, Tuple, Any
-
-def check_motion_config(config_path: str) -> Tuple[str, List[str]]:
+def check_motion_config(config_path):
""" Check the config file to make sure the settings match what Strix needs.
"""
picture_filename = "%Y-%m-%d/%v/%H-%M-%S-%q"
on_event_end_re = r".*touch (.*)/queue/Camera%t_%Y-%m-%d_%v"
target_dir_re = r"(.*)/Camera\d+"
- errors = [] # type: List[str]
+ errors = []
base_target_dir = ""
base_queue_dir = ""
found_pf = False
@@ -55,7 +53,7 @@ def check_motion_config(config_path: str) -> Tuple[str, List[str]]:
on_event_end = c.get("on_event_end", "")
if on_event_end:
- em = re.match(on_event_end_re, on_event_end) # type: Optional[Match[str]]
+ em = re.match(on_event_end_re, on_event_end)
if not em or not em.groups():
continue
# Above errors will be caught by not having base_queue_dir set.
@@ -66,7 +64,7 @@ def check_motion_config(config_path: str) -> Tuple[str, List[str]]:
target_dir = c.get("target_dir", "")
if target_dir:
- tm = re.match(target_dir_re, target_dir) # type: Optional[Match[str]]
+ tm = re.match(target_dir_re, target_dir)
if not tm or not tm.groups():
continue
# Above errors will be caught by not having base_target_dir set.
@@ -87,7 +85,7 @@ def check_motion_config(config_path: str) -> Tuple[str, List[str]]:
return (base_target_dir, errors)
-def run() -> bool:
+def run():
parser = cmdline.parser(queue.max_cores())
opts = parser.parse_args()
@@ -97,13 +95,13 @@ def run() -> bool:
errors = [str(e)]
if errors:
- def p_e(e: str) -> None:
+ def p_e(e):
print("ERROR: %s" % e)
list(map(p_e, errors))
return False
# Start logger thread
- logger_queue = mp.JoinableQueue() # type: mp.JoinableQueue[List[Any]]
+ logger_queue = mp.JoinableQueue()
logger_quit = mp.Event()
logger_thread = mp.Process(name="logger-thread",
target=logger.listener,
diff --git a/src/strix/api.py b/src/strix/api.py
@@ -30,21 +30,21 @@ from . import logger
from .events import camera_events, EventCache
TIME_FORMAT = "%Y-%m-%d %H:%M:%S"
-def timestr_to_dt(rfc_str: str) -> datetime:
+def timestr_to_dt(rfc_str):
return datetime.strptime(rfc_str, TIME_FORMAT)
-def run_api(logging_queue: mp.Queue, base_dir: str, host: str, port: int, debug: bool) -> None:
+def run_api(logging_queue, base_dir, host, port, debug):
log = logger.log(logging_queue)
log.info("Starting API", base_dir=base_dir, host=host, port=port, debug=debug)
EventCache.logger(log)
@route('/')
@route('/<filename>')
- def serve_root(filename: str = "index.html") -> Response:
+ def serve_root(filename="index.html"):
return static_file(filename, root=os.path.dirname(__file__)+"/ui")
@route('/motion/<filepath:path>')
- def serve_motion(filepath: str) -> Response:
+ def serve_motion(filepath):
return static_file(filepath, root=base_dir)
@route('/api/cameras/list')
@@ -52,7 +52,9 @@ def run_api(logging_queue: mp.Queue, base_dir: str, host: str, port: int, debug:
return {"cameras": sorted([d for d in os.listdir(base_dir) if d != "queue"])}
@route('/api/events/<cameras>')
- def serve_events(cameras: str) -> Response:
+ def serve_events(cameras):
+ # request.query is a bottle.MultiDict which pylint doesn't understand
+ # pylint: disable=no-member
start = timestr_to_dt(request.query.get("start", "1985-10-26 01:22:00"))
end = timestr_to_dt(request.query.get("end", datetime.now().strftime(TIME_FORMAT)))
offset= int(request.query.get("offset", "0"))
diff --git a/src/strix/cmdline.py b/src/strix/cmdline.py
@@ -18,7 +18,7 @@ import argparse
version = "DEVEL"
-def parser(max_cores) -> argparse.ArgumentParser:
+def parser(max_cores):
""" Return the ArgumentParser"""
parser = argparse.ArgumentParser(description="Motion Camera Web Interface")
diff --git a/src/strix/events.py b/src/strix/events.py
@@ -26,8 +26,6 @@ import threading
import structlog
-from typing import Dict, List
-
class EventCacheClass:
def __init__(self):
self._log = None
@@ -44,10 +42,6 @@ class EventCacheClass:
return self._cache[key]
def set(self, key, value):
- # DEBUG
- if self._log:
- self._log.info(f"EventCache: {key} = {value}")
-
with self._lock:
# Convert start/end to datetime object
if "start" in value and type(value["start"]) == type(""):
@@ -177,14 +171,14 @@ def preload_cache(log, base_dir):
EventCache.force_expire(False)
-def path_to_dt(path: str) -> datetime:
+def path_to_dt(path):
# Use the last 2 elements of the path to construct a Datatime
(date, time) = path.split("/")[-2:]
time = time.replace("-", ":")
dt_str = "{0} {1}".format(date, time)
return datetime.strptime(dt_str, "%Y-%m-%d %H:%M:%S")
-def image_to_dt(event_date: str, image: str) -> datetime:
+def image_to_dt(event_date, image):
""" Convert an event date (YYYY-MM-DD) and image HH-MM-SS-FF
returns a datetime object
@@ -194,9 +188,7 @@ def image_to_dt(event_date: str, image: str) -> datetime:
return datetime.strptime(event_date+"/"+image_time, "%Y-%m-%d/%H-%M-%S")
-def event_details(log: structlog.BoundLogger, event_path: str) -> Dict:
-# log.info("event_details", path=event_path)
-
+def event_details(log, event_path):
# Check the cache for the details
try:
return EventCache.get(event_path)
@@ -257,9 +249,6 @@ def event_details(log: structlog.BoundLogger, event_path: str) -> Dict:
is_saved = os.path.exists(event_path+"/.saved")
-# log.debug("event_details", thumbnail=thumbnail, start_time=str(start_time), end_time=str(end_time),
-# video=video, debug_video=debug_video, saved=is_saved)
-
details = {
"start": start_time,
"end": end_time,
@@ -281,15 +270,14 @@ def event_details(log: structlog.BoundLogger, event_path: str) -> Dict:
return details
-def camera_events(log: structlog.BoundLogger, base_dir: str, camera: str,
- start: datetime, end: datetime, offset: int, limit: int) -> List[Dict]:
+def camera_events(log, base_dir, camera, start, end, offset, limit):
# YYYY-MM-DD/HH-MM-SS is the format of the event directories.
glob_path="%s/%s/????-??-??/??-??-??" % (base_dir, camera)
# Newest to oldest, limited by offset and limit
skipped = 0
added = 0
- events = [] # type: List[Dict]
+ events = []
for event_path in sorted(glob(glob_path), reverse=True):
dt = path_to_dt(event_path)
if dt < start or dt > end:
diff --git a/src/strix/logger.py b/src/strix/logger.py
@@ -39,7 +39,7 @@ structlog.configure(
cache_logger_on_first_use=True,
)
-def listener(queue: mp.Queue, stop_event: mp.Event, log_path: str) -> None:
+def listener(queue, stop_event, log_path):
handler = RotatingFileHandler(log_path, maxBytes=100*1024**2, backupCount=10)
formatter = logging.Formatter('%(message)s')
handler.setFormatter(formatter)
@@ -60,7 +60,7 @@ def listener(queue: mp.Queue, stop_event: mp.Event, log_path: str) -> None:
import sys, traceback
traceback.print_exc(file=sys.stderr)
-def log(queue: mp.Queue) -> structlog.BoundLogger:
+def log(queue):
handler = QueueHandler(queue)
root = structlog.get_logger()
root.addHandler(handler)
diff --git a/src/strix/motion/config.py b/src/strix/motion/config.py
@@ -16,15 +16,13 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import os
-from typing import Dict, Tuple
-
class MotionConfig():
""" Parse a motion configuration file into dicts
Last key wins. Threads are stored in self.thread["thread-N"]
"""
- config = {} # type: Dict
- thread = {} # type: Dict
+ config = {}
+ thread = {}
_thread_n = 0
def thread_n(self) -> str:
@@ -35,7 +33,7 @@ class MotionConfig():
self._thread_n += 1
return "thread-%d" % self._thread_n
- def split(self, s: str) -> Tuple[str, str]:
+ def split(self, s):
""" Split the line into key and optional values.
:returns: (k, v) where v may be ""
@@ -51,7 +49,7 @@ class MotionConfig():
v = ""
return (k, v)
- def parse(self, config_path: str) -> Dict:
+ def parse(self, config_path):
""" Parse a motion config file
:returns: dict
@@ -62,7 +60,7 @@ class MotionConfig():
for line in f.readlines()
if line.strip() and not line.startswith("#")])
- def __init__(self, config_path: str) -> None:
+ def __init__(self, config_path):
self.config = self.parse(config_path)
for t in filter(lambda k: k.startswith("thread") or k.startswith("camera"), self.config.keys()):
thread_path = self.config[t]
diff --git a/src/strix/queue.py b/src/strix/queue.py
@@ -24,8 +24,6 @@ import time
from PIL import Image
import structlog
-from typing import List
-
from . import logger
THUMBNAIL_SIZE = (640, 480)
@@ -98,8 +96,8 @@ def process_event(log: structlog.BoundLogger, base_dir: str, event: str) -> None
except Exception as e:
log.error("Moving to destination failed", event_path=event_path, exception=str(e))
-def monitor_queue(logging_queue: mp.Queue, base_dir: str, quit: mp.Event, max_threads: int) -> None:
- threads = [] # type: List[mp.Process]
+def monitor_queue(logging_queue, base_dir, quit, max_threads):
+ threads = []
log = logger.log(logging_queue)
queue_path = os.path.abspath(os.path.join(base_dir, "queue/"))
diff --git a/tox.ini b/tox.ini
@@ -2,8 +2,8 @@
envlist = py38
[testenv]
+deps=.[testing]
commands=
- mypy --strict --ignore-missing-imports src/strix/ src/bin/strix
pylint --rcfile=pylint.rc -E src/strix/ src/bin/strix
nosetests --with-coverage [] # substitute with tox' positional arguments
# pytest -v
