SimLeek/displayarray
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Updated major version due to breaking changes. Reformatted files and imports. Removed unused ctrl commands. Made all pubsub channels camel case. changed sub_win_loop to class. Removed implementation specific code. Removed unreliable 'x to quit' code. Made windows quit from pubsub. Added WinCtrl class for window specific commands. Added threaded interactive key press test. Removed testt file.

Browse Source
This commit is contained in:
SimLeek
2018-04-08 14:12:02 -07:00
parent 496a5eaffe
commit 2b7899377e
12 changed files with 143 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cv_pubsubs/webcam_pub/listen_default.py → cv_pubsubs/listen_default.py
View File
cv_pubsubs/webcam_pub/__init__.py
+2 -3
View File
@@ -1,5 +1,4 @@
from .listen_default import listen_default
from .camctrl import CamCtrl
from .frame_handler import frame_handler_thread
from .get_cam_ids import get_cam_ids
from .pub_cam import pub_cam_thread
from .frame_handler import frame_handler_thread
from .camctrl import CamCtrl
cv_pubsubs/webcam_pub/camctrl.py
+1 -11
View File
@@ -9,14 +9,4 @@ class CamCtrl:
@staticmethod
def stop_cam(cam_id # type: Union[int, str]
):
pubsub.publish("cvcamhandlers." + str(cam_id) + ".cmd", 'q')
@staticmethod
def reset_vid(cam_id # type: Union[int, str]
):
pubsub.publish("cvcamhandlers." + str(cam_id) + ".cmd", 'r')
@staticmethod
def key_stroke(key_entered):
pubsub.publish("cvKeyStroke.", key_entered)
pubsub.publish("CVCamHandlers." + str(cam_id) + ".Cmd", 'quit')
cv_pubsubs/webcam_pub/frame_handler.py
+9 -7
View File
@@ -1,7 +1,9 @@
import pubsub
import numpy as np
import threading
from .listen_default import listen_default
import numpy as np
import pubsub
from cv_pubsubs.listen_default import listen_default
from .pub_cam import pub_cam_thread
if False:
@@ -15,16 +17,16 @@ def frame_handler_loop(cam_id, # type: Union[int, str]
fps_limit=240 # type: float
):
t = pub_cam_thread(cam_id, request_size, high_speed, fps_limit)
sub_cam = pubsub.subscribe("cvcams." + str(cam_id) + ".vid")
sub_owner = pubsub.subscribe("cvcamhandlers." + str(cam_id) + ".cmd")
sub_cam = pubsub.subscribe("CVCams." + str(cam_id) + ".Vid")
sub_owner = pubsub.subscribe("CVCamHandlers." + str(cam_id) + ".Cmd")
msg_owner = ''
while msg_owner != 'q':
while msg_owner != 'quit':
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')
pubsub.publish("CVCams." + str(cam_id) + ".Cmd", 'quit')
t.join()
cv_pubsubs/webcam_pub/pub_cam.py
+15 -13
View File
@@ -1,9 +1,11 @@
import pubsub
import threading
import time
import cv2
import numpy as np
import time
import threading
from .listen_default import listen_default
import pubsub
from cv_pubsubs.listen_default import listen_default
if False:
from typing import Union, Tuple
@@ -14,9 +16,9 @@ def pub_cam_loop(cam_id, # type: Union[int, str]
high_speed=False, # type: bool
fps_limit=240 # type: float
): # type: (...)->bool
"""Publishes whichever camera you select to cvcams.<cam_id>.vid
You can send a quit command 'q' to cvcams.<cam_id>.cmd
Status information, such as failure to open, will be posted to cvcams.<cam_id>.status
"""Publishes whichever camera you select to CVCams.<cam_id>.Vid
You can send a quit command 'quit' to CVCams.<cam_id>.Cmd
Status information, such as failure to open, will be posted to CVCams.<cam_id>.Status
:param high_speed: Selects mjpeg transferring, which most cameras seem to support, so speed isn't limited
@@ -25,7 +27,7 @@ def pub_cam_loop(cam_id, # type: Union[int, str]
:param request_size: A tuple with width, then height, to request the video size.
:return: True if loop ended normally, False if it failed somehow.
"""
sub = pubsub.subscribe("cvcams." + str(cam_id) + ".cmd")
sub = pubsub.subscribe("CVCams." + str(cam_id) + ".Cmd")
msg = ''
cam = cv2.VideoCapture(cam_id)
# cam.set(cv2.CAP_PROP_CONVERT_RGB, 0)
@@ -38,23 +40,23 @@ def pub_cam_loop(cam_id, # type: Union[int, str]
cam.set(cv2.CAP_PROP_FRAME_HEIGHT, request_size[1])
if not cam.isOpened():
pubsub.publish("cvcams." + str(cam_id) + ".status", "failed")
pubsub.publish("CVCams." + str(cam_id) + ".Status", "failed")
return False
now = time.time()
while msg != 'q':
while msg != 'quit':
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")
pubsub.publish("CVCams." + str(cam_id) + ".Status", "failed")
return False
if cam.get(cv2.CAP_PROP_FRAME_COUNT) > 0:
frame_counter+=1
frame_counter += 1
if frame_counter >= cam.get(cv2.CAP_PROP_FRAME_COUNT):
frame_counter = 0
cam = cv2.VideoCapture(cam_id)
pubsub.publish("cvcams." + str(cam_id) + ".vid", (frame,))
pubsub.publish("CVCams." + str(cam_id) + ".Vid", (frame,))
msg = listen_default(sub, block=False, empty='')
cam.release()
cv_pubsubs/window_sub/__init__.py
+1 -1
View File
@@ -1 +1 @@
from .cv_window_sub import sub_win_loop, frame_dict
from .cv_window_sub import SubscriberWindows
cv_pubsubs/window_sub/cv_window_sub.py
+50 -48
View File
@@ -1,64 +1,66 @@
import cv2
from ..webcam_pub.camctrl import CamCtrl
import warnings
import cv2
import pubsub
from .winctrl import WinCtrl
from ..listen_default import listen_default
from ..webcam_pub.camctrl import CamCtrl
if False:
from typing import List
frame_dict = {}
def triangle_seen():
print("a triangle was seen")
def square_seen():
print("a square was seen")
def nothing_seen():
print("nothing was seen")
class SubscriberWindows(object):
frame_dict = {}
command_dict = {
"t": triangle_seen,
"s": square_seen,
" ": nothing_seen
}
esc_key_codes = [27] # ESC key on most keyboards
# todo: figure out how to get the red x button to work. Try: https://stackoverflow.com/a/37881722/782170
def sub_win_loop(
names, # type: List[str]
def __init__(self,
window_names, # type: List[str]
input_vid_global_names, # type: List[str]
callbacks=(None,),
input_cams=(0,)
):
global frame_dict
self.window_names = window_names
self.input_vid_global_names = input_vid_global_names
self.callbacks = callbacks
self.input_cams = input_cams
while True:
for i in range(len(input_vid_global_names)):
if input_vid_global_names[i] in frame_dict and frame_dict[input_vid_global_names[i]] is not None:
if callbacks[i % len(callbacks)] is not None:
frames = callbacks[i % len(callbacks)](frame_dict[input_vid_global_names[i]])
def handle_keys(self,
key_input, # type: int
):
if key_input in self.esc_key_codes:
for name in self.window_names:
cv2.destroyWindow(name + " (press ESC to quit)")
for c in self.input_cams:
CamCtrl.stop_cam(c)
WinCtrl.quit()
elif key_input not in [-1, 0]:
try:
WinCtrl.key_stroke(chr(key_input))
except ValueError:
warnings.warn(
RuntimeWarning("Unknown key code: [{}]. Please report to cv_pubsubs issue page.".format(key_input))
)
def update_window_frames(self):
for i in range(len(self.input_vid_global_names)):
if self.input_vid_global_names[i] in self.frame_dict and self.frame_dict[
self.input_vid_global_names[i]] is not None:
if self.callbacks[i % len(self.callbacks)] is not None:
frames = self.callbacks[i % len(self.callbacks)](self.frame_dict[self.input_vid_global_names[i]])
else:
frames = frame_dict[input_vid_global_names[i]]
frames = self.frame_dict[self.input_vid_global_names[i]]
for f in range(len(frames)):
cv2.imshow(names[f % len(names)]+" (press q to quit)", frames[f])
if cv2.getWindowProperty(names[f % len(names)]+" (press q to quit)", 0) != 0:
print("X was pressed")
for name in names:
cv2.destroyWindow(name)
for c in input_cams:
CamCtrl.stop_cam(c)
key_criteria = cv2.waitKey(1) & 0xFF
if key_criteria == ord("q"):
for name in names:
cv2.destroyWindow(name)
for c in input_cams:
CamCtrl.stop_cam(c)
return
if chr(key_criteria) in command_dict:
command_dict[chr(key_criteria)]()
CamCtrl.key_stroke(chr(key_criteria))
elif chr(key_criteria) != "ÿ":
print(chr(key_criteria))
cv2.imshow(self.window_names[f % len(self.window_names)] + " (press ESC to quit)", frames[f])
# todo: figure out how to get the red x button to work. Try: https://stackoverflow.com/a/37881722/782170
def loop(self):
sub_cmd = pubsub.subscribe("CVWinCmd")
msg_cmd = ''
while msg_cmd != 'quit':
self.update_window_frames()
self.handle_keys(cv2.waitKey(1))
msg_cmd = listen_default(sub_cmd, block=False, empty='')
pubsub.publish("CVWinCmd", 'quit')
cv_pubsubs/window_sub/winctrl.py
+12
View File
@@ -0,0 +1,12 @@
import pubsub
class WinCtrl:
@staticmethod
def key_stroke(key_entered):
pubsub.publish("CVKeyStroke", key_entered)
@staticmethod
def quit():
pubsub.publish("CVWinCmd", "quit")
setup.py
+5 -2
View File
@@ -1,9 +1,10 @@
from distutils.core import setup
from setuptools import find_packages
setup(
name= 'cv_pubsubs',
packages = ['cv_pubsubs', 'cv_pubsubs.webcam_pub', 'cv_pubsubs.window_sub'],
version='0.1',
version='1.0.0',
packages = find_packages(),
description='Pubsub interface for Python OpenCV',
author='Josh Miklos',
author_email='simulatorleek@gmail.com',
@@ -16,5 +17,7 @@ setup(
'License :: OSI Approved :: MIT License',
'Programming Language :: Python :: 3',
'Programming Language :: Python :: 3.6',
'Programming Language :: Python :: 2',
'Programming Language :: Python :: 2.7',
]
)
tests_interactive/__init__.py
View File
tests_interactive/test_sub_win.py
+48 -12
View File
@@ -1,22 +1,36 @@
import threading
import unittest as ut
import cv_pubsubs.webcam_pub as w
from cv_pubsubs.window_sub import frame_dict, sub_win_loop
def subscribe_to_key_command(cam_id, # type: Union[int, str]
frame_handler, # type: Callable[[int, np.ndarray], Any]
request_size=(1280, 720), # type: Tuple[int, int]
high_speed=False, # type: bool
fps_limit=240 # type: float
): # type: (...) -> threading.Thread
t = threading.Thread(target=frame_handler_loop, args=(cam_id, frame_handler, request_size, high_speed, fps_limit))
import pubsub
import cv_pubsubs.webcam_pub as w
from cv_pubsubs.listen_default import listen_default
from cv_pubsubs.window_sub import SubscriberWindows
def print_keys_thread():
sub_key = pubsub.subscribe("CVKeyStroke")
sub_cmd = pubsub.subscribe("CVWinCmd")
msg_cmd = ''
while msg_cmd != 'quit':
key_chr = listen_default(sub_key, timeout=.1) # type: np.ndarray
if key_chr is not None:
print("key pressed: " + str(key_chr))
msg_cmd = listen_default(sub_cmd, block=False, empty='')
pubsub.publish("CVWinCmd", 'quit')
def start_print_keys_thread(): # type: (...) -> threading.Thread
t = threading.Thread(target=print_keys_thread, args=())
t.start()
return t
class TestSubWin(ut.TestCase):
def test_sub(self):
def cam_handler(frame, cam_id):
frame_dict[str(cam_id) + "Frame"] = (frame, frame)
SubscriberWindows.frame_dict[str(cam_id) + "Frame"] = (frame, frame)
t = w.frame_handler_thread(0, cam_handler,
request_size=(1280, 720),
@@ -24,9 +38,31 @@ class TestSubWin(ut.TestCase):
fps_limit=240
)
sub_win_loop(names=['cammy', 'cammy2'],
input_vid_global_names=[str(0) + "Frame"])
SubscriberWindows(window_names=['cammy', 'cammy2'],
input_vid_global_names=[str(0) + "Frame"]
).loop()
w.CamCtrl.stop_cam(0)
t.join()
def test_key_sub(self):
def cam_handler(frame, cam_id):
SubscriberWindows.frame_dict[str(cam_id) + "Frame"] = (frame, frame)
t = w.frame_handler_thread(0, cam_handler,
request_size=(1280, 720),
high_speed=True,
fps_limit=240
)
kt = start_print_keys_thread()
SubscriberWindows(window_names=['cammy', 'cammy2'],
input_vid_global_names=[str(0) + "Frame"]
).loop()
w.CamCtrl.stop_cam(0)
t.join()
kt.join()
testt.py
-8
View File
@@ -1,8 +0,0 @@
def hi():
print("hi")
if __name__ == '__main__':
dic = {}
dic['a'] = hi
dic['a']()