Fix MJPEG streams not connecting.

Author: RamesTheGeneric

BabbleApp/babbleapp.spec

@@ -23,7 +23,7 @@ exe = EXE(pyz,
           [],
           exclude_binaries=True,
           name='Babble_App',
-          debug=False,
+          debug=True,
           bootloader_ignore_signals=False,
           strip=False,
           upx=True,

BabbleApp/camera.py

@@ -10,10 +10,9 @@
 from enum import Enum
 import serial.tools.list_ports
 from lang_manager import LocaleStringManager as lang
-
 from colorama import Fore
 from config import BabbleConfig, BabbleSettingsConfig
-from utils.misc_utils import get_camera_index_by_name, list_camera_names
+from utils.misc_utils import get_camera_index_by_name, list_camera_names, is_nt
 
 from vivefacialtracker.vivetracker import ViveTracker
 from vivefacialtracker.camera_controller import FTCameraController
@@ -95,6 +94,7 @@ def run(self):
                 self.config.capture_source is not None
                 and self.config.capture_source != ""
             ):
+                self.current_capture_source = self.config.capture_source
                 isSerial = any(x in str(self.config.capture_source) for x in PORTS)
 
                 if isSerial:
@@ -103,7 +103,7 @@ def run(self):
                         self.cv2_camera = None
                     if self.vft_camera is not None:
                         self.vft_camera.close()
-                    self.device_is_vft = False;
+                    self.device_is_vft = False
                     if (
                         self.serial_connection is None
                         or self.camera_status == CameraState.DISCONNECTED
@@ -116,7 +116,7 @@ def run(self):
                     if self.cv2_camera is not None:
                         self.cv2_camera.release()
                         self.cv2_camera = None
-                    self.device_is_vft = True;
+                    self.device_is_vft = True
 
                     if self.vft_camera is None:
                         print(self.error_message.format(self.config.capture_source))
@@ -141,18 +141,18 @@ def run(self):
                         self.cv2_camera is None
                         or not self.cv2_camera.isOpened()
                         or self.camera_status == CameraState.DISCONNECTED
-                        or get_camera_index_by_name(self.config.capture_source) != self.current_capture_source 
+                        #or get_camera_index_by_name(self.config.capture_source) != self.current_capture_source 
+                        or self.config.capture_source != self.current_capture_source 
                     ):
                         if self.vft_camera is not None:
                             self.vft_camera.close()
-                        self.device_is_vft = False;
+                        self.device_is_vft = False
 
                         print(self.error_message.format(self.config.capture_source))
                         # This requires a wait, otherwise we can error and possible screw up the camera
                         # firmware. Fickle things.
                         if self.cancellation_event.wait(WAIT_TIME):
                             return
-                        
                         if self.config.capture_source not in self.camera_list:
                             self.current_capture_source = self.config.capture_source
                         else:
@@ -163,9 +163,8 @@ def run(self):
                                 self.current_capture_source, cv2.CAP_FFMPEG
                             )
                         else:
-                            self.cv2_camera = cv2.VideoCapture(
-                                self.current_capture_source
-                            )
+                            self.cv2_camera = cv2.VideoCapture()
+                            self.cv2_camera.open(self.current_capture_source)
 
                         if not self.settings.gui_cam_resolution_x == 0:
                             self.cv2_camera.set(
@@ -181,7 +180,6 @@ def run(self):
                             self.cv2_camera.set(
                                 cv2.CAP_PROP_FPS, self.settings.gui_cam_framerate
                             )
-                            
                         should_push = False
             else:
                 # We don't have a capture source to try yet, wait for one to show up in the GUI.
@@ -215,15 +213,14 @@ def get_camera_picture(self, should_push):
                     return
                 self.frame_number = self.frame_number + 1
             elif self.cv2_camera is not None and self.cv2_camera.isOpened():
-                ret, image = self.cv2_camera.read()
+                ret, image = self.cv2_camera.read()     # MJPEG Stream reconnects are currently limited by the hard coded 30 second timeout time on VideoCapture.read(). We can get around this by recompiling OpenCV or using a custom MJPEG stream imp.   
                 if not ret:
                     self.cv2_camera.set(cv2.CAP_PROP_POS_FRAMES, 0)
                     raise RuntimeError(lang._instance.get_string("error.frame"))
                 self.frame_number = self.cv2_camera.get(cv2.CAP_PROP_POS_FRAMES) + 1
             else:
                 # Switching from a Vive Facial Tracker to a CV2 camera
                 return
-                
             self.FRAME_SIZE = image.shape
             # Calculate FPS
             current_frame_time = time.time()    # Should be using "time.perf_counter()", not worth ~3x cycles?

BabbleApp/mjpeg_client.py

@@ -0,0 +1,33 @@
+from mjpeg.client import MJPEGClient
+from time import sleep
+import numpy as np
+import cv2
+
+
+url='http://192.168.1.186:8080/video'
+
+# Create a new client thread
+client = MJPEGClient(url)
+
+# Allocate memory buffers for frames
+bufs = client.request_buffers(65536, 50)
+for b in bufs:
+    client.enqueue_buffer(b)
+    
+# Start the client in a background thread
+client.start()
+while True:
+    client.print_stats()
+    buf = client.dequeue_buffer()
+    client.enqueue_buffer(buf)
+    data = memoryview(buf.data)[:buf.used]
+    img_array = np.frombuffer(data, dtype=np.uint8)
+    frame = cv2.imdecode(img_array, cv2.IMREAD_COLOR)
+        # Check if the frame was decoded correctly
+    if frame is not None:
+        cv2.imshow("MJPEG Stream", frame)
+    # Exit loop if 'q' is pressed
+    if cv2.waitKey(1) & 0xFF == ord("q"):
+        break
+    #print(frame.shape)
+    #print(len(array))

BabbleApp/mjpeg_test.py

@@ -0,0 +1,77 @@
+import requests
+import numpy as np
+import cv2
+import threading
+import io
+
+class MJPEGVideoCapture:
+    def __init__(self, url):
+        self.url = url
+        self.session = requests.Session()
+        self.stream = None
+        self.byte_buffer = b""
+        self.frame = None
+        self.running = False
+        self.lock = threading.Lock()
+        self.thread = None
+    
+    def open(self):
+        if not self.running:
+            self.running = True
+            self.thread = threading.Thread(target=self._update, daemon=True)
+            self.thread.start()
+    
+    def _update(self):
+        while self.running:
+            try:
+                self.stream = self.session.get(self.url, stream=True, timeout=3)
+                for chunk in self.stream.iter_content(chunk_size=1024):
+                    if not self.running:
+                        break
+                    self.byte_buffer += chunk
+                    start = self.byte_buffer.find(b'\xff\xd8')  # JPEG start
+                    end = self.byte_buffer.find(b'\xff\xd9')  # JPEG end
+                    
+                    if start != -1 and end != -1:
+                        jpg = self.byte_buffer[start:end+2]
+                        self.byte_buffer = self.byte_buffer[end+2:]
+                        
+                        image = np.frombuffer(jpg, dtype=np.uint8)
+                        if image.size != 0:
+                            frame = cv2.imdecode(image, cv2.IMREAD_GRAYSCALE)
+                            if frame is not None:
+                                with self.lock:
+                                    self.frame = frame
+            except requests.RequestException:
+                continue  # Retry on failure
+    
+    def read(self):
+        with self.lock:
+            return self.frame is not None, self.frame.copy() if self.frame is not None else None
+    
+    def isOpened(self):
+        return self.running
+    
+    def release(self):
+        self.running = False
+        if self.thread is not None:
+            self.thread.join()
+        self.stream = None
+        self.frame = None
+        self.byte_buffer = b""
+        self.session.close()
+
+if __name__ == "__main__":
+    cap = MJPEGVideoCapture("http://openiristracker.local")
+    cap.open()
+    
+    while cap.isOpened():
+        ret, frame = cap.read()
+        if ret:
+            cv2.imshow("MJPEG Stream", frame)
+        
+        if cv2.waitKey(1) & 0xFF == ord("q"):
+            break
+    
+    cap.release()
+    cv2.destroyAllWindows()

BabbleApp/mjpeg_videocapture.py

@@ -0,0 +1,115 @@
+import requests
+import numpy as np
+import cv2
+import threading
+import time
+import queue
+
+class MJPEGVideoCapture:
+    def __init__(self, url, max_buffer_size=1024*1024):
+        self.url = url
+        self.session = requests.Session()
+        self.stream = None
+        self.byte_buffer = b""
+        self.max_buffer_size = max_buffer_size  # Maximum allowed size for byte_buffer (e.g., 1 MB)
+        self.running = False
+        self.thread = None
+        # Use a queue with maxsize=1 so that only the latest frame is kept.
+        self.frame_queue = queue.Queue(maxsize=1)
+        print("VC Opened")
+    
+    def open(self):
+        if not self.running:
+            self.running = True
+            self.thread = threading.Thread(target=self._update, daemon=True)
+            self.thread.start()
+            # Allow some time for the update thread to start and capture a frame.
+            time.sleep(2)
+    
+    def _update(self):
+        while self.running:
+            try:
+                self.stream = self.session.get(self.url, stream=True, timeout=1)
+                for chunk in self.stream.iter_content(chunk_size=512):
+                    if not self.running:
+                        break
+                    self.byte_buffer += chunk
+
+                    # Flush the buffer if it grows too large.
+                    if len(self.byte_buffer) > self.max_buffer_size:
+                        print("Warning: byte_buffer exceeded maximum size; flushing buffer.")
+                        self.byte_buffer = b""
+                        continue
+
+                    # Instead of taking the first complete JPEG frame, we look for the latest complete one.
+                    # Find the last occurrence of the JPEG start and end markers.
+                    start = self.byte_buffer.rfind(b'\xff\xd8')
+                    end = self.byte_buffer.rfind(b'\xff\xd9')
+                    
+                    # If a complete JPEG is found, extract it and flush the entire buffer.
+                    if start != -1 and end != -1 and end > start:
+                        jpg = self.byte_buffer[start:end+2]
+                        self.byte_buffer = b""  # Discard all other data.
+                        
+                        image = np.frombuffer(jpg, dtype=np.uint8)
+                        if image.size != 0:
+                            # Decode as grayscale.
+                            frame = cv2.imdecode(image, cv2.IMREAD_GRAYSCALE)
+                            if frame is not None:
+                                # Resize to 240x240.
+                                frame = cv2.resize(frame, (240, 240))
+                                # Convert grayscale to 3-channel image with shape (240, 240, 3).
+                                frame = np.stack([frame] * 3, axis=-1)
+                                
+                                # Try to put the frame into the queue.
+                                # If the queue is full, remove the old frame first.
+                                try:
+                                    self.frame_queue.put(frame, block=False)
+                                except queue.Full:
+                                    try:
+                                        self.frame_queue.get_nowait()
+                                    except queue.Empty:
+                                        pass
+                                    self.frame_queue.put(frame, block=False)
+            except requests.RequestException:
+                # On failure, simply continue to try again.
+                continue
+    
+    def read(self):
+        """
+        Block until a frame is available and return the latest frame.
+        This mimics the blocking behavior of cap.read(), but always returns the latest frame.
+        """
+        frame = self.frame_queue.get()  # This call blocks until a frame is available.
+        return True, frame.copy()
+    
+    def isOpened(self):
+        return self.running
+    
+    def release(self):
+        self.running = False
+        if self.thread is not None:
+            self.thread.join()
+        self.stream = None
+        self.byte_buffer = b""
+        self.session.close()
+
+# Testing code:
+test = True
+if test:
+    if __name__ == "__main__":
+        cap = MJPEGVideoCapture("http://192.168.1.186:8080/video")
+        cap.open()
+        
+        while cap.isOpened():
+            ret, frame = cap.read()
+            # Print the current byte_buffer length to monitor its size.
+            print("Current byte_buffer length:", len(cap.byte_buffer))
+            if ret:
+                cv2.imshow("MJPEG Stream", frame)
+            
+            if cv2.waitKey(1) & 0xFF == ord("q"):
+                break
+        
+        cap.release()
+        cv2.destroyAllWindows()