optimize latency

app.py

@@ -6,6 +6,8 @@ import base64
 import numpy as np
 import json
 import logging
+import threading
+import queue
 
 app = FastAPI()
 
@@ -43,19 +45,30 @@ async def detect(websocket: WebSocket):
     logging.info("WebSocket connection accepted")
 
     # Move streams inside the detect function
-    streams = {}
+    streams = {}  # Modify to store {camera_id: {'cap': VideoCapture, 'buffer': Queue, 'thread': Thread}}
+
+    def frame_reader(camera_id, cap, buffer):
+        while True:
+            ret, frame = cap.read()
+            if not ret:
+                logging.warning(f"Failed to read frame from camera: {camera_id}")
+                break
+            if not buffer.empty():
+                try:
+                    buffer.get_nowait()  # Discard the old frame
+                except queue.Empty:
+                    pass
+            buffer.put(frame)
 
     async def process_streams():
         logging.info("Started processing streams")
         while True:
             start_time = time.time()
             # Round-robin processing
-            for camera_id, cap in list(streams.items()):
+            for camera_id, stream in list(streams.items()):
-                logging.debug(f"Processing camera: {camera_id}")
+                buffer = stream['buffer']
-                ret, frame = cap.read()
+                if not buffer.empty():
-                if not ret:
+                    frame = buffer.get()
-                    logging.warning(f"Failed to read frame from camera: {camera_id}")
-                    continue
                     results = model(frame, stream=False)
                     boxes = []
                     for r in results:
@@ -93,11 +106,16 @@ async def detect(websocket: WebSocket):
 
             if camera_id and rtsp_url:
                 if camera_id not in streams and len(streams) < max_streams:
-                    streams[camera_id] = cv2.VideoCapture(rtsp_url)
+                    cap = cv2.VideoCapture(rtsp_url)
+                    buffer = queue.Queue(maxsize=1)
+                    thread = threading.Thread(target=frame_reader, args=(camera_id, cap, buffer))
+                    thread.daemon = True
+                    thread.start()
+                    streams[camera_id] = {'cap': cap, 'buffer': buffer, 'thread': thread}
                     logging.info(f"Subscribed to camera {camera_id} with URL {rtsp_url}")
                 elif camera_id and camera_id in streams:
-                cap = streams.pop(camera_id)
+                    stream = streams.pop(camera_id)
-                cap.release()
+                    stream['cap'].release()
                     logging.info(f"Unsubscribed from camera {camera_id}")
             elif data.get("command") == "stop":
                 logging.info("Received stop command")
@@ -106,8 +124,8 @@ async def detect(websocket: WebSocket):
         logging.error(f"Error in WebSocket connection: {e}")
     finally:
         task.cancel()
-        for camera_id, cap in streams.items():
+        for camera_id, stream in streams.items():
-            cap.release()
+            stream['cap'].release()
             logging.info(f"Released camera {camera_id}")
         streams.clear()
         logging.info("WebSocket connection closed")