Reorganized cv_pubsubs to match PEP8 fully. Modified rgc_dev accordingly.

__init__.py

@@ -0,0 +1,2 @@
+from .cv_webcam_pub import *
+from .sub_win import sub_win_loop, frameDict

cv_webcam_pub.py

@@ -1,107 +0,0 @@
-import pubsub
-import queue
-import cv2
-import numpy as np
-import threading
-
-if False:  # don't include if actually running
-    from typing import List, Union, Tuple, Any, Callable
-
-
-def get_open_cv_cam_ids():  # type: () -> List[cv2.VideoCapture]
-    cam_list = []  # type: List[int]
-
-    while True:
-        cam = cv2.VideoCapture(len(cam_list))
-        if not cam.isOpened():
-            break
-        cam_list.append(len(cam_list))
-
-    return cam_list
-import time
-
-def pub_cv_cam_thread(camId,  # type: Union[int, str]
-                      requestSize=(1280, 720)  # type: Tuple[int, int]
-                      ):
-    sub = pubsub.subscribe("cvcams."+str(camId)+".cmd")
-    msg = ''
-    cam = cv2.VideoCapture(camId)
-    cam.set(cv2.CAP_PROP_FRAME_WIDTH, requestSize[0])
-    cam.set(cv2.CAP_PROP_FRAME_HEIGHT, requestSize[1])
-    if not cam.isOpened():
-        pubsub.publish("cvcams." + str(camId) + ".status", "failed")
-        return False
-    now = time.time()
-    while (  msg != 'q'):
-        time.sleep(1./(60-(time.time()-now)))
-        now = time.time()
-        (ret, frame) = cam.read()  # type: Tuple[bool, np.ndarray ]
-        if ret is False or not isinstance(frame, np.ndarray):
-            cam.release()
-            pubsub.publish("cvcams." + str(camId) + ".status", "failed")
-            return False
-        pubsub.publish("cvcams."+str(camId)+".vid", (frame,))
-        msg = listenFixed(sub, block=False, empty='')
-
-        pass
-    cam.release()
-    return True
-
-def listenFixed(sub, block=True, timeout=None, empty=None):
-    try:
-        msg = (sub.listen(block=block, timeout=timeout))
-        try:
-            msg = next(msg)['data']
-        except StopIteration:
-            msg = empty
-    except queue.Empty:
-        msg = empty
-    return msg
-
-
-def init_cv_cam_pub_thread(camId, # type: Union[int, str]
-                           requestSize=(1280, 720)  # type: Tuple[int, int]
-                      ):
-    # type: (...) -> threading.Thread
-    t = threading.Thread(target=pub_cv_cam_thread, args = (camId,requestSize))
-    t.start()
-    return t
-
-def cv_cam_pub_handler(camId, # type: Union[int, str]
-                       frameHandler, # type: Callable[[int, np.ndarray], Any]
-                       requestSize=(1280, 720)  # type: Tuple[int, int]
-                       ):
-    t = init_cv_cam_pub_thread(camId, requestSize)
-    subCam = pubsub.subscribe("cvcams."+str(camId)+".vid")
-    subOwner = pubsub.subscribe("cvcamhandlers."+str(camId)+".cmd")
-    msgOwner = ''
-    while msgOwner != 'q':
-        frame = listenFixed(subCam, timeout=.1)  # type: np.ndarray
-        if frame is not None:
-            frame = frame[0]
-            frameHandler(frame, camId)
-        msgOwner = listenFixed(subOwner, block=False, empty='')
-    pubsub.publish("cvcams."+str(camId)+".cmd", 'q')
-    t.join()
-
-def init_cv_cam_pub_handler(camId, # type: Union[int, str]
-                            frameHandler, # type: Callable[[int, np.ndarray], Any]
-                            requestSize=(1280, 720)  # type: Tuple[int, int]
-                      ):
-    # type: (...) -> threading.Thread
-    t = threading.Thread(target=cv_cam_pub_handler, args = (camId, frameHandler, requestSize))
-    t.start()
-    return t
-
-if __name__ == '__main__': # todo: add to tests
-    i = 0
-    def testFrameHandler(frame, camId):
-        global i
-        if i == 200:
-            pubsub.publish("cvcamhandlers."+str(camId)+".cmd", 'q')
-        if i % 100 == 0:
-            print(frame.shape)
-        i += 1
-
-    cv_cam_pub_handler(0, testFrameHandler)
-

cv_webcam_pub/__init__.py

@@ -0,0 +1,5 @@
+from .listen_default import _listen_default
+from .get_cam_ids import get_cam_ids
+from .pub_cam import pub_cam_thread
+from .frame_handler import frame_handler_thread
+from .cam_ctrl import cam_ctrl

cv_webcam_pub/cam_ctrl.py

@@ -0,0 +1,10 @@
+import pubsub
+if False:
+    from typing import Union
+
+class cam_ctrl:
+
+    @staticmethod
+    def stop_cam(cam_id  # type: Union[int, str]
+                 ):
+        pubsub.publish("cvcamhandlers." + str(cam_id) + ".cmd", 'q')

cv_webcam_pub/frame_handler.py

@@ -0,0 +1,33 @@
+import pubsub
+import numpy as np
+import threading
+from .listen_default import _listen_default
+from .pub_cam import pub_cam_thread
+
+if False:
+    from typing import Union, Tuple, Any, Callable
+
+def frame_handler_loop(cam_id,  # type: Union[int, str]
+                       frame_handler,  # type: Callable[[int, np.ndarray], Any]
+                       request_size=(1280, 720)  # type: Tuple[int, int]
+                       ):
+    t = pub_cam_thread(cam_id, request_size)
+    sub_cam = pubsub.subscribe("cvcams." + str(cam_id) + ".vid")
+    sub_owner = pubsub.subscribe("cvcamhandlers." + str(cam_id) + ".cmd")
+    msg_owner = ''
+    while msg_owner != 'q':
+        frame = _listen_default(sub_cam, timeout=.1)  # type: np.ndarray
+        if frame is not None:
+            frame = frame[0]
+            frame_handler(frame, cam_id)
+        msg_owner = _listen_default(sub_owner, block=False, empty='')
+    pubsub.publish("cvcams." + str(cam_id) + ".cmd", 'q')
+    t.join()
+
+def frame_handler_thread(cam_id,  # type: Union[int, str]
+                            frame_handler,  # type: Callable[[int, np.ndarray], Any]
+                            request_size=(1280, 720)  # type: Tuple[int, int]
+                         ):                        # type: (...) -> threading.Thread
+    t = threading.Thread(target=frame_handler_loop, args=(cam_id, frame_handler, request_size))
+    t.start()
+    return t

cv_webcam_pub/get_cam_ids.py

@@ -0,0 +1,18 @@
+import cv2
+
+if False:
+    from typing import List
+
+
+def get_cam_ids():  # type: () -> List[int]
+    cam_list = []
+
+    while True:
+        cam = cv2.VideoCapture(len(cam_list))
+        if not cam.isOpened():
+            break
+        cam_list.append(len(cam_list))
+
+    return cam_list
+
+

cv_webcam_pub/listen_default.py

@@ -0,0 +1,17 @@
+if False:
+    from typing import Any, Optional, queue
+
+def _listen_default(sub,           # type: queue
+                    block=True,    # type: bool
+                    timeout=None,  # type: Optional[float]
+                    empty=None     # type: Any
+                    ):             # type: (...)->Any
+    try:
+        msg = (sub.listen(block=block, timeout=timeout))
+        try:
+            msg = next(msg)['data']
+        except StopIteration:
+            msg = empty
+    except queue.Empty:
+        msg = empty
+    return msg

cv_webcam_pub/pub_cam.py

@@ -0,0 +1,53 @@
+import pubsub
+import cv2
+import numpy as np
+import time
+import threading
+from .listen_default import _listen_default
+
+if False:
+    from typing import Union, Tuple
+
+def pub_cam_loop(cam_id,  # type: Union[int, str]
+                      request_size=(1280, 720),  # type: Tuple[int, int]
+                      fps_limit = 60
+                      ):  # type: (...)->bool
+    """
+
+
+    :param cam_id:
+    :param request_size:
+    :return:
+    """
+    sub = pubsub.subscribe("cvcams." + str(cam_id) + ".cmd")
+    msg = ''
+    cam = cv2.VideoCapture(cam_id)
+    cam.set(cv2.CAP_PROP_FRAME_WIDTH, request_size[0])
+    cam.set(cv2.CAP_PROP_FRAME_HEIGHT, request_size[1])
+    if not cam.isOpened():
+        pubsub.publish("cvcams." + str(cam_id) + ".status", "failed")
+        return False
+    now = time.time()
+    while msg != 'q':
+        time.sleep(1. / (fps_limit - (time.time() - now)))
+        now = time.time()
+        (ret, frame) = cam.read()  # type: Tuple[bool, np.ndarray ]
+        if ret is False or not isinstance(frame, np.ndarray):
+            cam.release()
+            pubsub.publish("cvcams." + str(cam_id) + ".status", "failed")
+            return False
+        pubsub.publish("cvcams." + str(cam_id) + ".vid", (frame,))
+        msg = _listen_default(sub, block=False, empty='')
+
+        pass
+    cam.release()
+    return True
+
+def pub_cam_thread(cam_id,                  # type: Union[int, str]
+                           request_ize=(1280, 720)  # type: Tuple[int, int]
+                           ):
+    # type: (...) -> threading.Thread
+    t = threading.Thread(target=pub_cam_loop, args=(cam_id, request_ize))
+    t.start()
+    return t
+

cv_window_sub.py

@@ -1,62 +0,0 @@
-import cv2, pubsub
-
-if False:
-    from typing import List
-
-cvWindows = []
-frameDict={}
-import time
-def cv_win_sub(*,
-               names,  # type: List[str]
-               inputVidGlobalNames,  # type: List[str]
-               callbacks=(None,)
-               ):
-    global cvWindows
-
-    global frameDict
-    firstRun=True
-    timeMod = None
-    while True:
-        t = int(time.time()) * 1000
-        if firstRun:
-            timeMod = t % 1000
-            firstRun = False
-        #global camImg
-        for i in range(len(inputVidGlobalNames)):
-            if inputVidGlobalNames[i] in frameDict and frameDict[inputVidGlobalNames[i]] is not None:
-                if callbacks[i%len(callbacks)] is not None:
-                    frames = callbacks[i%len(callbacks)](frameDict[inputVidGlobalNames[i]])
-                else:
-                    frames = frameDict[inputVidGlobalNames[i]]
-                for i in range(len(frames)):
-                    if t % 1000 == timeMod:
-                        if names[i%len(names)]+str(i) not in cvWindows:
-                            cvWindows.append(names[i%len(names)]+str(i))
-                            cv2.namedWindow(names[i%len(names)]+str(i))
-                    cv2.imshow(names[i%len(names)]+str(i), frames[i])
-            if cv2.waitKey(1)& 0xFF==ord('q'):
-                for name in cvWindows:
-                    cv2.destroyWindow(name)
-                for n in names:
-                    pubsub.publish("cvcamhandlers." + str(n) + ".cmd", "q")
-
-                return
-
-
-camImg = None
-if __name__ == '__main__':
-    import importlib
-    import os, sys
-    sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
-
-    cv_webcam_pub = importlib.import_module('cv_webcam_pub')
-
-    def camHandler(frame, camId):
-        frameDict[str(camId)+"Frame"]= frame
-
-    t = cv_webcam_pub.init_cv_cam_pub_handler(0, camHandler)
-
-    cv_win_sub(names=['cammy', 'cammy2'], inputVidGlobalNames=[str(0)+"Frame"])
-
-    pubsub.publish("cvcamhandlers.0.cmd", 'q')
-    t.join()

sub_win.py

@@ -0,0 +1,40 @@
+import cv2
+from .cv_webcam_pub.cam_ctrl import cam_ctrl
+
+if False:
+    from typing import List
+
+cvWindows = []
+frameDict = {}
+
+
+def sub_win_loop(*,
+                 names,  # type: List[str]
+                 input_vid_global_names,  # type: List[str]
+                 callbacks=(None,)
+                 ):
+    global cvWindows
+    global frameDict
+
+    first_run = True
+    while True:
+        for i in range(len(input_vid_global_names)):
+            if input_vid_global_names[i] in frameDict and frameDict[input_vid_global_names[i]] is not None:
+                if callbacks[i % len(callbacks)] is not None:
+                    frames = callbacks[i % len(callbacks)](frameDict[input_vid_global_names[i]])
+                else:
+                    frames = frameDict[input_vid_global_names[i]]
+                for f in range(len(frames)):
+                    if first_run:
+                        if names[f % len(names)] not in cvWindows:
+                            cvWindows.append(names[f % len(names)])
+                            cv2.namedWindow(names[f % len(names)])
+                    cv2.imshow(names[f % len(names)], frames[f])
+            if cv2.waitKey(1) & 0xFF == ord('q'):
+                for name in cvWindows:
+                    cv2.destroyWindow(name)
+                for n in names:
+                    cam_ctrl.stop_cam(n)
+
+                return
+        first_run = False

tests_interactive/test_pub_cam.py

@@ -0,0 +1,17 @@
+import cv_pubsubs.cv_webcam_pub as w
+import unittest as ut
+
+
+class TestFrameHandler(ut.TestCase):
+    i = 0
+
+    def test_handler(self):
+
+        def test_frame_handler(frame, cam_id):
+            if self.i == 200:
+                w.cam_ctrl.stop_cam(cam_id)
+            if self.i % 100 == 0:
+                print(frame.shape)
+            self.i += 1
+
+        w.frame_handler_thread(0, test_frame_handler)

tests_interactive/test_sub_win.py

@@ -0,0 +1,17 @@
+import unittest as ut
+import cv_pubsubs.cv_webcam_pub as w
+from cv_pubsubs.sub_win import frameDict, sub_win_loop
+
+class TestSubWin(ut.TestCase):
+
+    def test_sub(self):
+        def camHandler(frame, camId):
+            frameDict[str(camId) + "Frame"] = (frame,frame)
+
+        t = w.frame_handler_thread(0, camHandler)
+
+        sub_win_loop(names=['cammy', 'cammy2'], input_vid_global_names=[str(0)+"Frame"])
+
+        w.cam_ctrl.stop_cam(0)
+
+        t.join()