Project 8

• Perform Optical flow on the video file provided and highlight the features of interest using the Shi Tomasi algorithm

The features were detected in the given video using the Shi Tomasi Algorithm. The function used to generate the Shi Tomasi features is given below:

def find_features(gray_image: ndarray):
    """
    :param gray_image: An ndarray representing the image in grayscale to detect the features on.
    :returns: The feature points detected.
    """
    points = cv2.goodFeaturesToTrack(gray_image, mask=None, maxCorners = 100, qualityLevel = 0.3, 
                                     minDistance = 10, blockSize = 7)
    return points

The cv2.goodFeaturesToTrack() function was used to detect the corners of the vehicles(features of interest) in the given video.

The Function used to find the Lucas Kanade optical flow is given below:

def LK_model(gray_t_1: ndarray, gray_t: ndarray, prev_points) -> Tuple:
    """
    :param gray_t_1: An ndarray representing the previous frame to perform Lucas Kannade optical flow.
    :param gray_t: An ndarray representing the next frame to perform Lucas Kannade optical flow.
    :param prev_points: The feature points in the previous frame used to track the optical flow.
    :returns: A tuple of the feature points in the current frame detected using Lucas Kannade Algorithm, 
              the status of the process, and whether it produced any errors.
    """
    points, status, error = cv2.calcOpticalFlowPyrLK(gray_t_1, gray_t, prev_points, None, winSize  = (10, 10), maxLevel = 2,
                                                     criteria = (cv2.TERM_CRITERIA_EPS | cv2.TERM_CRITERIA_COUNT, 10, 0.03))
    return (points, status, error)

The function cv2.calcOpticalFlowPyrLK() method was used to find the optical flow. The function returns the features detected on the new frame, status and errors detected.

The main method or the script runner is given below:

def main():
    exec_path = sys.path[0]
    filepath = os.path.join(exec_path, f'traffic.mp4')
    
    # Assign a random color to show the optical flow.
    color = np.random.randint(0, 255, (100, 3))
    capture = cv2.VideoCapture(filepath)
    
    # Get the width and height of the video feed from the capture object above.
    # Assign the encoding format as *"mp4v" to save the video as mp4 video.
    # Initialise two cv2.VideoWriter() objects to store the mask output and the processed video output respectively.
    frame_width = int(capture.get(3))
    frame_height = int(capture.get(4))
    size = (frame_width, frame_height)
    fourcc = cv2.VideoWriter_fourcc(*"mp4v")
    mask_out = cv2.VideoWriter('Optical flow mask.mp4', fourcc, 23.976, size, isColor = True)
    processed_video_out = cv2.VideoWriter('Optical Flow superimposed video.mp4', fourcc, 23.976, size, isColor = True)
    features_video_out = cv2.VideoWriter('Features detected.mp4', fourcc, 23.976, size, isColor = True)
    
    # Read  the first frame
    # Inititialise a blank ndarray for the empty mask onto which the optical flow is written.
    # Convert the first frame to grayscale 
    # Find the feature to use for optical flow using the find_features() function.
    ret, old_frame = capture.read()
    mask = np.zeros_like(old_frame)
    prev_gray = cv2.cvtColor(old_frame, cv2.COLOR_BGR2GRAY)
    prev_points = find_features(prev_gray)

    while(1):
        # Read the current frame
        ret, frame = capture.read()
        
        # Break the while loop at the end of the video feed.
        if not ret:
            break
        
        # Convert the current frame to grapscale
        # Call the Lucas Kannade model to find the optical flow of the features detected above.
        current_gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        current_points, status, error = LK_model(prev_gray, current_gray, prev_points)
        
        # If the features are detected.
        if current_points is not None:
            current_points = current_points[status == 1]
            old_points = prev_points[status == 1]
        
        # Enumerate over all the feature points detected 
        # draw a circle over the feature points detected
        # draw a line on the mask from the coordinates of the feature in the previous position to the current position
        for i, (current, prev) in enumerate(zip(current_points, old_points)):
            a, b = current.ravel()
            c, d = prev.ravel()
            mask = cv2.line(mask, (int(a), int(b)), (int(c), int(d)), color[i].tolist(), 2)
            frame = cv2.circle(frame, (int(a), int(b)), 5, color[i].tolist(), -1)
            
        # Write the image to the respective cv2.VideoWriter() objects
        mask_out.write(mask)
        features_video_out.write(frame)
        img = cv2.add(frame, mask)
        processed_video_out.write(img)
        
        # if 'q' is pressed terminate the while loop.
        k = cv2.waitKey(5) & 0xff
        if k == 27:
            break
            
        # Set the current frame to previous frame and 
        # set te current frame features to the previous frame features variables
        # to continue iteration.
        prev_gray = current_gray.copy()
        prev_points = current_points.reshape(-1, 1, 2)
    
    # Release all the cv2.VideoWriter() objects to ensure there is no error during saving.
    capture.release()
    mask_out.release()
    processed_video_out.release()
    features_video_out.release()
    
    cv2.destroyAllWindows()
    # If process execute successfully print 0.
    print("0")

The main() method was triggered using the code below:

if __name__ == '__main__':
    main()

The Shi Tomasi features highlighted on the video output are:

Alternate link

Features of interest Highlighted using Shi Tomasi Algorithm - Features of interest highlighted using Shi Tomasi Algorithm

The masked video sequence - The Masked video sequence of the Optical flow

Video with optical flow lines superimposed on the video - Optical flow lines superimposed on the give video