Merge pull request #26 from ReolinkCameraAPI/dev

adding blocking and non-blocking rtsp stream

RtspClient.py

@@ -1,11 +1,22 @@
 import os
+from threading import ThreadError
+
 import cv2
 
+from util import threaded
+
 
 class RtspClient:
-
-    def __init__(self, ip, username, password, port=554, profile="main", use_udp=True, **kwargs):
     """
+    Inspiration from:
+        - https://benhowell.github.io/guide/2015/03/09/opencv-and-web-cam-streaming
+        - https://stackoverflow.com/questions/19846332/python-threading-inside-a-class
+        - https://stackoverflow.com/questions/55828451/video-streaming-from-ip-camera-in-python-using-opencv-cv2-videocapture
+    """
+
+    def __init__(self, ip, username, password, port=554, profile="main", use_udp=True, callback=None, **kwargs):
+        """
+        RTSP client is used to retrieve frames from the camera in a stream
 
         :param ip: Camera IP
         :param username: Camera Username
@@ -15,6 +26,10 @@ class RtspClient:
         :param use_upd: True to use UDP, False to use TCP
         :param proxies: {"host": "localhost", "port": 8000}
         """
+        self.capture = None
+        self.thread_cancelled = False
+        self.callback = callback
+
         capture_options = 'rtsp_transport;'
         self.ip = ip
         self.username = username
@@ -30,18 +45,67 @@ class RtspClient:
 
         os.environ["OPENCV_FFMPEG_CAPTURE_OPTIONS"] = capture_options
 
-    def preview(self):
+        # opens the stream capture, but does not retrieve any frames yet.
-        """ Blocking function. Opens OpenCV window to display stream. """
+        self._open_video_capture()
-        win_name = self.ip
-        cap = cv2.VideoCapture(self.url, cv2.CAP_FFMPEG)
-        ret, frame = cap.read()
 
-        while ret:
+    def _open_video_capture(self):
-            cv2.imshow(win_name, frame)
+        # To CAP_FFMPEG or not To ?
+        self.capture = cv2.VideoCapture(self.url, cv2.CAP_FFMPEG)
 
-            ret, frame = cap.read()
+    def _stream_blocking(self):
-            if (cv2.waitKey(1) & 0xFF == ord('q')):
+        while True:
-                break
+            try:
+                if self.capture.isOpened():
+                    ret, frame = self.capture.read()
+                    if ret:
+                        yield frame
+                else:
+                    print("stream closed")
+                    self.capture.release()
+                    return
+            except Exception as e:
+                print(e)
+                self.capture.release()
+                return
 
-        cap.release()
+    @threaded
-        cv2.destroyAllWindows()
+    def _stream_non_blocking(self):
+        while not self.thread_cancelled:
+            try:
+                if self.capture.isOpened():
+                    ret, frame = self.capture.read()
+                    if ret:
+                        self.callback(frame)
+                else:
+                    print("stream is closed")
+                    self.stop_stream()
+            except ThreadError as e:
+                print(e)
+                self.stop_stream()
+
+    def stop_stream(self):
+        self.capture.release()
+        self.thread_cancelled = True
+
+    def open_stream(self):
+        """
+        Opens OpenCV Video stream and returns the result according to the OpenCV documentation
+        https://docs.opencv.org/3.4/d8/dfe/classcv_1_1VideoCapture.html#a473055e77dd7faa4d26d686226b292c1
+
+        :param callback: The function to callback the cv::mat frame to if required to be non-blocking. If this is left
+            as None, then the function returns a generator which is blocking.
+        """
+
+        # Reset the capture object
+        if self.capture is None or not self.capture.isOpened():
+            self._open_video_capture()
+
+        print("opening stream")
+
+        if self.callback is None:
+            return self._stream_blocking()
+        else:
+            # reset the thread status if the object was not re-created
+            if not self.thread_cancelled:
+                self.thread_cancelled = False
+            return self._stream_non_blocking()

api/__init__.py

@@ -1,4 +1,4 @@
 from .APIHandler import APIHandler
 
-__version__ = "0.0.4"
+__version__ = "0.0.5"
 VERSION = __version__

api/recording.py

@@ -9,6 +9,7 @@ from RtspClient import RtspClient
 
 class RecordingAPIMixin:
     """API calls for recording/streaming image or video."""
+
     def get_recording_encoding(self) -> object:
         """
         Get the current camera encoding settings for "Clear" and "Fluent" profiles.
@@ -72,15 +73,16 @@ class RecordingAPIMixin:
     ###########
     # RTSP Stream
     ###########
-    def open_video_stream(self, profile: str = "main", proxies=None) -> None:
+    def open_video_stream(self, callback=None, profile: str = "main", proxies=None):
         """
         'https://support.reolink.com/hc/en-us/articles/360007010473-How-to-Live-View-Reolink-Cameras-via-VLC-Media-Player'
+        Blocking function creates a generator and returns the frames as it is spawned
         :param profile: profile is "main" or "sub"
         :param proxies: Default is none, example: {"host": "localhost", "port": 8000}
         """
         rtsp_client = RtspClient(
-            ip=self.ip, username=self.username, password=self.password, proxies=proxies)
+            ip=self.ip, username=self.username, password=self.password, proxies=proxies, callback=callback)
-        rtsp_client.preview()
+        return rtsp_client.open_stream()
 
     def get_snap(self, timeout: int = 3, proxies=None) -> Image or None:
         """

examples/streaming_video.py

@@ -0,0 +1,82 @@
+import cv2
+
+from Camera import Camera
+
+
+def non_blocking():
+    print("calling non-blocking")
+
+    def inner_callback(img):
+        cv2.imshow("name", maintain_aspect_ratio_resize(img, width=600))
+        print("got the image non-blocking")
+        key = cv2.waitKey(1)
+        if key == ord('q'):
+            cv2.destroyAllWindows()
+            exit(1)
+
+    c = Camera("192.168.1.112", "admin", "jUa2kUzi")
+    # t in this case is a thread
+    t = c.open_video_stream(callback=inner_callback)
+
+    print(t.is_alive())
+    while True:
+        if not t.is_alive():
+            print("continuing")
+            break
+        # stop the stream
+        # client.stop_stream()
+
+
+def blocking():
+    c = Camera("192.168.1.112", "admin", "jUa2kUzi")
+    # stream in this case is a generator returning an image (in mat format)
+    stream = c.open_video_stream()
+
+    # using next()
+    # while True:
+    #     img = next(stream)
+    #     cv2.imshow("name", maintain_aspect_ratio_resize(img, width=600))
+    #     print("got the image blocking")
+    #     key = cv2.waitKey(1)
+    #     if key == ord('q'):
+    #         cv2.destroyAllWindows()
+    #         exit(1)
+
+    # or using a for loop
+    for img in stream:
+        cv2.imshow("name", maintain_aspect_ratio_resize(img, width=600))
+        print("got the image blocking")
+        key = cv2.waitKey(1)
+        if key == ord('q'):
+            cv2.destroyAllWindows()
+            exit(1)
+
+
+# Resizes a image and maintains aspect ratio
+def maintain_aspect_ratio_resize(image, width=None, height=None, inter=cv2.INTER_AREA):
+    # Grab the image size and initialize dimensions
+    dim = None
+    (h, w) = image.shape[:2]
+
+    # Return original image if no need to resize
+    if width is None and height is None:
+        return image
+
+    # We are resizing height if width is none
+    if width is None:
+        # Calculate the ratio of the height and construct the dimensions
+        r = height / float(h)
+        dim = (int(w * r), height)
+    # We are resizing width if height is none
+    else:
+        # Calculate the ratio of the 0idth and construct the dimensions
+        r = width / float(w)
+        dim = (width, int(h * r))
+
+    # Return the resized image
+    return cv2.resize(image, dim, interpolation=inter)
+
+
+# Call the methods. Either Blocking (using generator) or Non-Blocking using threads
+# non_blocking()
+blocking()

util.py

@@ -0,0 +1,11 @@
+from threading import Thread
+
+
+def threaded(fn):
+    def wrapper(*args, **kwargs):
+        thread = Thread(target=fn, args=args, kwargs=kwargs)
+        thread.daemon = True
+        thread.start()
+        return thread
+
+    return wrapper