This is the script that I use to generate black and white videos with differences, it's still in early phase, I change it a lot.
If you want to run it locally you will need:
Python might not be perfect programming language for this problem, however it is easy to run, and everyone can review source code.
import cv2
import numpy as np
import time
import os
import shutil
from concurrent.futures import ThreadPoolExecutor, as_completed
import mimetypes
input_folder = '/Users/mac/Downloads/rawmedia'
output_folder = '/Users/mac/Downloads/processed_videos/'
def is_video_file(file_path):
mime_type, _ = mimetypes.guess_type(file_path)
return mime_type and mime_type.startswith('video')
def process_video(file_name, input_video_path, output_video_path, output_changes_video_path):
cap = cv2.VideoCapture(input_video_path)
fps = cap.get(cv2.CAP_PROP_FPS)
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
out = cv2.VideoWriter(output_video_path, fourcc, fps, (width, height), isColor=False)
out_changes = cv2.VideoWriter(output_changes_video_path, fourcc, fps, (width, height), isColor=False)
font = cv2.FONT_HERSHEY_SIMPLEX
font_scale = 1
color = (255) # White color for grayscale frames
thickness = 2
position = (10, height - 10) # Bottom-left corner of the frame
ret, prev_frame = cap.read()
if ret:
prev_frame_gray = cv2.cvtColor(prev_frame, cv2.COLOR_BGR2GRAY)
frame_count = 0
while ret:
ret, current_frame = cap.read()
if not ret:
break
current_frame_gray = cv2.cvtColor(current_frame, cv2.COLOR_BGR2GRAY)
# Compute the absolute difference between frames
diff_frame = cv2.absdiff(current_frame_gray, prev_frame_gray)
_, diff_thresh = cv2.threshold(diff_frame, 10, 255, cv2.THRESH_BINARY)
# Write to the main output video
out.write(diff_thresh)
# If there are any changes (white pixels in the diff), write to the changes video
if np.any(diff_thresh):
# Calculate timestamp in minutes, seconds, and frame
timestamp = frame_count / fps
minutes = int(timestamp // 60)
seconds = int(timestamp % 60)
frame_in_second = frame_count % int(fps) # Frame number within the current second
timestamp_text = f"Time: {minutes:02d}:{seconds:02d}.{frame_in_second:02d} (mm:ss.ff)"
frame_with_timestamp = cv2.putText(diff_thresh.copy(), timestamp_text, position, font, font_scale, color,
thickness, cv2.LINE_AA)
out_changes.write(frame_with_timestamp)
prev_frame_gray = current_frame_gray
progress = (frame_count / total_frames) * 100
print(f"\r[{file_name}] Processing frame {frame_count}/{total_frames} ({progress:.2f}% complete)", end="")
frame_count += 1
cap.release()
out.release()
out_changes.release()
print(f"\n[{file_name}] Finished processing video. Output saved as: {output_video_path}")
print(f"[{file_name}] Changes-only video with timestamps saved as: {output_changes_video_path}")
def create_subdirectory_and_move_files(file_name, input_video_path, output_video_path, output_changes_video_path):
subdirectory = os.path.join(output_folder, file_name)
os.makedirs(subdirectory, exist_ok=True)
shutil.move(output_video_path, os.path.join(subdirectory, os.path.basename(output_video_path)))
shutil.move(output_changes_video_path, os.path.join(subdirectory, os.path.basename(output_changes_video_path)))
shutil.move(input_video_path, os.path.join(subdirectory, os.path.basename(input_video_path)))
def handle_file(file_name):
input_video_path = os.path.join(input_folder, file_name)
output_video_path = os.path.join(input_folder, 'output_diff_' + file_name)
output_changes_video_path = os.path.join(input_folder, 'output_only_changes_' + file_name)
print(f"\n[{file_name}] Starting processing.")
process_video(file_name, input_video_path, output_video_path, output_changes_video_path)
create_subdirectory_and_move_files(file_name, input_video_path, output_video_path, output_changes_video_path)
return file_name
def main():
start_time = time.time()
file_list = [file for file in os.listdir(input_folder) if is_video_file(os.path.join(input_folder, file))]
with ThreadPoolExecutor() as executor:
futures = {executor.submit(handle_file, file_name): file_name for file_name in file_list}
for future in as_completed(futures):
file_name = futures[future]
try:
future.result()
print(f"\n[{file_name}] Processing completed successfully.")
except Exception as exc:
print(f"\n[{file_name}] generated an exception: {exc}")
elapsed_time = time.time() - start_time
print(f"\nTotal time taken: {elapsed_time:.2f} seconds")
if __name__ == '__main__':
main()