method3.py

import cv2
import threading
import queue
import time
from google.colab.patches import cv2_imshow



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):
    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)

    input_semaphore.acquire()
    input_queue.put(None)
    cap.release()

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

        if frame is None:
            input_queue.put(None)
            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)
    producer_thread = threading.Thread(target=producer, args=(input_video_path,input_queue,input_semaphore))
    producer_thread.start()

    num_consumers = 2
    consumer_threads = []
    for _ in range(num_consumers):
        consumer_thread = threading.Thread(target=consumer,args=(input_queue,input_semaphore,output_semaphore,output_queue))
        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"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")
    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)")