I am facing issues using ffmpeg python to process camera frames. My first approach with process.communicate() worked well, but has latency issues.

process = (
            ffmpeg
            .input('pipe:', format='rawvideo', pix_fmt='rgb24', s='{}x{}'.format(width, height))
            .filter(<filter_params>)
            .output('pipe:', format='rawvideo', pix_fmt='rgb24')
            .run_async(pipe_stdin=True, pipe_stdout=True, pipe_stderr=True)
        )
out, err = process.communicate(input=img.tobytes())
output_image = np.frombuffer(out, np.uint8).reshape((height, width, channels))

To reduce the latency, I’m trying to keep the ffmpeg process open and feed in camera frames for processing. This runs fine for a couple of minutes with acceptable latency values but ends up in deadlock. What is the best way to fix this?

import cv2
import numpy as np
import math

import ffmpeg 

def start_ffmpeg_process_async(width, height):
    return (
        ffmpeg
            .input('pipe:', format='rawvideo', pix_fmt='rgb24', s='{}x{}'.format(width, height))
            .filter('<filter variables>')
            .output('pipe:', format='rawvideo', pix_fmt='rgb24')
            .run_async(pipe_stdin=True, pipe_stdout=True, pipe_stderr=True)
    )

def main():

    cap = cv2.VideoCapture(0)

    if not cap.isOpened():
        print("Camera can't open nExiting...")
        return -1

    ffmpeg_process_async = start_ffmpeg_process_async(cap.get(cv2.CAP_PROP_FRAME_WIDTH), 
        cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
 
    try:
        while True:

            success, img = cap.read() 
            if success:
                height, width, channels = img.shape
                ffmpeg_process_async.stdin.write(img.tobytes())
                raw_output = ffmpeg_process_async.stdout.read(width * height * channels)
                output_image = np.frombuffer(raw_output, np.uint8).reshape((height, width, channels))
                cv2.imshow('Webcam', output_image)
            else:
                print ("Camera read error")
            if cv2.waitKey(1) == ord('q'):
                print ("Exiting . . .")
                break
    finally:
        print ("Finalizing . . .")
        ffmpeg_process_async.stdin.close()
        ffmpeg_process_async.wait()
        cap.release()
        cv2.destroyAllWindows()


if __name__ == "__main__":
    main()