import datetime from glob import glob import cv2 import os from tqdm import tqdm def mp4_to_png(input_path: str, save_path: str) -> str: """ Converts mp4 to pngs for each frame of the video. Args: input_path is the path to the mp4 file, save_path is the directory to save the frames. Returns: save_path, fps the number of frames per second. """ # get frames per second fps = int(cv2.VideoCapture(input_path).get(cv2.CAP_PROP_FPS)) # run subprocess to convert mp4 to pngs os.system(f"ffmpeg -i {input_path} -vf fps={fps} {save_path}/frame%08d.png") return fps def frame_to_timestamp(frame_number: int, fps: int): # Calculate the timestamp in seconds timestamp_seconds = frame_number / fps # Convert the timestamp to hh:mm:ss minutes, seconds = divmod(timestamp_seconds, 60) hours, minutes = divmod(minutes, 60) # print("Timestamp:", hours, ":", minutes, ":", seconds) timestamp = f"{int(hours):02d}:{int(minutes):02d}:{seconds:06.3f}" return timestamp def vid_stitcher(frames_dir: str, output_path: str, fps: int = 30) -> str: """ Takes a list of frames as numpy arrays and writes them to a video file. """ # Get the list of frames frame_list = sorted(glob(os.path.join(frames_dir, 'frame*.png'))) # Prepare the VideoWriter frame = cv2.imread(frame_list[0]) height, width, _ = frame.shape fourcc = cv2.VideoWriter_fourcc(*'mp4v') out = cv2.VideoWriter(output_path, fourcc, fps, (width, height)) # Use multithreading to read frames faster from concurrent.futures import ThreadPoolExecutor with ThreadPoolExecutor() as executor: frames = list(executor.map(cv2.imread, frame_list)) # Write frames to the video with tqdm(total=len(frame_list), desc='Stitching frames') as pbar: for frame in frames: out.write(frame) pbar.update(1) return output_path