Add pytest-tornado

requirements-dev.txt

@@ -28,12 +28,13 @@ pyparsing==2.4.7          # via packaging
 pytest-flake8==1.0.6      # via streamz-opencv
 pytest-forked==1.2.0      # via pytest-xdist
 pytest-mypy==0.6.2        # via streamz-opencv
+pytest-tornado==0.8.1     # via streamz-opencv
 pytest-xdist==1.33.0      # via streamz-opencv
-pytest==5.4.3             # via pytest-flake8, pytest-forked, pytest-mypy, pytest-xdist, streamz-opencv
+pytest==5.4.3             # via pytest-flake8, pytest-forked, pytest-mypy, pytest-tornado, pytest-xdist, streamz-opencv
 six==1.15.0               # via packaging, pytest-xdist, streamz
 streamz==0.5.4            # via streamz-opencv
 toolz==0.10.0             # via streamz
-tornado==6.0.4            # via streamz
+tornado==6.0.4            # via pytest-tornado, streamz
 typed-ast==1.4.1          # via mypy
 typing-extensions==3.7.4.2  # via mypy
 wcwidth==0.2.5            # via pytest

setup.py

@@ -16,6 +16,7 @@
             "pytest",
             "pytest-flake8",
             "pytest-mypy",
+            "pytest-tornado",
             "pytest-xdist",
             "flake8",
             "flake8-import-order",

streamz_opencv/__init__.py

@@ -0,0 +1,43 @@
+import cv2
+from streamz.core import Source, Stream
+from tornado import gen
+
+
+@Stream.register_api(staticmethod)
+class from_opencv(Source):
+    """ Stream data from opencv """
+    def __init__(self, src: str, poll_interval=0.1, start=False, **kwargs):
+        self.src = src
+        self.poll_interval = poll_interval
+
+        super(from_opencv, self).__init__(ensure_io_loop=True, **kwargs)
+
+        self.stopped = True
+        self.started = False
+
+        if start:
+            self.start()
+
+    def start(self):
+        if self.stopped:
+            self.stopped = False
+            self.started = True
+            self.stream = cv2.VideoCapture(self.src)
+            self.loop.add_callback(self.poll_opencv)
+
+    def do_poll(self):
+        grabbed, frame = self.stream.read()
+        if grabbed:
+            return frame
+
+    @gen.coroutine
+    def poll_opencv(self):
+        while True:
+            frame = self.do_poll()
+            if frame:
+                yield self._emit(frame)
+            else:
+                yield gen.sleep(self.poll_interval)
+            if self.stopped:
+                break
+        self.stream.release()

tests/test_webcam.py

@@ -0,0 +1,11 @@
+from streamz import Stream
+from tornado import gen
+
+from streamz_opencv import from_opencv
+
+
+def test_from_webcam():
+    stream = Stream.from_opencv("0", asynchronous=True)
+    out = stream.sink_to_list()
+    stream.start()
+    yield gen.sleep(0.1)