File size: 5,600 Bytes
dde1cc7 acbe906 dde1cc7 acbe906 dde1cc7 acbe906 dde1cc7 acbe906 dde1cc7 acbe906 dde1cc7 acbe906 dde1cc7 acbe906 dde1cc7 acbe906 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 | from scenedetect import VideoManager, SceneManager
from scenedetect.detectors import ContentDetector
import cv2
import os
import numpy as np
import ffmpeg
from moviepy.editor import VideoFileClip
import time
from utils import read_files
from grok_analyze import analyze_image_with_grok_f
import json
import random
import string
import random
def extract_scenes_scenedetect(video_path, threshold=30.0, min_scene_duration=5.0):
vid_obj={"type":"video", "path":video_path}
vid_obj["scenes"]=[ ]
# Initialize video and scene manager
video = VideoManager([video_path])
scene_manager = SceneManager()
scene_manager.add_detector(ContentDetector(threshold=threshold))
# Detect scenes
video.start()
scene_manager.detect_scenes(video)
scene_list = scene_manager.get_scene_list()
# Get video FPS for time calculations
cap = cv2.VideoCapture(video_path)
fps = cap.get(cv2.CAP_PROP_FPS)
min_frames = int(min_scene_duration * fps) # Convert 25 seconds to frames
# Filter scenes to ensure at least 25 seconds (min_frames) between them
filtered_scenes = []
last_frame = -min_frames # Allow the first scene to start at frame 0
for scene in scene_list:
print(scene)
start_frame = scene[0].get_frames()
if start_frame >= last_frame + min_frames:
filtered_scenes.append(scene)
last_frame = start_frame
print(f"Total scenes detected: {len(scene_list)}")
print(f"Filtered scenes (at least {min_scene_duration} seconds apart): {len(filtered_scenes)}")
if not filtered_scenes:
print("No scenes detected; extracting the first frame of the video.")
cap.set(cv2.CAP_PROP_POS_FRAMES, 0)
ret, frame = cap.read()
if ret:
# Calculate timestamp for frame 0
timestamp = 0 / fps
minutes = int(timestamp // 60)
seconds = int(timestamp % 60)
milliseconds = int((timestamp % 1) * 1000)
print(f"First Frame - Timestamp: {minutes:02d}:{seconds:02d}.{milliseconds:03d}")
# Analyze frame and store scene info
description = analyze_image_with_grok_f(frame)
scene_info = {
"timestamp": f"{minutes:02d}:{seconds:02d}.{milliseconds:03d}",
"description": description
}
vid_obj["scenes"].append(scene_info)
else:
print("Failed to read the first frame of the video.")
cap.release()
video.release()
return vid_obj
else:
# Extract frames from filtered scenes
for i, scene in enumerate(filtered_scenes):
# Calculate and format timestamp
start_frame = scene[0].get_frames()
timestamp = start_frame / fps
minutes = int(timestamp // 60)
seconds = int(timestamp % 60)
milliseconds = int((timestamp % 1) * 1000)
print(f"Scene {i:04d} - Timestamp: {minutes:02d}:{seconds:02d}.{milliseconds:03d}")
# Seek to the start of the scene
cap.set(cv2.CAP_PROP_POS_FRAMES, start_frame)
ret, frame = cap.read()
description=analyze_image_with_grok_f(frame)
scene={"timestamp":f"{minutes:02d}:{seconds:02d}.{milliseconds:03d}", "description":description}
vid_obj["scenes"].append(scene)
# if ret:
# output_path = os.path.join(output_folder, f"scene_{i:04d}.jpg")
# cv2.imwrite(output_path, frame)
print(vid_obj)
cap.release()
video.release()
return vid_obj
def get_metadata(all_files):
all_files_info=[]
for file_path in all_files:
print(file_path)
if file_path.split(".")[-1] in ["MOV","mp4"]:
vid_info=extract_scenes_scenedetect(file_path)
all_files_info.append(vid_info)
random_string = ''.join(random.choices(string.ascii_lowercase + string.digits, k=8))
file_name = f"all_files_info_{random_string}.json"
with open(file_name, "w") as file:
json.dump(all_files_info, file, indent=4)
# upload_to_s3(file_name,"bucket","s3key")
return all_files_info,file_name
def get_scenes_metadata(metadata, no_scenes=10):
selected = []
while len(selected) < no_scenes and metadata: # Pick 3, stop if list empties
item = random.choice(metadata)
selected.append(item)
metadata.remove(item)
# print(selected)
# print(len(selected))
return(selected)
if __name__ == "__main__":
with open('/Users/georgia.bucea/products/ShortsAI/all_files_metadata.json', 'r') as file:
data = json.load(file)
print(len(data))
get_scenes_metadata(data)
# all_files_info=[]
# # dir_path="/Users/georgia.bucea/products/ShortsAI/21_aug"
# # all_files=read_files(dir_path)
# dir_path="/Users/georgia.bucea/products/ShortsAI/16_aug"
# # all_files2=read_files(dir_path)
# all_files=read_files(dir_path)
# # all_files=all_files+all_files2
# for file in all_files:
# file_path=dir_path+"/"+file
# if file_path.split(".")[-1] in ["MOV","mp4"]:
# vid_info=extract_scenes_scenedetect(file_path)
# all_files_info.append(vid_info)
# with open("all_files_info_21_aug_16_aug.json", "w") as file:
# json.dump(all_files_info, file, indent=4)
|