272 lines
8.4 KiB
Python
272 lines
8.4 KiB
Python
import os
|
|
import re
|
|
import subprocess
|
|
|
|
import bpy
|
|
|
|
from vse_toolbox import bl_utils
|
|
|
|
|
|
def get_scene_scores(strip, frame_start=None, frame_end=None, crop=None):
|
|
"""Run ffmpeg over the strip and return the scene score of every frame.
|
|
|
|
The command mirrors the ANIMATED detection pass (same crop and seek), but
|
|
uses the metadata filter to print each frame's scene score instead of
|
|
selecting frames above a threshold.
|
|
|
|
Args:
|
|
strip (bpy.types.Strip): blender sequence strip to score.
|
|
frame_start (int, optional): first timeline frame to score.
|
|
Defaults to None.
|
|
frame_end (int, optional): last timeline frame to score.
|
|
Defaults to None.
|
|
crop (list, optional): crop values [min_x, max_x, min_y, max_y].
|
|
Defaults to None.
|
|
|
|
Returns:
|
|
dict: timeline frames -> scene score (float, from 0 to 1).
|
|
"""
|
|
|
|
path = bl_utils.abspath(strip.filepath)
|
|
fps = bpy.context.scene.render.fps
|
|
|
|
if frame_start is None:
|
|
frame_start = 0
|
|
if frame_end is None:
|
|
frame_end = strip.frame_duration
|
|
|
|
start_time = (frame_start - strip.frame_start) / fps
|
|
end_time = frame_end / fps
|
|
|
|
if crop is None:
|
|
crop = [0, 0, 0, 0]
|
|
|
|
crop_expr = f"crop=iw-{crop[0]}-{crop[1]}:ih-{crop[2]}-{crop[3]}:{crop[0]}:{crop[-1]}"
|
|
|
|
command = [
|
|
'ffmpeg',
|
|
'-nostats',
|
|
'-i',
|
|
str(path),
|
|
'-ss', str(start_time),
|
|
'-t', str(end_time),
|
|
'-vf',
|
|
f"{crop_expr}, select='gte(scene,0)', metadata=print",
|
|
'-f',
|
|
'null',
|
|
'-'
|
|
]
|
|
|
|
print(command)
|
|
|
|
process = subprocess.Popen(
|
|
command,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
universal_newlines=True)
|
|
|
|
scores = {}
|
|
offset_frame = None
|
|
for line in process.stdout:
|
|
match = re.search(r'pts_time:([\d.]+)', line)
|
|
if match:
|
|
offset_frame = round(float(match.group(1)) * fps)
|
|
|
|
match = re.search(r'lavfi.scene_score=([\d.]+)', line)
|
|
if match and offset_frame is not None:
|
|
# pts_time restarts at the -ss seek point, shift back to timeline frames.
|
|
scores[offset_frame + int(frame_start)] = float(match.group(1))
|
|
|
|
process.wait()
|
|
return scores
|
|
|
|
|
|
def compute_scene_threshold(scene_scores, known_frames, noise_floor=0.01):
|
|
"""Compute a scene-change threshold that detects every known cut frame.
|
|
|
|
The threshold is placed just below the weakest known cut. When there is a
|
|
clear gap under it (highest non-known score at least 10% lower), the
|
|
threshold sits in the middle of that gap. Otherwise it defaults to 90% of
|
|
the weakest known cut score to stay robust against small score shifts.
|
|
|
|
Known cuts scoring below `noise_floor` are considered undetectable by the
|
|
select filter (dissolves or misread frames) and are skipped: pulling the
|
|
threshold under the noise level would split on noise.
|
|
|
|
Args:
|
|
scene_scores (dict): timeline frames -> scene score.
|
|
known_frames (list): timeline frames with manually checked cuts.
|
|
noise_floor (float, optional): score under which a known cut is
|
|
considered noise. Defaults to 0.01.
|
|
|
|
Returns:
|
|
tuple: (threshold or None if no usable known frame was found,
|
|
list of known frames missing from the scores,
|
|
list of known frames skipped as noise).
|
|
"""
|
|
|
|
known_frames = list(known_frames)
|
|
missing = [frame for frame in known_frames if frame not in scene_scores]
|
|
found = {frame: scene_scores[frame] for frame in known_frames if frame in scene_scores}
|
|
if not found:
|
|
return None, missing, []
|
|
|
|
weak = [frame for frame, score in found.items() if score < noise_floor]
|
|
usable = {frame: score for frame, score in found.items() if score >= noise_floor}
|
|
if not usable:
|
|
return None, missing, weak
|
|
|
|
min_known = min(usable.values())
|
|
known = set(usable)
|
|
below = [score for frame, score in scene_scores.items()
|
|
if frame not in known and score < min_known]
|
|
max_below = max(below) if below else 0.0
|
|
|
|
if min_known - max_below > 0.1 * min_known:
|
|
threshold = (min_known + max_below) / 2
|
|
else:
|
|
threshold = 0.9 * min_known
|
|
|
|
return threshold, missing, weak
|
|
|
|
|
|
def detect_scene_change(strip, movie_type="ANIMATED", threshold=0.5, frame_start=None, frame_end=None, crop=None):
|
|
"""Launch ffmpeg command to detect changing frames from a movie strip.
|
|
|
|
Args:
|
|
movie_strip (bpy.types.Strip): blender sequence strip to detect changes.
|
|
threshold (float): value of the detection factor (from 0 to 1).
|
|
frame_start (int, optional): first frame to detect.
|
|
Defaults to None.
|
|
frame_end (int, optional): last frame to detect.
|
|
Defaults to None.
|
|
|
|
Returns:
|
|
str: ffmpeg command log.
|
|
"""
|
|
|
|
path = bl_utils.abspath(strip.filepath)
|
|
fps = bpy.context.scene.render.fps
|
|
|
|
if frame_start is None:
|
|
frame_start = 0
|
|
if frame_end is None:
|
|
frame_end = strip.frame_duration
|
|
|
|
frame_start = frame_start - strip.frame_start
|
|
|
|
start_time = frame_start/fps
|
|
end_time = frame_end/fps
|
|
|
|
#frame_start += strip.frame_offset_start
|
|
#frame_end -= strip.frame_offset_end
|
|
|
|
if movie_type == 'ANIMATED':
|
|
return select_generator(path, start_time=start_time, end_time=end_time, threshold=threshold, crop=crop)
|
|
|
|
elif movie_type == 'STILL':
|
|
return freeze_detect_generator(path, start_time=start_time, end_time=end_time, threshold=threshold, crop=crop)
|
|
|
|
else:
|
|
raise Exception(f'movie_type: {movie_type} not implemented')
|
|
|
|
|
|
def freeze_detect_generator(path, start_time, end_time, threshold=0.005, crop=None):
|
|
"""Generate the ffmpeg command which detect change from a movie.
|
|
|
|
Args:
|
|
path (_type_): path to detect changes.
|
|
threshold (_type_): value of the detection factor (from 0 to 1).
|
|
frame_start (_type_): first frame to detect.
|
|
frame_end (_type_): last frame to detect.
|
|
fps (_type_): framerate of the movie.
|
|
|
|
Returns:
|
|
list: ffmpeg command as list for subprocess module.
|
|
"""
|
|
|
|
if crop is None:
|
|
crop = [0, 0, 0, 0]
|
|
|
|
crop_expr = f"crop=iw-{crop[0]}-{crop[1]}:ih-{crop[2]}-{crop[3]}:{crop[0]}:{crop[-1]}"
|
|
|
|
command = [
|
|
'ffmpeg',
|
|
'-nostats',
|
|
'-i',
|
|
str(path),
|
|
'-ss', str(start_time),
|
|
'-t', str(end_time),
|
|
'-vf',
|
|
f"{crop_expr}, freezedetect=n={threshold}:d=0.01,metadata=print", #,
|
|
'-f',
|
|
'null',
|
|
'-'
|
|
]
|
|
|
|
print(command)
|
|
|
|
process = subprocess.Popen(
|
|
command,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
universal_newlines=True)
|
|
|
|
def parse_split_time(log):
|
|
print(log)
|
|
timecodes = re.findall(r'freeze_end: ([\d.]+)', log)
|
|
if timecodes:
|
|
return round(float(timecodes[0]) * bpy.context.scene.render.fps)
|
|
|
|
return filter(None, (parse_split_time(x) for x in process.stdout))
|
|
|
|
def select_generator(path, start_time, end_time, threshold, crop=None):
|
|
"""Generate the ffmpeg command which detect change from a movie.
|
|
|
|
Args:
|
|
path (_type_): path to detect changes.
|
|
threshold (_type_): value of the detection factor (from 0 to 1).
|
|
frame_start (_type_): first frame to detect.
|
|
frame_end (_type_): last frame to detect.
|
|
fps (_type_): framerate of the movie.
|
|
|
|
Returns:
|
|
list: ffmpeg command as list for subprocess module.
|
|
"""
|
|
|
|
if crop is None:
|
|
crop = [0, 0, 0, 0]
|
|
|
|
crop_expr = f"crop=iw-{crop[0]}-{crop[1]}:ih-{crop[2]}-{crop[3]}:{crop[0]}:{crop[-1]}"
|
|
|
|
command = [
|
|
'ffmpeg',
|
|
'-nostats',
|
|
'-i',
|
|
str(path),
|
|
'-ss', str(start_time),
|
|
'-t', str(end_time),
|
|
'-vf',
|
|
f"{crop_expr}, select='gt(scene, {threshold})', showinfo",
|
|
'-f',
|
|
'null',
|
|
'-'
|
|
]
|
|
|
|
print(command)
|
|
|
|
process = subprocess.Popen(
|
|
command,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
universal_newlines=True)
|
|
|
|
def parse_split_time(log):
|
|
print(log)
|
|
timecodes = re.findall(r'pts_time:([\d.]+)', log)
|
|
if timecodes:
|
|
return round(float(timecodes[0]) * bpy.context.scene.render.fps)
|
|
|
|
return filter(None, (parse_split_time(x) for x in process.stdout))
|
|
|