From 3ba9c2a003fd2dd7b119d99f0d9794f889210752 Mon Sep 17 00:00:00 2001 From: Hannes Date: Mon, 16 Mar 2026 16:44:21 +0100 Subject: [PATCH] renamed opencv... --- ...ross_correlation.py => cc_motion_signal.py | 65 ++-- cut_to_length.py | 307 ++++++++++++++---- 2 files changed, 287 insertions(+), 85 deletions(-) rename opencv_cross_correlation.py => cc_motion_signal.py (55%) diff --git a/opencv_cross_correlation.py b/cc_motion_signal.py similarity index 55% rename from opencv_cross_correlation.py rename to cc_motion_signal.py index 048e505..567c044 100644 --- a/opencv_cross_correlation.py +++ b/cc_motion_signal.py @@ -4,53 +4,65 @@ from scipy import signal import matplotlib.pyplot as plt import sys +TARGET_FPS = 30 # common reference rate + def get_video_signal(path): cap = cv2.VideoCapture(path) + fps = cap.get(cv2.CAP_PROP_FPS) + 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) + + return np.array(motion_signal), fps + + +def resample_signal(sig, original_fps, target_fps): + duration = len(sig) / original_fps + target_length = int(duration * target_fps) + + return signal.resample(sig, target_length) + 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") - # Normalize signals (subtract mean, divide by std dev) to improve correlation + sig_a, fps_a = get_video_signal(video_a) + sig_b, fps_b = get_video_signal(video_b) + + # 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 multiple(): # Usage @@ -64,8 +76,11 @@ def multiple(): sys.exit() def single(): - video_a = f"/home/honney/mount/Media/torent/[Anime Time] One Piece (0001-1071+Movies+Specials) [BD+CR] [Dual Audio] [1080p][HEVC 10bit x265][AAC][Multi Sub]/Season 01 - East Blue/[Anime Time] One Piece - 0001 - I'm Luffy! The Man Who's Gonna Be King of the Pirates!.mkv" - video_b = f"/home/honney/mount/Media/tmp/serien/complicated/One Piece (1999-2025) [imdbid-tt0388629]/Season 01/One Piece S01E001.mkv" + video_a = f"/home/honney/mount/Media/tmp/serien/ger_low_quality/Black Bullet/Black Bullet Season 01 E02 Die Maske des Wahnsinns.mp4" + video_b = f"/home/honney/mount/Media/tmp/serien/high_quality/Black Bullet (Season 1)/Season 01/Black Bullet - S01E02.mkv" - offset_frames = find_offset(video_a, video_b) - print(f"Video B is offset by {offset_frames} frames relative to Video A.") \ No newline at end of file + frames, seconds = find_offset(video_a, video_b) + print(f"Video B is offset by {frames} frames or {seconds} seconds relative to Video A.") + +if __name__ == '__main__': + single() \ No newline at end of file diff --git a/cut_to_length.py b/cut_to_length.py index e190188..d8ee83d 100644 --- a/cut_to_length.py +++ b/cut_to_length.py @@ -6,12 +6,61 @@ import cv2 import numpy as np from scipy import signal import sys +import os +import glob +import tempfile +import shutil + +DEBUG = True def run(cmd): - """Run a subprocess command and raise on error.""" - subprocess.run(cmd, check=True) + """ + Run a subprocess command. + + Args: + cmd: Command and arguments as a list. + + Raises: + subprocess.CalledProcessError: If the command exits with a non-zero status. + """ + try: + result = subprocess.run( + cmd, + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True + ) + return result + except subprocess.CalledProcessError as e: + 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 @@ -38,7 +87,37 @@ def get_video_signal(path): cap.release() return np.array(motion_signal) -def find_offset(video_a, video_b): +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") @@ -59,7 +138,7 @@ def find_offset(video_a, video_b): return best_lag -def probe_video(path): +def probe_video(path: Path): """Get video FPS, duration, codec, and stream start time.""" import json cmd = [ @@ -83,8 +162,23 @@ def probe_video(path): raise RuntimeError("No video stream found") -def trim_audio(German_Videout_mkv, start_time, end_time, out_audio): - """Trim only the first (and only) audio track from German_Videout_mkv.""" +def trim_audio(German_Videout_mkv: Path, start_time: float, end_time: float, out_audio: Path): + """ + Extract a segment of the first audio track from a video file. + + Uses FFmpeg to copy the selected audio range without re-encoding, + which makes the operation fast and preserves the original audio quality. + + Args: + German_Videout_mkv: Path to the source video file. + start_time: Start time of the audio segment (seconds or FFmpeg time format). + end_time: End time of the audio segment. + out_audio: Path to the output audio file. + + Notes: + The function selects the first audio stream (`0:a:0`) and copies it + directly (`-c:a copy`), meaning the codec and quality remain unchanged. + """ run([ "ffmpeg", "-y", @@ -101,7 +195,8 @@ def mux_audio_into_output(Quality_Video, audio_file, output_dir): output_dir = Path(output_dir) output_dir.mkdir(exist_ok=True) - final_output = output_dir / f"{Path(Quality_Video).stem}_ger.mkv" + # final_output = output_dir / f"{Path(Quality_Video).stem}_ger.mkv" + final_output = output_dir / f"{Path(Quality_Video).stem}.mkv" run([ "mkvmerge", @@ -114,83 +209,175 @@ def mux_audio_into_output(Quality_Video, audio_file, output_dir): return final_output -def only_one(German_Video, Quality_Video): - n1 = find_offset(German_Video, Quality_Video) # first frame to keep - print(n1) - sys.exit() - output_dir = "./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. + """ + 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}") + + # 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) + ]) + + # 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() + +def process_one(German_Video: Path, Quality_Video: Path, output_dir: Path): # --- Probe video --- fps, duration, _, vstart = probe_video(German_Video) - _, duration2, _, _ = probe_video(Quality_Video) + fps2, duration2, _, vstart2 = probe_video(Quality_Video) - # --- Compute timestamps --- - start = vstart + (n1 / fps) - end = start + duration2 - - if end <= start: - raise ValueError("n2 must be greater than n1") + if fps != fps2: + raise ValueError("FPS of both Videos must be the same") + 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 + print(f"Offset: {n1}") - - # if duration < end: - # raise ValueError("German_Video must be longer than what it needs to be cut to") - - print(f"fps: {fps}") - print(f"Cutting audio from frame {n1} (timestamp {start:.6f}s) to timestamp {end:.6f}s") + # print(f"fps: {fps}") + # print(f"Cutting audio from frame {n1} (timestamp {start:.6f}s) to timestamp {end:.6f}s") # --- Trim only audio --- - audio_file = Path(output_dir) / f"{Path(Quality_Video).stem}_german_audio.aac" - trim_audio(German_Video, start, end, audio_file) + 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: + # German starts Later than Quality + # Get the start from the original Audio + start2 = vstart2 + end2 = start2 + (n1 / fps) + 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: + # When the german audio is shorter than the original video + start = vstart + 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 + end2 = start2 + (duration2 - duration - (n1 / fps)) + 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) + + if os.path.isfile(quality_audio_file1): + stick_together_audio(quality_audio_file1, german_audio_file, german_audio_file) + if os.path.isfile(quality_audio_file2): + stick_together_audio(german_audio_file, quality_audio_file2, german_audio_file) # --- Mux trimmed audio --- - final_file = mux_audio_into_output(Quality_Video, audio_file, output_dir) + final_file = mux_audio_into_output(Quality_Video, german_audio_file, output_dir) 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) + def main(): - for i in range(1, 12): + output_dir = "/home/honney/mount/Media/tmp/serien/ready/Black Bullet/Season 01" + for i in range(1, 14): # --- German_Videouts --- - German_Video = f"/home/honney/projects/mkv/AnoHana The Flower We Saw That Day (2011-2011) [imdbid-tt1913273]/Season 01/AnoHana The Flower We Saw That Day S01E0{i:02}.mkv" - Quality_Video = f"/home/honney/projects/mkv/Anohana - The Flower We Saw That Day/S01/AnoHana The Flower We Saw That Day S01E0{i:02}.mkv" - n1 = find_offset(German_Video, Quality_Video) # first frame to keep - print(n1) - sys.exit() - output_dir = "./output_dir" + # German_Video = f"/home/honney/mount/Media/tmp/serien/ger_low_quality/Black Bullet/Black Bullet Season 01 E{i:02}*.mp4" + # Quality_Video = f"/home/honney/mount/Media/tmp/serien/high_quality/Black Bullet (Season 1)/Season 01/Black Bullet - S01E{i:02}.mkv" + pattern = f"/home/honney/mount/Media/tmp/serien/ger_low_quality/Black Bullet/Black Bullet Season 01 E{i:02}*.mp4" + German_Video = glob.glob(pattern) + pattern = f"/home/honney/mount/Media/tmp/serien/high_quality/Black Bullet (Season 1)/Season 01/Black Bullet - S01E{i:02}.mkv" + Quality_Video = glob.glob(pattern) - # --- Probe video --- - fps, duration, _, vstart = probe_video(German_Video) - _, duration2, _, _ = probe_video(Quality_Video) - - # --- Compute timestamps --- - start = vstart + (n1 / fps) - end = start + duration2 - - if end <= start: - raise ValueError("n2 must be greater than n1") + if len(German_Video) > 1: + raise ValueError("There are to many videos matching the German wildcard") + if len(Quality_Video) > 1: + raise ValueError("There are to many videos matching the Quality wildcard") - if duration < end: - raise ValueError("German_Video must be longer than what it needs to be cut to") - - print(f"fps: {fps}") - print(f"Cutting audio from frame {n1} (timestamp {start:.6f}s) to timestamp {end:.6f}s") - - # --- Trim only audio --- - audio_file = Path(output_dir) / f"{Path(Quality_Video).stem}_german_audio.aac" - trim_audio(German_Video, start, end, audio_file) - - # --- Mux trimmed audio --- - final_file = mux_audio_into_output(Quality_Video, audio_file, output_dir) - - print(f"Final file with trimmed audio created at: {final_file}") + process_one(Path(German_Video[0]), Path(Quality_Video[0]), Path(output_dir)) if __name__ == "__main__": if len(sys.argv) == 3: print(sys.argv) print(f"German_Video={sys.argv[1]}") print(f"Quality_Video={sys.argv[2]}") - only_one(German_Video=sys.argv[1], Quality_Video=sys.argv[2]) + process_one(German_Video=Path(sys.argv[1]), Quality_Video=Path(sys.argv[2]), output_dir=Path("./output_dir")) + elif len(sys.argv) == 4: + print(sys.argv) + print(f"German_Video={sys.argv[1]}") + print(f"Quality_Video={sys.argv[2]}") + process_one(German_Video=Path(sys.argv[1]), Quality_Video=Path(sys.argv[2]), output_dir=Path(sys.argv[3])) else: # print(len(sys.argv), sys.argv) # sys.exit() main() + + +