adding blocking and non-blocking rtsp stream

Benehiko · Benehiko · commit 5d03c62b39bd · 2020-12-04T22:27:50.000+02:00
This might fix the stream errors in issue #25 Also refactored the code a bit since we do not want to display a window from within the API (offload to the person implementing it).
diff --git a/RtspClient.py b/RtspClient.py
@@ -1,11 +1,22 @@
 import os
+from threading import ThreadError
+
 import cv2
 
+from util import threaded
+
 
 class RtspClient:
+    """
+    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, **kwargs):
+    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,33 +26,86 @@ def __init__(self, ip, username, password, port=554, profile="main", use_udp=Tru
         :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
         self.password = password
         self.port = port
         self.proxy = kwargs.get("proxies")
         self.url = "rtsp://" + self.username + ":" + self.password + "@" + \
-            self.ip + ":" + str(self.port) + "//h264Preview_01_" + profile
+                   self.ip + ":" + str(self.port) + "//h264Preview_01_" + profile
         if use_udp:
             capture_options = capture_options + 'udp'
         else:
             capture_options = capture_options + 'tcp'
 
         os.environ["OPENCV_FFMPEG_CAPTURE_OPTIONS"] = capture_options
 
-    def preview(self):
-        """ Blocking function. Opens OpenCV window to display stream. """
-        win_name = self.ip
-        cap = cv2.VideoCapture(self.url, cv2.CAP_FFMPEG)
-        ret, frame = cap.read()
+        # opens the stream capture, but does not retrieve any frames yet.
+        self._open_video_capture()
+
+    def _open_video_capture(self):
+        # To CAP_FFMPEG or not To ?
+        self.capture = cv2.VideoCapture(self.url, cv2.CAP_FFMPEG)
+
+    def _stream_blocking(self):
+        while True:
+            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
+
+    @threaded
+    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()
 
-        while ret:
-            cv2.imshow(win_name, frame)
+    def stop_stream(self):
+        self.capture.release()
+        self.thread_cancelled = True
 
-            ret, frame = cap.read()
-            if (cv2.waitKey(1) & 0xFF == ord('q')):
-                break
+    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.
+        """
 
-        cap.release()
-        cv2.destroyAllWindows()
+        # 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()
diff --git a/api/__init__.py b/api/__init__.py
@@ -1,4 +1,4 @@
 from .APIHandler import APIHandler
 
-__version__ = "0.0.4"
+__version__ = "0.0.5"
 VERSION = __version__
diff --git a/api/recording.py b/api/recording.py
@@ -9,6 +9,7 @@
 
 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.
@@ -53,34 +54,35 @@ def set_recording_encoding(self,
         body = [{"cmd": "SetEnc",
                  "action": 0,
                  "param":
-                 {"Enc":
-                  {"audio": audio,
-                   "channel": 0,
-                   "mainStream": {
-                       "bitRate": main_bit_rate,
-                       "frameRate": main_frame_rate,
-                       "profile": main_profile,
-                       "size": main_size},
-                   "subStream": {
-                       "bitRate": sub_bit_rate,
-                       "frameRate": sub_frame_rate,
-                       "profile": sub_profile,
-                       "size": sub_size}}
-                  }}]
+                     {"Enc":
+                          {"audio": audio,
+                           "channel": 0,
+                           "mainStream": {
+                               "bitRate": main_bit_rate,
+                               "frameRate": main_frame_rate,
+                               "profile": main_profile,
+                               "size": main_size},
+                           "subStream": {
+                               "bitRate": sub_bit_rate,
+                               "frameRate": sub_frame_rate,
+                               "profile": sub_profile,
+                               "size": sub_size}}
+                      }}]
         return self._execute_command('SetEnc', body)
 
     ###########
     # 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)
-        rtsp_client.preview()
+            ip=self.ip, username=self.username, password=self.password, proxies=proxies, callback=callback)
+        return rtsp_client.open_stream()
 
     def get_snap(self, timeout: int = 3, proxies=None) -> Image or None:
         """
diff --git a/examples/streaming_video.py b/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.isAlive())
+    while True:
+        if not t.isAlive():
+            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()
diff --git a/util.py b/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
