multi_thread_proc.py

import cv2
import threading
import queue
import time


selected_filters = ["invert"]

MAX_QUEUE_SIZE = 10


def grayscale(frame):
    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
    return cv2.cvtColor(gray, cv2.COLOR_GRAY2BGR)


def blur(frame):
    return cv2.GaussianBlur(frame, (15, 15), 0)


def edges(frame):
    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
    blurred = cv2.GaussianBlur(gray, (5, 5), 0)
    edges = cv2.Canny(blurred, 50, 150)
    return cv2.cvtColor(edges, cv2.COLOR_GRAY2BGR)


def invert(frame):
    return cv2.bitwise_not(frame)


FILTERS = {
    "grayscale": grayscale,
    "blur": blur,
    "edges": edges,
    "invert": invert,
}


def apply_filters(frame, filters):
    for filter_name in filters:
        if filter_name in FILTERS:
            frame = FILTERS[filter_name](frame)
        else:
            print(f"Filter '{filter_name}' not found. Skipping.")
    return frame


def producer(input_path, input_queue, input_semaphore, num_consumers):
    cap = cv2.VideoCapture(input_path)
    if not cap.isOpened():
        print("Error opening video file.")
        return

    while cap.isOpened():
        ret, frame = cap.read()
        if not ret:
            break
        input_semaphore.acquire()
        input_queue.put(frame)

    # Put sentinel values for all consumers
    for _ in range(num_consumers):
        input_semaphore.acquire()
        input_queue.put(None)

    cap.release()


def consumer(input_queue, input_semaphore, output_semaphore, output_queue, done_counter, lock, num_consumers):
    while True:
        frame = input_queue.get()
        input_semaphore.release()

        if frame is None:
            with lock:
                done_counter[0] += 1
                if done_counter[0] == num_consumers:
                    output_queue.put(None)
            break

        processed_frame = apply_filters(frame, selected_filters)

        output_semaphore.acquire()
        output_queue.put(processed_frame)


def writer(input_path, output_path, output_queue, output_semaphore):
    cap = cv2.VideoCapture(input_path)
    frame_width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    frame_height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
    fps = cap.get(cv2.CAP_PROP_FPS)
    cap.release()

    fourcc = cv2.VideoWriter_fourcc(*"mp4v")
    out = cv2.VideoWriter(output_path, fourcc, fps, (frame_width, frame_height))

    while True:
        output_semaphore.release()
        processed_frame = output_queue.get()
        if processed_frame is None:
            break

        out.write(processed_frame)

    out.release()


def process_video(input_video_path, output_video_path):
    input_queue = queue.Queue(maxsize=MAX_QUEUE_SIZE)
    output_queue = queue.Queue(maxsize=MAX_QUEUE_SIZE)
    input_semaphore = threading.Semaphore(MAX_QUEUE_SIZE)
    output_semaphore = threading.Semaphore(MAX_QUEUE_SIZE)

    num_consumers = 3
    done_counter = [0]  
    lock = threading.Lock()

    producer_thread = threading.Thread(
        target=producer, args=(input_video_path, input_queue, input_semaphore, num_consumers)
    )
    producer_thread.start()

    consumer_threads = []
    for _ in range(num_consumers):
        consumer_thread = threading.Thread(
            target=consumer,
            args=(input_queue, input_semaphore, output_semaphore, output_queue, done_counter, lock, num_consumers),
        )
        consumer_thread.start()
        consumer_threads.append(consumer_thread)

    writer_thread = threading.Thread(
        target=writer,
        args=(input_video_path, output_video_path, output_queue, output_semaphore),
    )
    writer_thread.start()

    producer_thread.join()
    for consumer_thread in consumer_threads:
        consumer_thread.join()
    writer_thread.join()

    print("Video processing complete.")

def worker(i, pipe):
    # """task_queue, result_queue"""
    # while True:
    # video_path = task_queue.get()
    video_path = pipe.recv()
    # if video_path is None:
    #     break

    output_path = f"multi_thread_output/output_{i}.mp4"
    process_video(video_path, output_path)

    pipe.send(f"Processed: {video_path}")
    pipe.close()


import os


def find_mp4_files(directory):
    mp4_files = []
    for filename in os.listdir(directory):
        if filename.endswith(".mp4"):
            mp4_files.append(directory + "/" + filename)
    return mp4_files


if __name__ == "__main__":
    from multiprocessing import Process, Pipe

    process_list = []
    process_pipe_list = []

    now = time.time()
    mp4_files = find_mp4_files("./videos")
    # mp4_files=mp4_files[0:1]
    j = 0
    for i in range(len(mp4_files)):

        print(mp4_files[i])
        # parent_conn,child_conn
        process_pipe_list.append(Pipe())
        p = Process(target=worker, args=(i, process_pipe_list[j][1],))
        process_list.append(p)
        p.start()
        process_pipe_list[j][0].send(mp4_files[i])
        j += 1
    for pipe in process_pipe_list:
        print(pipe[0].recv())
    for p in process_list:
        p.join()

    print(f"Processing Duration: {time.time() - now}(s)")