OpenCV video writing decreases FPS drastically. How to optimize performance?

Question

I'm working on a project that involves object detection + sort tracking. I have scripts to work with videos an camera using OpenCV in the Coral dev board.

The main issue is when make usage of the VideoWriter to save the output of the detections.

For camera script it's usage decreases fps rate from 11 to 2.3 and for video script from 6-7 to 2.

Is there a way to solve/optimize this issue.

Here is my portion of code that grabs frames, detects and tracks, and then writes.

# Read frames
while(video.isOpened()):

    # Acquire frame and resize to expected shape [1xHxWx3]
    ret, frame = video.read()


    if not ret:
        break

    # Debug info
    frame_count += 1
    print("[INFO] Processing frame: {}".format(frame_count))

    if FLIP:
        frame = cv2.flip(frame, 1)

    if ROTATE != 0:
        frame = cv2.rotate(frame, ROTATE) # Rotate image on given angle


    frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) # Convert to RGB
    frame = cv2.resize(frame, (VIDEO_WIDTH, VIDEO_HEIGHT)) # resize frame to output dims
    frame_resized = cv2.resize(frame_rgb, (width, height)) # resize to fit tf model dims
    input_data = np.expand_dims(frame_resized, axis=0)

    # Normalize pixel values if using a floating model (i.e. if model is non-quantized)
    if floating_model:
        input_data = (np.float32(input_data) - input_mean) / input_std

    # Initialize writer
    if (writer is None) and (SAVE_VIDEO) :  
        writer = cv2.VideoWriter(VIDEO_OUTPUT, cv2.VideoWriter_fourcc(*'XVID'), args.fps, (VIDEO_WIDTH, VIDEO_HEIGHT))

    # Perform the actual detection by running the model with the image as input
    #s_detection_time = time.time()
    interpreter.set_tensor(input_details[0]['index'],input_data)
    interpreter.invoke()
    #e_detection_time = time.time()

    #print("[INFO] Detection time took: {} seconds".format(e_detection_time-s_detection_time))

    # Retrieve detection results
    boxes = interpreter.get_tensor(output_details[0]['index'])[0] # Bounding box coordinates of detected objects
    classes = interpreter.get_tensor(output_details[1]['index'])[0] # Class index of detected objects
    scores = interpreter.get_tensor(output_details[2]['index'])[0] # Confidence of detected objects
    #num = interpreter.get_tensor(output_details[3]['index'])[0]  # Total number of detected objects (inaccurate and not needed)

    #print("[INFO] Boxes: {}".format(boxes))

    detections = np.array([[]])

    #s_detections_loop = time.time()
    # Loop over all detections and draw detection box if confidence is above minimum threshold
    for i in range(len(scores)):
        if ((scores[i] > min_conf_threshold) and (scores[i] <= 1.0)):

            #print("[INFO] Box ", i , ": ", boxes[i])

            # Get bounding box coordinates and draw box
            # Interpreter can return coordinates that are outside of image dimensions, need to force them to be within image using max() and min()
            ymin = int(max(1,(boxes[i][0] * VIDEO_HEIGHT)))  
            xmin = int(max(1,(boxes[i][1] * VIDEO_WIDTH)))
            ymax = int(min(VIDEO_HEIGHT,(boxes[i][2] * VIDEO_HEIGHT)))
            xmax = int(min(VIDEO_WIDTH,(boxes[i][3] * VIDEO_WIDTH)))

            # Calculate centroid of bounding box
            #centroid_x = int((xmin + xmax) / 2)
            #centroid_y = int((ymin + ymax) / 2)



            # Format detection for sort and append to current detections
            detection = np.array([[xmin, ymin, xmax, ymax]])

            #f.write("Box {}: {}\n".format(i, detection[:4]))
            #print("[INFO] Size of detections: ", detections.size)

            if detections.size == 0: 
                detections = detection
            else:
                detections = np.append(detections, detection, axis=0)

            # Draw a circle indicating centroid
            #print("[INFO] Centroid of box ", i, ": ", (centroid_x, centroid_y))
            #cv2.circle(frame, (centroid_x, centroid_y), 6, (0, 0, 204), -1)

            # Calculate area of rectangle
            #obj_height = (ymin + ymax)
            #print("[INFO] Object height: ", obj_height)

            # Check if centroid passes ROI
            # Draw the bounding box
            #cv2.rectangle(frame, (xmin,ymin), (xmax,ymax), (0, 0, 255), 4)
            #print("[INFO] Object passing ROI")
            #print("[INFO] Object height: ", obj_height)
            #counter += 1
            #print("[INFO] Object out of ROI")
            # Draw the bounding box
            #cv2.rectangle(frame, (xmin,ymin), (xmax,ymax), (10, 255, 0), 4)

            #print("[INFO] Total objects counted: ", counter)


            # Draw label
            """object_name = labels[int(classes[i])] # Look up object name from "labels" array using class index
            label = '%s: %d%%' % (object_name, int(scores[i]*100)) # Example: 'person: 72%'
            labelSize, baseLine = cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, 0.7, 2) # Get font size
            label_ymin = max(ymin, labelSize[1] + 10) # Make sure not to draw label too close to top of window
            cv2.rectangle(frame, (xmin, label_ymin-labelSize[1]-10), (xmin+labelSize[0], label_ymin+baseLine-10), (255, 255, 255), cv2.FILLED) # Draw white box to put label text in
            cv2.putText(frame, label, (xmin, label_ymin-7), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 0, 0), 2) # Draw label text
            """
    #f.write("\n")
    #e_detection_loop = time.time()

    #print("[INFO] Detection loop time took {} seconds".format(e_detection_loop-s_detections_loop))

    #s_tracker_update = time.time()
    # Update sort tracker
    print("[INFO] Current Detections: ", detections.astype(int))
    objects_tracked = tracker.update(detections.astype(int))
    #e_tracker_update = time.time()

    #print("[INFO] Updating trackers state took {} seconds".format(e_tracker_update-s_tracker_update))

    #s_draw_tracked = time.time()
    # Process every tracked object
    for object_tracked in objects_tracked:
        if object_tracked.active:
            bbox_color = (0, 128, 255)
        else:
            bbox_color = (10, 255, 0)

        bbox = object_tracked.get_state().astype(int)

        # Draw the bbox rectangle
        cv2.rectangle(frame, (int(bbox[0]), int(bbox[1])), (int(bbox[2]), int(bbox[3])), bbox_color, 4)

        # Calculate centroid of bounding box
        centroid = (object_tracked.last_centroid[0], object_tracked.last_centroid[1])    

        # Draw the centroid
        cv2.circle(frame, centroid, 6, (0, 0, 204), -1)

        label = '{} [{}]'.format(OBJECT_NAME,object_tracked.id) # Example: 'object [1]'
        labelSize, baseLine = cv2.getTextSize(label, FONT, 0.7, 2) # Get font size
        label_ymin = max(bbox[1], labelSize[1] + 10) # Make sure not to draw label too close to top of window
        cv2.rectangle(frame, (bbox[0], label_ymin-labelSize[1]-10), (bbox[0]+labelSize[0], label_ymin+baseLine-10), (255, 255, 255), cv2.FILLED) # Draw white box to put label text in
        cv2.putText(frame, label, (bbox[0], label_ymin-7), FONT, 0.7, (0, 0, 0), 2) # Draw label text
    #e_draw_tracked = time.time()

    #print("[INFO] Drawing tracked objects took {} seconds".format(e_draw_tracked-s_draw_tracked))


    # Update fps count
    fps.update()
    fps.stop()

    # Prepare fps display
    fps_label = "FPS: {0:.2f}".format(fps.fps())
    cv2.rectangle(frame, (0, 0), (int(VIDEO_WIDTH*0.6), int(VIDEO_HEIGHT*0.07)), (255, 255, 255), cv2.FILLED)
    cv2.putText(frame, fps_label, (int(VIDEO_WIDTH*0.01), int(VIDEO_HEIGHT*0.05)), FONT, 1.5, (10, 255, 0), 3)

    # Prepare total and active objects count display
    total_objects_text = "TOTAL {}S: {}".format(OBJECT_NAME,tracker.total_trackers)
    active_objects_text = "ACTIVE {}S: {}".format(OBJECT_NAME,tracker.active_trackers)
    cv2.putText(frame, total_objects_text, (int(VIDEO_WIDTH*0.1+VIDEO_WIDTH*0.06), int(VIDEO_HEIGHT*0.05)), FONT, 1.5, (0, 0, 255), 3) # Draw label text
    cv2.putText(frame, active_objects_text, (int(VIDEO_WIDTH*0.1+VIDEO_WIDTH*0.27), int(VIDEO_HEIGHT*0.05)), FONT, 1.5, (0, 128, 255), 3) # Draw label text

    # Draw horizontal boundaries
    cv2.line(frame, (LEFT_BOUNDARY, int(VIDEO_HEIGHT*0.07)), (LEFT_BOUNDARY, VIDEO_HEIGHT), (0, 255, 255), 4)
    #cv2.line(frame, (RIGHT_BOUNDARY, 0), (RIGHT_BOUNDARY, VIDEO_HEIGHT), (0, 255, 255), 4)

    #s_trackers_state = time.time()
    tracker.update_trackers_state()
    #e_trackers_state = time.time()

    #print("[INFO] Updating trackers state took {} seconds".format(e_trackers_state-s_trackers_state))

    # All the results have been drawn on the frame, so it's time to display it.
    cv2.imshow('Object detector', frame)

    # Center window
    if not IS_CENTERED:
        cv2.moveWindow('Object detector', 0, 0)
        IS_CENTERED = True

    if SAVE_VIDEO:
        writer.write(frame)

    print("\n\n")

    # Press 'q' to quit
    if cv2.waitKey(1) == ord('q'):
        break

Thanks in advance for any help!

Answer 1

An important thing when trying to optimize/improve code performance is to categorize and measure your code execution. Only after identifying what is actually causing the bottleneck or performance decrease then can you can work on improving those sections of code. For this approach I'm assuming that you're both reading and saving frames in the same thread. So if you're facing a performance reduction due to I/O latency then this method can help otherwise if you find out that your problem is due to CPU processing limitations then this method will not give you a performance boost.

That being said, the approach is to use threading. The idea is to create another separate thread for obtaining the frames as cv2.VideoCapture.read() is blocking. This can be expensive and cause latency as the main thread has to wait until it has obtained a frame. By putting this operation into a separate thread that just focuses on grabbing frames and processing/saving the frames in the main thread, it dramatically improves performance due to I/O latency reduction. Here's a simple example on how to use threading to read frames in one thread and show/save frames in the main thread. Be sure to change the capture_src to your stream.

Code

from threading import Thread
import cv2

class VideoWritingThreading(object):
    def __init__(self, src=0):
        # Create a VideoCapture object
        self.capture = cv2.VideoCapture(src)

        # Default resolutions of the frame are obtained (system dependent)
        self.frame_width = int(self.capture.get(3))
        self.frame_height = int(self.capture.get(4))

        # Set up codec and output video settings
        self.codec = cv2.VideoWriter_fourcc('M','J','P','G')
        self.output_video = cv2.VideoWriter('output.avi', self.codec, 30, (self.frame_width, self.frame_height))

        # Start the thread to read frames from the video stream
        self.thread = Thread(target=self.update, args=())
        self.thread.daemon = True
        self.thread.start()

    def update(self):
        # Read the next frame from the stream in a different thread
        while True:
            if self.capture.isOpened():
                (self.status, self.frame) = self.capture.read()

    def show_frame(self):
        # Display frames in main program
        if self.status:
            cv2.imshow('frame', self.frame)

        # Press Q on keyboard to stop recording
        key = cv2.waitKey(1)
        if key == ord('q'):
            self.capture.release()
            self.output_video.release()
            cv2.destroyAllWindows()
            exit(1)

    def save_frame(self):
        # Save obtained frame into video output file
        self.output_video.write(self.frame)

if __name__ == '__main__':
    capture_src = 'your stream link!'
    video_writing = VideoWritingThreading(capture_src)
    while True:
        try:
            video_writing.show_frame()
            video_writing.save_frame()
        except AttributeError:
            pass