method2.py

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

input_path = "2.mp4"
output_path = "output.mp4"

selected_filters = ["invert"]

MAX_QUEUE_SIZE = 10
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)

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):
    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():
    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):
    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):
    producer_thread = threading.Thread(target=producer, args=(input_video_path,))
    producer_thread.start()

    num_consumers = 2
    consumer_threads = []
    for _ in range(num_consumers):
        consumer_thread = threading.Thread(target=consumer)
        consumer_thread.start()
        consumer_threads.append(consumer_thread)

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

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

    print("Video processing complete.")

def worker(task_queue, result_queue):
    while True:
        video_path = task_queue.get()
        if video_path is None:
            break

        output_path = "output.mp4"
        process_video(video_path, output_path)

        result_queue.put(f"Processed: {video_path}")

if __name__ == "__main__":
    now = time.time()
    process_video(input_path, output_path)
    print(f"Processing Duration: {time.time() - now}(s)")