python multiprocess

main-nomp.py

import cv2
import time
import sys
import logging


def get_frames_parts(total_frames=0):
    max_workers = 4
    part = total_frames / max_workers
    results = [(i * part, (i + 1) * part) for i in range(max_workers)]
    logging.info(f"frame division: {results}")
    return results


def extract_pixel(video_path, start_frame, end_frame):
    logging.info(
        f"starting from {start_frame} to {end_frame} "
    )
    capture = cv2.VideoCapture(video_path)  # open the video using OpenCV
    total = int(capture.get(cv2.CAP_PROP_FRAME_COUNT))
    if start_frame < 0:  # if start_frame isn't specified lets assume 0
        start_frame = 0
    if end_frame < 0:  # if end_frame isn't specified assume the end of the video
        end_frame = total
    capture.set(1, start_frame)  # set the starting frame of the capture
    frame = start_frame  # keep track of which frame we are up to, starting from start
    sum_pixel_values = 0
    p1_start = time.perf_counter()
    while frame <= end_frame:  # lets loop through the frames until the end
        f1_start = time.time()
        success, captured_frame = capture.read()  # read a frame from the capture
        current_frame = capture.get(cv2.CAP_PROP_POS_FRAMES)

        if (
            captured_frame is None and success is False
        ):  # if we get a bad return flag or the image we read is None, lets not save
            break  # skip

        if not (start_frame <= current_frame <= end_frame):
            break  # out of range, then skip

        # red = captured_frame.item((10, 10, 2))
        red = 0
        sum_pixel_values += red
        f1_end = time.time()
        # print(f"frame: {frame} red: {red} | saved_count: {saved_count}")
        logging.info(
            f"frame: {frame} | part: {start_frame} - {end_frame} | done in {round(f1_end - f1_start, 3)} seconds"
        )
        frame += 1  # increment our frame count
    p1_end = time.perf_counter()
    logging.debug(
        f"part: {start_frame} - {end_frame} | done in {round(p1_end - p1_start, 3)} seconds | sum: {sum_pixel_values}"
    )
    capture.release()  # after the while has finished close the capture
    return sum_pixel_values


def calculate_sum_pixel(video_path):
    if video_path is None:
        logging.error("no video found")
        sys.exit(0)
    capture = cv2.VideoCapture(video_path)  # load the video
    total = int(capture.get(cv2.CAP_PROP_FRAME_COUNT))  # get its total frame count
    logging.info(f"total frame: {total}")
    capture.release()  # release the capture straight away
    parts = get_frames_parts(total)
    results = []
    # process parts
    for part in parts:
        result = extract_pixel(video_path=video_path, start_frame=part[0], end_frame=part[1])
        results.append(result)
    return sum(results)


if __name__ == "__main__":
    VIDEO_PATH = "test_video.mp4"
    logging.basicConfig(
        filename='singleprocess.log', filemode='w',
        format="%(levelname)s: %(asctime)s | %(message)s",
        datefmt="%I:%M:%S %p",
        level=logging.DEBUG,
    )
    logging.debug(f"running on single process")
    t1_start = time.perf_counter()
    result = calculate_sum_pixel(video_path=VIDEO_PATH)
    t1_stop = time.perf_counter()
    logging.info(f"Elapsed time in seconds: {t1_stop - t1_start}")
    logging.info(f"sum of red pixel values: {result}")

main.py

import cv2
import time
import multiprocessing
import concurrent.futures
import sys
import logging


def get_frames_parts(total_frames=0):
    max_workers = multiprocessing.cpu_count()
    part = total_frames / max_workers
    results = [(i * part, (i + 1) * part) for i in range(max_workers)]
    logging.info(f"frame division: {results}")
    return results


def extract_pixel(video_path, start_frame, end_frame):
    worker_pid = multiprocessing.current_process().pid
    logging.info(
        f"starting worker PID: {worker_pid} from {start_frame} to {end_frame} "
    )
    capture = cv2.VideoCapture(video_path)  # open the video using OpenCV
    total = int(capture.get(cv2.CAP_PROP_FRAME_COUNT))
    if start_frame < 0:  # if start_frame isn't specified lets assume 0
        start_frame = 0
    if end_frame < 0:  # if end_frame isn't specified assume the end of the video
        end_frame = total
    capture.set(1, start_frame)  # set the starting frame of the capture
    frame = start_frame  # keep track of which frame we are up to, starting from start
    sum_pixel_values = 0
    p1_start = time.perf_counter()
    while frame <= end_frame:  # lets loop through the frames until the end
        f1_start = time.time()
        success, captured_frame = capture.read()  # read a frame from the capture
        current_frame = capture.get(cv2.CAP_PROP_POS_FRAMES)

        if (
            captured_frame is None and success is False
        ):  # if we get a bad return flag or the image we read is None, lets not save
            break  # skip

        if not (start_frame <= current_frame <= end_frame):
            break  # out of range, then skip

        red = captured_frame.item((10, 10, 2))
        sum_pixel_values += red
        f1_end = time.time()
        # print(f"frame: {frame} red: {red} | saved_count: {saved_count}")
        logging.info(
            f"PID:{worker_pid} | frame: {frame} | part: {start_frame} - {end_frame} | done in {round(f1_end - f1_start, 3)} seconds"
        )
        frame += 1  # increment our frame count
    p1_end = time.perf_counter()
    logging.debug(
        f"PID:{worker_pid} | part: {start_frame} - {end_frame} | done in {round(p1_end - p1_start, 3)} seconds | sum: {sum_pixel_values}"
    )
    capture.release()  # after the while has finished close the capture
    return sum_pixel_values


def calculate_sum_pixel(video_path=None):
    if video_path is None:
        logging.error("no video found")
        sys.exit(0)
    capture = cv2.VideoCapture(video_path)  # load the video
    total = int(capture.get(cv2.CAP_PROP_FRAME_COUNT))  # get its total frame count
    logging.info(f"total frame: {total}")
    capture.release()  # release the capture straight away
    parts = get_frames_parts(total)
    sum = 0
    # execute across multiple cpu cores to speed up processing, get the count automatically
    with concurrent.futures.ProcessPoolExecutor(
        max_workers=multiprocessing.cpu_count()
    ) as executor:
        futures = [
            executor.submit(extract_pixel, video_path, part[0], part[1])
            for part in parts
        ]
        for i, f in enumerate(concurrent.futures.as_completed(futures), start=1):
            # print(f"sum of part{i}: {f.result()}")
            sum += f.result()
    return sum


if __name__ == "__main__":
    # multiprocessing.log_to_stderr(logging.DEBUG)
    VIDEO_PATH = "test_video.mp4"
    logging.basicConfig(
        filename='multiprocess.log', filemode='w',
        format="%(levelname)s: %(asctime)s | %(message)s",
        datefmt="%I:%M:%S %p",
        level=logging.DEBUG,
    )
    logging.debug(f"running on multi process")
    t1_start = time.perf_counter()
    result = calculate_sum_pixel(video_path=VIDEO_PATH)
    t1_stop = time.perf_counter()
    logging.info(f"Elapsed time in seconds: {t1_stop - t1_start}")
    logging.info(f"sum of red pixel values: {result}")