commit 0ec147ed02a7fd8dced3cf1c91f75db99da71c4e
parent 3fbc103d37013f40cb3576366e3390e0e5cc08d6
Author: Brian C. Lane <bcl@brianlane.com>
Date: Sun, 16 Jul 2017 08:49:34 -0700
Add type annotations and cleanup pylint errors
Diffstat:
7 files changed, 64 insertions(+), 40 deletions(-)
diff --git a/src/bin/strix b/src/bin/strix
@@ -16,7 +16,7 @@
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
-import strix
+from strix import run
if __name__ == '__main__':
- strix.run()
+ run()
diff --git a/src/strix/__init__.py b/src/strix/__init__.py
@@ -19,24 +19,25 @@ import re
import time
import threading
-import strix.cmdline
-import strix.queue
-import strix.motion as motion
+from . import cmdline
+from . import queue
+from . import motion
## Check the motion args
## Start the queue watcher thread
## Start the bottle/API thread
## Wait for a signal to shutdown
+from typing import List, Match, Tuple
-def check_motion_config(config_path):
+def check_motion_config(config_path: str) -> Tuple[str, List[str]]:
""" 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 = []
+ errors = [] # type: List[str]
base_target_dir = ""
base_queue_dir = ""
found_pf = False
@@ -46,24 +47,24 @@ def check_motion_config(config_path):
# If there are threads, check their settings instead
for c in [cfg.config] + \
[cfg.thread[cc] for cc in filter(lambda k: k.startswith("thread"), cfg.config.keys())]:
- if c.get("picture_filename") == picture_filename:
+ if c.get("picture_filename", "") == picture_filename:
found_pf = True
- if c.get("on_event_end"):
- m = re.match(on_event_end_re, c.get("on_event_end"))
- if not m:
- continue
- if not m.groups():
+
+ on_event_end = c.get("on_event_end", "")
+ if on_event_end:
+ m = re.match(on_event_end_re, on_event_end) # type: Match[str]
+ if not m or not m.groups():
continue
# Above errors will be caught by not having base_queue_dir set.
if not base_queue_dir:
base_queue_dir = m.group(1)
elif base_queue_dir != m.group(1):
errors += ["All of the paths in on_event_end MUST match."]
- if c.get("target_dir"):
- m = re.match(target_dir_re, c.get("target_dir"))
- if not m:
- continue
- if not m.groups():
+
+ target_dir = c.get("target_dir", "")
+ if target_dir:
+ m = re.match(target_dir_re, target_dir)
+ if not m or not m.groups():
continue
# Above errors will be caught by not having base_target_dir set.
if not base_target_dir:
@@ -83,17 +84,17 @@ def check_motion_config(config_path):
return (base_target_dir, errors)
-def run():
- parser = strix.cmdline.parser()
+def run() -> bool:
+ parser = cmdline.parser()
opts = parser.parse_args()
try:
(base_dir, errors) = check_motion_config(opts.config)
except Exception as e:
- errors = [e]
+ errors = [str(e)]
if errors:
- def p_e(e):
+ def p_e(e: str) -> None:
print("ERROR: %s" % e)
list(map(p_e, errors))
return False
@@ -104,7 +105,7 @@ def run():
return False
queue_quit = threading.Event()
queue_thread = threading.Thread(name="queue-thread",
- target=strix.queue.monitor_queue,
+ target=queue.monitor_queue,
args=(queue_path,queue_quit))
queue_thread.start()
@@ -117,10 +118,12 @@ def run():
print("Exiting due to ^C")
# Tell the threads to quit
- for t in [queue_quit]:
- t.set()
+ for event in [queue_quit]:
+ event.set()
# Wait until everything is done
print("Waiting for threads to quit")
- for t in [queue_thread]:
- t.join()
+ for thread in [queue_thread]:
+ thread.join()
+
+ return True
diff --git a/src/strix/api.py b/src/strix/api.py
@@ -17,5 +17,5 @@
import threading
from bottle import route, run
-class Api(threading.Threading):
+class Api(threading.Thread):
pass
diff --git a/src/strix/cmdline.py b/src/strix/cmdline.py
@@ -18,7 +18,7 @@ import argparse
version = "DEVEL"
-def parser():
+def parser() -> argparse.ArgumentParser:
""" Return the ArgumentParser"""
parser = argparse.ArgumentParser(description="Motion Camera Web Interface")
diff --git a/src/strix/motion/__init__.py b/src/strix/motion/__init__.py
@@ -16,4 +16,4 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from . import config
-__all__ = [config]
+__all__ = ["config"]
diff --git a/src/strix/motion/config.py b/src/strix/motion/config.py
@@ -1,15 +1,33 @@
+# strix/motion/config.py
+#
+# Copyright (C) 2017 Brian C. Lane
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# 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 = {}
- thread = {}
+ config = {} # type: Dict
+ thread = {} # type: Dict
_thread_n = 0
- def thread_n(self):
+ def thread_n(self) -> str:
""" Return an incrementing thread-N string
:returns: str
@@ -17,7 +35,7 @@ class MotionConfig():
self._thread_n += 1
return "thread-%d" % self._thread_n
- def split(self, s):
+ def split(self, s: str) -> Tuple[str, str]:
""" Split the line into key and optional values.
:returns: (k, v) where v may be ""
@@ -33,7 +51,7 @@ class MotionConfig():
v = ""
return (k, v)
- def parse(self, config_path):
+ def parse(self, config_path: str) -> Dict:
""" Parse a motion config file
:returns: dict
@@ -44,7 +62,7 @@ class MotionConfig():
for line in f.readlines()
if line.strip() and not line.startswith("#")])
- def __init__(self, config_path):
+ def __init__(self, config_path: str) -> None:
self.config = self.parse(config_path)
for t in filter(lambda k: k.startswith("thread"), self.config.keys()):
thread_path = self.config[t]
diff --git a/src/strix/queue.py b/src/strix/queue.py
@@ -15,13 +15,16 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from glob import glob
+import multiprocessing as mp
import os
import time
-import multiprocessing as mp
+import threading
+
+from typing import List
## Handle watching the queue and dispatching movie creation and directory moving
-def process_event(event):
+def process_event(event: str) -> None:
print("Processing %s" % event)
# Convert event to a path, replace _ with /
@@ -32,8 +35,8 @@ def process_event(event):
# Create a thumbnail
# Move the directory to its final location
-def monitor_queue(queue_path, quit):
- threads = []
+def monitor_queue(queue_path: str, quit: threading.Event) -> None:
+ threads = [] # type: List[mp.Process]
while not quit.is_set():
time.sleep(5)
