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

Used mypy, pydocstyle, pycodestyle, sonarlint, removed py 2.7, and renamed.

Browse Source
This commit is contained in:
simleek
2019-09-30 22:25:45 -07:00
parent 6d45a81936
commit eaa5fe13e0
35 changed files with 894 additions and 741 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
tests/test_pub_cam.py
View File

@ -1,4 +1,4 @@
import cvpubsubs.webcam_pub as w
import displayarray.webcam_pub as w
import unittest as ut
@ -6,15 +6,17 @@ class TestFrameHandler(ut.TestCase):
i = 0
def test_handler(self):
def test_frame_handler(frame, cam_id):
if self.i == 200:
w.CamCtrl.stop_cam(cam_id)
w.camctrl.stop_cam(cam_id)
if self.i % 100 == 0:
print(frame.shape)
self.i += 1
w.VideoHandlerThread(0, [test_frame_handler],
request_size=(1280, 720),
high_speed=True,
fps_limit=240)
w.VideoHandlerThread(
0,
[test_frame_handler],
request_size=(1280, 720),
high_speed=True,
fps_limit=240,
)

44
tests/test_simple_api.py
View File

@ -1,61 +1,69 @@
import unittest as ut
class TestSubWin(ut.TestCase):
class TestSubWin(ut.TestCase):
def test_display_numpy(self):
from cvpubsubs import display
from displayarray import display
import numpy as np
display(np.random.normal(0.5, .1, (500,500,3)))
s, vids = display(np.random.normal(0.5, 0.1, (500, 500, 3)))
s.end()
print("ended")
def test_display_numpy_callback(self):
from cvpubsubs import display
from displayarray import display
import numpy as np
arr = np.random.normal(0.5, .1, (500, 500, 3))
arr = np.random.normal(0.5, 0.1, (500, 500, 3))
def fix_arr_cv(arr_in):
arr_in[:] += np.random.normal(0.01, .005, (500, 500, 3))
arr_in%=1.0
arr_in[:] += np.random.normal(0.01, 0.005, (500, 500, 3))
arr_in %= 1.0
display(arr, callbacks= fix_arr_cv, blocking=True)
display(arr, callbacks=fix_arr_cv, blocking=True)
def test_display_numpy_loop(self):
from cvpubsubs import display
from displayarray import display
import numpy as np
arr = np.random.normal(0.5, .1, (500, 500, 3))
arr = np.random.normal(0.5, 0.1, (500, 500, 3))
displayer, ids = display(arr, blocking = False)
displayer, ids = display(arr, blocking=False)
while True:
arr[:] += np.random.normal(0.01, .005, (500, 500, 3))
arr[:] += np.random.normal(0.01, 0.005, (500, 500, 3))
arr %= 1.0
displayer.update(arr, ids[0])
displayer.end()
def test_display_tensorflow(self):
from cvpubsubs import display
from displayarray import display
import numpy as np
from tensorflow.keras import layers, models
import tensorflow as tf
for gpu in tf.config.experimental.list_physical_devices("GPU"):
tf.compat.v2.config.experimental.set_memory_growth(gpu, True)
#tf.keras.backend.set_floatx("float16")
displayer, ids = display(0, blocking = False)
displayer, ids = display(0, blocking=False)
displayer.wait_for_init()
autoencoder = models.Sequential()
autoencoder.add(
layers.Conv2D(20, (3, 3), activation="sigmoid", input_shape=displayer.frames[0].shape)
layers.Conv2D(
20, (3, 3), activation="sigmoid", input_shape=displayer.frames[0].shape
)
)
autoencoder.add(layers.Conv2DTranspose(3, (3, 3), activation="sigmoid"))
autoencoder.compile(loss="mse", optimizer="adam")
while True:
grab = tf.convert_to_tensor(displayer.frame_dict['0frame'][np.newaxis, ...].astype(np.float32)/255.0)
grab = tf.convert_to_tensor(
displayer.FRAME_DICT["0frame"][np.newaxis, ...].astype(np.float32)
/ 255.0
)
autoencoder.fit(grab, grab, steps_per_epoch=1, epochs=1)
output_image = autoencoder.predict(grab, steps=1)
displayer.update((output_image[0]*255.0).astype(np.uint8), "uid for autoencoder output")
displayer.update(
(output_image[0] * 255.0).astype(np.uint8), "uid for autoencoder output"
)

77
tests/test_sub_win.py
View File

@ -1,20 +1,15 @@
import threading
import unittest as ut
import cvpubsubs.webcam_pub as w
from cvpubsubs.window_sub import SubscriberWindows
from cvpubsubs.window_sub.winctrl import WinCtrl
from cvpubsubs import display
from cvpubsubs.input import mouse_loop, key_loop
import numpy as np
import displayarray.webcam_pub as w
from displayarray.window_sub import SubscriberWindows
from displayarray import display
from displayarray.input import mouse_loop, key_loop
if False:
import numpy as np
from cvpubsubs.window_sub.mouse_event import MouseEvent
import numpy as np
from displayarray.window_sub.mouse_event import MouseEvent
class TestSubWin(ut.TestCase):
def test_mouse_loop(self):
@mouse_loop
def print_mouse_thread(mouse_event):
@ -41,11 +36,9 @@ class TestSubWin(ut.TestCase):
w.VideoHandlerThread(video_source=img, request_size=(300, -1)).display()
def test_sub_with_args(self):
video_thread = w.VideoHandlerThread(video_source=0,
request_size=(800, 600),
high_speed=False,
fps_limit=8
)
video_thread = w.VideoHandlerThread(
video_source=0, request_size=(800, 600), high_speed=False, fps_limit=8
)
video_thread.display()
@ -67,10 +60,10 @@ class TestSubWin(ut.TestCase):
self.assertEqual(v.exception_raised, e)
def test_multi_cams_one_source(self):
display(0, window_names=['cammy','cammy2'], blocking=True)
display(0, window_names=["cammy", "cammy2"], blocking=True)
def test_multi_cams_multi_source(self):
display(0, np.random.uniform(0.0, 1.0, (500,500)), blocking=True)
display(0, np.random.uniform(0.0, 1.0, (500, 500)), blocking=True)
def test_nested_frames(self):
def nest_frame(frame):
@ -84,28 +77,33 @@ class TestSubWin(ut.TestCase):
frame = np.asarray([[[[[[frame + 1 / 0]]]]], [[[[[frame]]], [[[frame]]]]]])
return frame
v = w.VideoHandlerThread(callbacks=[nest_frame] + w.display_callbacks)
v = w.VideoHandlerThread(callbacks=[nest_frame])
v.start()
with self.assertRaises(ZeroDivisionError) as e:
SubscriberWindows(window_names=[str(i) for i in range(3)],
video_sources=[str(0)]
).loop()
SubscriberWindows(
window_names=[str(i) for i in range(3)], video_sources=[str(0)]
).loop()
self.assertEqual(v.exception_raised, e)
v.join()
def test_conway_life(self):
from cvpubsubs.webcam_pub import VideoHandlerThread
from cvpubsubs.callbacks import function_display_callback
from displayarray.webcam_pub import VideoHandlerThread
from displayarray.callbacks import function_display_callback
import numpy as np
import cv2
img = np.zeros((50, 50, 1))
img[0:5, 0:5, :] = 1
def conway(array, coords, finished):
neighbors = np.sum(array[max(coords[0] - 1, 0):min(coords[0] + 2, 50),
max(coords[1] - 1, 0):min(coords[1] + 2, 50)])
neighbors = np.sum(
array[
max(coords[0] - 1, 0) : min(coords[0] + 2, 50),
max(coords[1] - 1, 0) : min(coords[1] + 2, 50),
]
)
neighbors = max(neighbors - np.sum(array[coords[0:2]]), 0.0)
if array[coords] == 1.0:
if neighbors < 2 or neighbors > 3:
@ -117,15 +115,26 @@ class TestSubWin(ut.TestCase):
array[coords] = 1.0
@mouse_loop
def conway_add(mouse_event # type:MouseEvent
):
def conway_add(
mouse_event # type:MouseEvent
):
if 0 <= mouse_event.x < 50 and 0 <= mouse_event.y < 50:
if mouse_event.flags == cv2.EVENT_FLAG_LBUTTON:
img[mouse_event.y - 5:mouse_event.y + 10, mouse_event.x - 5:mouse_event.x + 10, :] = 0.0
img[
mouse_event.y - 5 : mouse_event.y + 10,
mouse_event.x - 5 : mouse_event.x + 10,
:,
] = 0.0
elif mouse_event.flags == cv2.EVENT_FLAG_RBUTTON:
img[mouse_event.y - 5:mouse_event.y + 10, mouse_event.x - 5:mouse_event.x + 10, :] = 1.0
img[
mouse_event.y - 5 : mouse_event.y + 10,
mouse_event.x - 5 : mouse_event.x + 10,
:,
] = 1.0
VideoHandlerThread(video_source=img, callbacks=function_display_callback(conway)).display()
VideoHandlerThread(
video_source=img, callbacks=function_display_callback(conway)
).display()
def test_double_win(self):
vid1 = np.ones((100, 100))
@ -134,9 +143,9 @@ class TestSubWin(ut.TestCase):
t2 = w.VideoHandlerThread(vid2)
t1.start()
t2.start()
SubscriberWindows(window_names=['cammy', 'cammy2'],
video_sources=[vid1, vid2]
).loop()
SubscriberWindows(
window_names=["cammy", "cammy2"], video_sources=[vid1, vid2]
).loop()
t1.join()
t1.join()