honney/Merge-Anime-Lanuages
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Cutter works? 0.2

Browse Source
This commit is contained in:
Hannes committed 2026-03-16 18:50:17 +01:00
1 parent 3ba9c2a003
commit 69e9a9037c
2 files changed
+336 -139
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

cc_scene_change.py
+151
View File
@@ -0,0 +1,151 @@
import subprocess
import numpy as np
import cv2
import torch
import torch.nn.functional as F
from scipy import signal
TARGET_FPS = 30
def get_device():
if torch.cuda.is_available():
device = torch.device("cuda")
print("GPU detected")
print("Device:", torch.cuda.get_device_name(0))
print("ROCm build:", torch.version.hip is not None)
return device
print("GPU not available, falling back to CPU")
return torch.device("cpu")
device = get_device()
def ffmpeg_frame_stream(path):
cmd = [
"ffmpeg",
"-i", path,
"-vf", "fps=30,scale=64:36,format=bgr24",
"-f", "rawvideo",
"-pix_fmt", "bgr24",
"-"
]
proc = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
bufsize=10**8
)
frame_size = 64 * 36 * 3
while True:
raw = proc.stdout.read(frame_size)
if len(raw) < frame_size:
break
frame = np.frombuffer(raw, np.uint8).reshape((36, 64, 3))
yield frame
proc.stdout.close()
proc.wait()
def compute_signal(video_path):
prev_blocks = None
signal_out = []
for frame in ffmpeg_frame_stream(video_path):
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
t = torch.from_numpy(gray).float().to(device) / 255.0
# reshape for conv
t = t.unsqueeze(0).unsqueeze(0)
# block averages using avg pool
blocks = F.avg_pool2d(t, kernel_size=8, stride=8)
blocks = blocks.squeeze()
if prev_blocks is None:
prev_blocks = blocks
signal_out.append(0)
continue
# block difference energy
diff = torch.abs(blocks - prev_blocks).sum()
# edge energy
sobel_x = torch.tensor(
[[-1,0,1],[-2,0,2],[-1,0,1]],
dtype=torch.float32, device=device
).view(1,1,3,3)
sobel_y = torch.tensor(
[[-1,-2,-1],[0,0,0],[1,2,1]],
dtype=torch.float32, device=device
).view(1,1,3,3)
gx = F.conv2d(t, sobel_x, padding=1)
gy = F.conv2d(t, sobel_y, padding=1)
edge_energy = torch.mean(torch.sqrt(gx**2 + gy**2))
value = diff + 0.5 * edge_energy
signal_out.append(value.item())
prev_blocks = blocks
signal_out = np.array(signal_out)
# temporal derivative
signal_out = np.diff(signal_out, prepend=signal_out[0])
return signal_out
def normalize(sig):
return (sig - np.mean(sig)) / (np.std(sig) + 1e-6)
def find_offset(video_a, video_b):
sig_a = compute_signal(video_a)
sig_b = compute_signal(video_b)
sig_a = normalize(sig_a)
sig_b = normalize(sig_b)
corr = signal.correlate(sig_a, sig_b, mode="full")
lags = signal.correlation_lags(len(sig_a), len(sig_b), mode="full")
best = lags[np.argmax(corr)]
seconds = best / TARGET_FPS
return best, seconds
def main():
video_a = "/home/honney/mount/Media/tmp/serien/ger_low_quality/Black Bullet/Black Bullet Season 01 E03 Kinder des Schicksals.mp4"
video_b = "/home/honney/mount/Media/tmp/serien/high_quality/Black Bullet (Season 1)/Season 01/Black Bullet - S01E03.mkv"
frames, seconds = find_offset(video_a, video_b)
print("offset frames:", frames)
print("offset seconds:", seconds)
if __name__ == "__main__":
main()
cut_to_length.py
+185 -139
View File
@@ -12,6 +12,7 @@ import tempfile
import shutil
DEBUG = True
TARGET_FPS = 30
def run(cmd):
"""
@@ -36,32 +37,112 @@ def run(cmd):
print(e.stdout)
raise
# def get_video_signal(path):
# """
# Compute a simple motion signal from a video.
# The function reads a video frame-by-frame and measures the amount of
# pixel change between consecutive frames. Each frame is resized and
# converted to grayscale before calculating the absolute difference
# from the previous frame. The summed pixel differences form a 1-D
# motion signal representing how much visual change occurs over time.
# Args:
# path: Path to the video file to analyze.
# Returns:
# A NumPy array where each element represents the magnitude of
# frame-to-frame change for a given frame. Higher values indicate
# stronger motion or scene changes.
# Raises:
# RuntimeError: If the video cannot be opened or read.
# Notes:
# Frames are resized to 160×90 before processing to reduce noise
# and computational cost.
# """
# cap = cv2.VideoCapture(path)
# motion_signal = []
# prev_gray = None
# while True:
# ret, frame = cap.read()
# if not ret:
# break
# # 1. Shrink the frame (makes it faster and removes noise)
# small_frame = cv2.resize(frame, (160, 90))
# gray = cv2.cvtColor(small_frame, cv2.COLOR_BGR2GRAY)
# if prev_gray is not None:
# # 2. Calculate the difference between this frame and the last
# # This detects the "movement" or "cuts"
# diff = cv2.absdiff(gray, prev_gray)
# motion_signal.append(np.sum(diff))
# else:
# motion_signal.append(0)
# prev_gray = gray
# cap.release()
# return np.array(motion_signal)
# def find_offset(video_a: Path, video_b: Path) -> int:
# """
# Estimate the temporal offset between two videos.
# The function extracts a motion signal from each video using
# `get_video_signal` and then computes the cross-correlation between
# the two signals. The lag corresponding to the maximum correlation
# indicates the frame offset where the videos align best.
# Args:
# video_a: Path to the first video file.
# video_b: Path to the second video file.
# Returns:
# The offset in frames between the two videos. A positive value
# means `video_b` starts later than `video_a`, while a negative
# value means `video_b` starts earlier.
# Raises:
# RuntimeError: If either input path does not point to a valid file.
# Notes:
# Before computing the correlation, both signals are normalized
# (mean subtracted and divided by standard deviation) to make the
# comparison robust to differences in overall motion intensity.
# """
# if not os.path.isfile(video_a):
# raise RuntimeError("video_a is not a file")
# if not os.path.isfile(video_b):
# raise RuntimeError("video_b is not a file")
# # print("Extracting signals (this may take a moment)...")
# sig_a = get_video_signal(video_a)
# # print("sig_a finished")
# sig_b = get_video_signal(video_b)
# # print("sig_b finished")
# # Normalize signals (subtract mean, divide by std dev) to improve correlation
# sig_a = (sig_a - np.mean(sig_a)) / (np.std(sig_a) + 1e-5)
# sig_b = (sig_b - np.mean(sig_b)) / (np.std(sig_b) + 1e-5)
# # Perform Cross-Correlation
# correlation = signal.correlate(sig_a, sig_b, mode='full')
# lags = signal.correlation_lags(len(sig_a), len(sig_b), mode='full')
# # plt.plot(lags)
# # The peak of the correlation is our offset in frames
# best_lag = lags[np.argmax(correlation)]
# return best_lag
def get_video_signal(path):
"""
Compute a simple motion signal from a video.
The function reads a video frame-by-frame and measures the amount of
pixel change between consecutive frames. Each frame is resized and
converted to grayscale before calculating the absolute difference
from the previous frame. The summed pixel differences form a 1-D
motion signal representing how much visual change occurs over time.
Args:
path: Path to the video file to analyze.
Returns:
A NumPy array where each element represents the magnitude of
frame-to-frame change for a given frame. Higher values indicate
stronger motion or scene changes.
Raises:
RuntimeError: If the video cannot be opened or read.
Notes:
Frames are resized to 160×90 before processing to reduce noise
and computational cost.
"""
cap = cv2.VideoCapture(path)
fps = cap.get(cv2.CAP_PROP_FPS)
motion_signal = []
prev_gray = None
@@ -70,13 +151,10 @@ def get_video_signal(path):
if not ret:
break
# 1. Shrink the frame (makes it faster and removes noise)
small_frame = cv2.resize(frame, (160, 90))
gray = cv2.cvtColor(small_frame, cv2.COLOR_BGR2GRAY)
if prev_gray is not None:
# 2. Calculate the difference between this frame and the last
# This detects the "movement" or "cuts"
diff = cv2.absdiff(gray, prev_gray)
motion_signal.append(np.sum(diff))
else:
@@ -85,58 +163,39 @@ def get_video_signal(path):
prev_gray = gray
cap.release()
return np.array(motion_signal)
def find_offset(video_a: Path, video_b: Path) -> int:
"""
Estimate the temporal offset between two videos.
return np.array(motion_signal), fps
The function extracts a motion signal from each video using
`get_video_signal` and then computes the cross-correlation between
the two signals. The lag corresponding to the maximum correlation
indicates the frame offset where the videos align best.
Args:
video_a: Path to the first video file.
video_b: Path to the second video file.
def resample_signal(sig, original_fps, target_fps):
duration = len(sig) / original_fps
target_length = int(duration * target_fps)
Returns:
The offset in frames between the two videos. A positive value
means `video_b` starts later than `video_a`, while a negative
value means `video_b` starts earlier.
return signal.resample(sig, target_length)
Raises:
RuntimeError: If either input path does not point to a valid file.
Notes:
Before computing the correlation, both signals are normalized
(mean subtracted and divided by standard deviation) to make the
comparison robust to differences in overall motion intensity.
"""
if not os.path.isfile(video_a):
raise RuntimeError("video_a is not a file")
if not os.path.isfile(video_b):
raise RuntimeError("video_b is not a file")
def find_offset(video_a, video_b):
# print("Extracting signals (this may take a moment)...")
sig_a = get_video_signal(video_a)
# print("sig_a finished")
sig_b = get_video_signal(video_b)
# print("sig_b finished")
sig_a, fps_a = get_video_signal(video_a)
sig_b, fps_b = get_video_signal(video_b)
# Normalize signals (subtract mean, divide by std dev) to improve correlation
# Resample both signals to common FPS
sig_a = resample_signal(sig_a, fps_a, TARGET_FPS)
sig_b = resample_signal(sig_b, fps_b, TARGET_FPS)
# Normalize
sig_a = (sig_a - np.mean(sig_a)) / (np.std(sig_a) + 1e-5)
sig_b = (sig_b - np.mean(sig_b)) / (np.std(sig_b) + 1e-5)
# Perform Cross-Correlation
correlation = signal.correlate(sig_a, sig_b, mode='full')
lags = signal.correlation_lags(len(sig_a), len(sig_b), mode='full')
# plt.plot(lags)
# The peak of the correlation is our offset in frames
best_lag = lags[np.argmax(correlation)]
return best_lag
# Convert lag to seconds
offset_seconds = best_lag / TARGET_FPS
return best_lag, offset_seconds
def probe_video(path: Path):
"""Get video FPS, duration, codec, and stream start time."""
@@ -211,52 +270,30 @@ def mux_audio_into_output(Quality_Video, audio_file, output_dir):
def stick_together_audio(left_audio: Path, right_audio: Path, output_audio: Path) -> None:
"""
Concatenate two audio files end-to-end into a single file.
Args:
left_audio: Path to the first audio file.
right_audio: Path to the second audio file.
output_audio: Path to the resulting audio file. Can overwrite
one of the input files safely.
Notes:
Works for audio files with the same codec and format. Internally,
writes to a temporary file first, then moves it to `output_audio`
to avoid overwriting input files prematurely.
Concatenate two audio files so that left_audio plays directly
before right_audio in the output file.
"""
left_audio = Path(left_audio)
right_audio = Path(right_audio)
output_audio = Path(output_audio)
# Temporary output file in the same directory to allow atomic move
tmp_output = output_audio.with_name(f"{output_audio.stem}_tmp{output_audio.suffix}")
# Use stem instead of prefix to create a temporary file
output_audio_tmp = output_audio.parent / (output_audio.stem + "_temp" + output_audio.suffix)
# Create a temporary file list for FFmpeg
with tempfile.NamedTemporaryFile("w", delete=False, suffix=".txt") as f:
f.write(f"file '{left_audio.resolve()}'\n")
f.write(f"file '{right_audio.resolve()}'\n")
list_file = Path(f.name)
try:
run([
"ffmpeg",
"-y", # overwrite temporary output if exists
"-f", "concat", # concat demuxer
"-safe", "0", # allow absolute paths
"-i", str(list_file),
"-c", "copy", # copy streams without re-encoding
str(tmp_output)
"-y",
"-i", str(left_audio),
"-i", str(right_audio),
"-filter_complex", "[0:a][1:a]concat=n=2:v=0:a=1[a]",
"-map", "[a]",
"-c:a", "opus", # or "aac" if you prefer
"-strict", "-2",
str(output_audio_tmp)
])
# Move temporary output to the final location
shutil.move(str(tmp_output), str(output_audio))
finally:
# Clean up the temporary file list
if list_file.exists():
list_file.unlink()
# Remove temporary output if something went wrong
if tmp_output.exists() and not output_audio.exists():
tmp_output.unlink()
shutil.move(output_audio_tmp, output_audio)
def process_one(German_Video: Path, Quality_Video: Path, output_dir: Path):
# --- Probe video ---
@@ -264,16 +301,18 @@ def process_one(German_Video: Path, Quality_Video: Path, output_dir: Path):
fps2, duration2, _, vstart2 = probe_video(Quality_Video)
if fps != fps2:
raise ValueError("FPS of both Videos must be the same")
raise ValueError("FPS of both Videos must be the same (May not work yet, more testing needs to be done)")
if duration <= 0:
raise ValueError("Duration of German_Video must be greater than 0")
if duration2 <= 0:
raise ValueError("Duration of Quality_Video must be greater than 0")
# n1 = find_offset(German_Video, Quality_Video)
n1 = -1
n1, offset = find_offset(Quality_Video, German_Video)
print(f"Offset: {n1}")
if offset > duration or offset > duration2:
raise ValueError("Offset needs to be smaller than the video durations")
# print(f"fps: {fps}")
# print(f"Cutting audio from frame {n1} (timestamp {start:.6f}s) to timestamp {end:.6f}s")
@@ -281,35 +320,35 @@ def process_one(German_Video: Path, Quality_Video: Path, output_dir: Path):
german_audio_file = output_dir / f"{Path(Quality_Video).stem}_german_audio.mka"
quality_audio_file1 = output_dir / f"{Path(Quality_Video).stem}_original_audio1.mka"
quality_audio_file2 = output_dir / f"{Path(Quality_Video).stem}_original_audio2.mka"
if n1 < 0:
n1 = -n1
# Quality starts later than German
start = vstart + (n1 / fps)
if duration < (n1 / fps) + duration2:
# When the german audio is shorter than the original video
end = start + duration
if DEBUG: print(f"Cutting audio ({german_audio_file}) (from timestamp {start:.6f}s) to timestamp {end:.6f}s")
trim_audio(German_Video, start, end, german_audio_file)
# Get the end from the first original audio
start2 = vstart2 + (end - vstart)
end2 = vstart2 + duration2
if end2 > start2:
if DEBUG: print(f"Cutting audio ({quality_audio_file2}) (from timestamp {start2:.6f}s) to timestamp {end2:.6f}s")
trim_audio(Quality_Video, start2, end2, quality_audio_file2)
else:
end = start + duration2
if DEBUG: print(f"Cutting audio ({german_audio_file}) (from timestamp {start:.6f}s) to timestamp {end:.6f}s")
trim_audio(German_Video, start, end, german_audio_file)
else:
# if offset >= 0:
# offset = -offset
# # Quality starts later than German
# start = vstart + offset
# if duration < offset + duration2:
# # When the german audio is shorter than the original video
# end = start + duration
# if DEBUG: print(f"Cutting audio ({german_audio_file}) (from timestamp {start:.6f}s) to timestamp {end:.6f}s")
# trim_audio(German_Video, start, end, german_audio_file)
# # Get the end from the first original audio
# start2 = vstart2 + (end - vstart)
# end2 = vstart2 + duration2
# if end2 > start2:
# if DEBUG: print(f"Cutting audio ({quality_audio_file2}) (from timestamp {start2:.6f}s) to timestamp {end2:.6f}s")
# trim_audio(Quality_Video, start2, end2, quality_audio_file2)
# else:
# end = start + duration2
# if DEBUG: print(f"Cutting audio ({german_audio_file}) (from timestamp {start:.6f}s) to timestamp {end:.6f}s")
# trim_audio(German_Video, start, end, german_audio_file)
# else:
# German starts Later than Quality
# Get the start from the original Audio
start2 = vstart2
end2 = start2 + (n1 / fps)
end2 = start2 + offset
if end2 > start2:
if DEBUG: print(f"Cutting audio ({quality_audio_file1}) (from timestamp {start2:.6f}s) to timestamp {end2:.6f}s")
trim_audio(Quality_Video, start2, end2, quality_audio_file1)
###
if duration - (n1 / fps) < duration2:
# if duration - offset < duration2:
# When the german audio is shorter than the original video
start = vstart
end = start + duration
@@ -321,11 +360,19 @@ def process_one(German_Video: Path, Quality_Video: Path, output_dir: Path):
if end2 > start2:
if DEBUG: print(f"Cutting audio ({quality_audio_file2}) (from timestamp {start2:.6f}s) to timestamp {end2:.6f}s")
trim_audio(Quality_Video, start2, end2, quality_audio_file2)
else:
start = vstart
end = start + duration2
if DEBUG: print(f"Cutting audio ({german_audio_file}) (from timestamp {start:.6f}s) to timestamp {end:.6f}s")
trim_audio(German_Video, start, end, german_audio_file)
# else:
# start = vstart
# end = start + duration2
# if DEBUG: print(f"Cutting audio ({german_audio_file}) (from timestamp {start:.6f}s) to timestamp {end:.6f}s")
# trim_audio(German_Video, start, end, german_audio_file)
# german_audio_file2 = output_dir / f"{Path(Quality_Video).stem}_german_audio2.mka"
# german_audio_file3 = output_dir / f"{Path(Quality_Video).stem}_german_audio3.mka"
# if os.path.isfile(quality_audio_file1):
# stick_together_audio(quality_audio_file1, german_audio_file, german_audio_file2)
# if os.path.isfile(quality_audio_file2):
# stick_together_audio(german_audio_file2, quality_audio_file2, german_audio_file3)
if os.path.isfile(quality_audio_file1):
stick_together_audio(quality_audio_file1, german_audio_file, german_audio_file)
@@ -337,16 +384,17 @@ def process_one(German_Video: Path, Quality_Video: Path, output_dir: Path):
print(f"Final file with trimmed audio created at: {final_file}")
# print(f"Removing tmp Audiofiles")
# if os.path.isfile(german_audio_file):
# os.remove(german_audio_file)
# if os.path.isfile(quality_audio_file1):
# os.remove(quality_audio_file1)
# if os.path.isfile(quality_audio_file2):
# os.remove(quality_audio_file2)
print(f"Removing tmp Audiofiles")
if os.path.isfile(german_audio_file):
os.remove(german_audio_file)
if os.path.isfile(quality_audio_file1):
os.remove(quality_audio_file1)
if os.path.isfile(quality_audio_file2):
os.remove(quality_audio_file2)
def main():
output_dir = "/home/honney/mount/Media/tmp/serien/ready/Black Bullet/Season 01"
# for i in range(1, 14):
for i in range(1, 14):
# --- German_Videouts ---
# German_Video = f"/home/honney/mount/Media/tmp/serien/ger_low_quality/Black Bullet/Black Bullet Season 01 E{i:02}*.mp4"
@@ -362,6 +410,7 @@ def main():
raise ValueError("There are to many videos matching the Quality wildcard")
process_one(Path(German_Video[0]), Path(Quality_Video[0]), Path(output_dir))
# sys.exit()
if __name__ == "__main__":
if len(sys.argv) == 3:
@@ -378,6 +427,3 @@ if __name__ == "__main__":
# print(len(sys.argv), sys.argv)
# sys.exit()
main()