Optimizing Real-time Video Frame Capture Python/cv2

Question

I have a thread dedicated to reading frames from a camera, aiming to read at a specific frequency (self.target_fps). Therefore, I need to perform a read every self.target_frequency = 1 / self.target_fps.

The video thread runs a loop where it reads at each iteration, regardless of the target frequency (target_fps), to empty the VideoCapture buffer (particularly useful when self.target_fps is low, for example, 1 fps). After the read, I want to add the frame to a queue only if it's the right time, i.e., if a time interval equal to self.target_frequency has passed. The "start" timer helps me determine when it's time to enqueue the frame and when it's not. Are there better approaches for efficiently reading frames from an RTSP camera and achieving a reasonably accurate frame rate?

            self.target_fps = 7
            self.target_frequency = 1 / self.target_fps

            start = time.time()

            while not self.stop_video_thread.is_set():
                success, frame = cap.read()

                interval = time.time() - start

                if interval >= self.target_frequency:
                    start = time.time()

                    if success:
                        frame_queue.put_nowait(frame)

Answer 1

• You can use the set method of VideoCapture to adjust the frame rate of the camera. For example, cap.set(cv2.CAP_PROP_FPS, self.target_fps) will set the frame rate to your desired value. This way, you don't need to check the interval every time you read a frame.
• You can use a threading.Timer object to schedule the reading of frames at regular intervals. For example, timer = threading.Timer(self.target_frequency, read_frame) will create a timer that calls the read_frame function every self.target_frequency seconds. better than a while loop
• You can use a multiprocessing.Queue instead of a queue.Queue to share the frames between processes. So you can avoid the overhead of pickling and unpickling the frames.

import cv2
import threading
import multiprocessing as mp

def read_frame(cap, frame_queue):
    success, frame = cap.read()
    if success:
        frame_queue.put(frame)
    timer = threading.Timer(self.target_frequency, read_frame, args=(cap, frame_queue))
    timer.start()

def main():
    # Create a video capture object and set the frame rate
    cap = cv2.VideoCapture('rtsp_address')
    cap.set(cv2.CAP_PROP_FPS, self.target_fps)

    frame_queue = mp.Queue()

    timer = threading.Timer(self.target_frequency, read_frame, args=(cap, frame_queue))
    timer.start()

    while True:
        frame = frame_queue.get()
        # Process the frame here
        #########################