From 5ac5b6615b9d665cc17137356e05d6809eed9404 Mon Sep 17 00:00:00 2001 From: Hannes Date: Fri, 20 Mar 2026 00:46:36 +0100 Subject: [PATCH] V 1.0 - Only Prints --- audio/__init__.py | 16 ++ audio/audio_cross_correlation.py | 54 ++++ audio/audio_utils.py | 333 ++++++++++++++++++++++++ main.py | 207 +++++++++++++++ mkv/__init__.py | 9 + mkv/chapter.py | 67 +++++ mkv/check_intro_outro.py | 18 ++ mkv/intro_outro.py | 55 ++++ readme.md | 1 + requirements.txt | 5 + utils.py | 25 ++ video/__init__.py | 5 + video/scene_change_cross_correlation.py | 117 +++++++++ video/temporal_cross_correlation.py | 93 +++++++ video/video_utils.py | 40 +++ 15 files changed, 1045 insertions(+) create mode 100644 audio/__init__.py create mode 100644 audio/audio_cross_correlation.py create mode 100644 audio/audio_utils.py create mode 100644 main.py create mode 100644 mkv/__init__.py create mode 100644 mkv/chapter.py create mode 100644 mkv/check_intro_outro.py create mode 100644 mkv/intro_outro.py create mode 100644 readme.md create mode 100644 requirements.txt create mode 100644 utils.py create mode 100644 video/__init__.py create mode 100644 video/scene_change_cross_correlation.py create mode 100644 video/temporal_cross_correlation.py create mode 100644 video/video_utils.py diff --git a/audio/__init__.py b/audio/__init__.py new file mode 100644 index 0000000..5b4dbef --- /dev/null +++ b/audio/__init__.py @@ -0,0 +1,16 @@ +# audio/__init__.py + +from .audio_utils import ( + extract_audio_segment_file, + extract_audio_segment_tmp, + get_audio_duration, + stick_together_audio, + get_audio_streams, + get_track_info, + shift_audio, + replace_with_shifted_mka +) + +from .audio_cross_correlation import ( + get_audio_offset +) \ No newline at end of file diff --git a/audio/audio_cross_correlation.py b/audio/audio_cross_correlation.py new file mode 100644 index 0000000..8f31055 --- /dev/null +++ b/audio/audio_cross_correlation.py @@ -0,0 +1,54 @@ +# for import needed +import sys +from pathlib import Path + +sys.path.append(str(Path(__file__).resolve().parent.parent)) + +from utils import run +### Normal Imports: +import tempfile +import subprocess +import soundfile as sf +import numpy as np +from scipy.signal import correlate + +def extract_audio(file: Path, duration: float) -> tuple[np.ndarray, int]: + tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False) + + subprocess.run([ + "ffmpeg", + "-y", + "-i", str(file), + "-map", "0:a:0", # first audio stream + "-ac", "1", # mono + "-ar", "44100", # fixed sample rate + "-t", str(duration), # limit duration + tmp.name + ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + + data, sr = sf.read(tmp.name) + if not isinstance(data, np.ndarray): + raise ChildProcessError(f"Could not get signal from {file}") + if not isinstance(sr, int): + try: + sr = int(sr) + except: + ValueError(f"sr could not be read from {file}") + return data, sr + +def get_audio_offset(file1: Path, file2: Path, important_time: float) -> float: + sig1, sr1 = extract_audio(file1, important_time) + sig2, sr2 = extract_audio(file2, important_time) + + if sr1 != sr2: + raise ValueError("Sample rates do not match") + + # normalize signals + sig1 = sig1 - np.mean(sig1) + sig2 = sig2 - np.mean(sig2) + + corr = correlate(sig1, sig2, mode="full") + lag = np.argmax(corr) - (len(sig2) - 1) + + offset_seconds = float(lag / sr1) + return offset_seconds \ No newline at end of file diff --git a/audio/audio_utils.py b/audio/audio_utils.py new file mode 100644 index 0000000..3d2c658 --- /dev/null +++ b/audio/audio_utils.py @@ -0,0 +1,333 @@ +# for import needed +import sys +from pathlib import Path + +sys.path.append(str(Path(__file__).resolve().parent.parent)) + +from utils import run +### Normal Imports: +import subprocess +import shutil +import tempfile +from typing import Tuple, Optional, overload, Union, List +import json + +def extract_audio_segment_file(video: 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", + "-i", video, + "-ss", str(start_time), + "-to", str(end_time), + "-map", "0:a:0", # first audio track + "-c:a", "copy", # no re-encode + out_audio + ]) + +@overload +def extract_audio_segment_tmp( + file: Path, + start_or_section: float, + end: float, + audio_stream: Optional[int] = None +) -> Path: ... + +# Overload 2: Using a tuple for the section +@overload +def extract_audio_segment_tmp( + file: Path, + start_or_section: tuple[float, float], + end: None = None, + audio_stream: Optional[int] = None +) -> Path: ... + +# Implementation +def extract_audio_segment_tmp( + file: Path, + start_or_section: float | tuple[float, float], + end: Optional[float] = None, + audio_stream: Optional[int] = None +) -> Path: + """ + Extract audio using (start, duration) from get_intro. + + Parameters: + file: input video file + intro: (start, duration) + audio_stream: index of audio stream (0 = first audio track, etc.) + + Returns: + Path to temporary audio file + """ + # 1. Handle Tuple Input: (start, end) + if isinstance(start_or_section, tuple): + if len(start_or_section) != 2: + raise ValueError("Tuple must strictly be (start, end)") + start = float(start_or_section[0]) + duration = float(start_or_section[1]) + + # 2. Handle Float Input: start, end + elif isinstance(start_or_section, (int, float)): + if end is None: + raise ValueError("The 'end' parameter must be provided when using a separate start float.") + start = float(start_or_section) + end_time = float(end) + duration = end_time - start + + else: + raise TypeError("Invalid type for start_or_section. Expected float or tuple.") + + if duration <= 0: + raise ValueError("Duration must be above 0") + + tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False) + tmp_path = Path(tmp.name) + tmp.close() + + cmd = [ + "ffmpeg", + "-y", + "-i", str(file), + "-ss", str(start), + "-t", str(duration), + ] + + # Select specific audio stream if provided + if audio_stream is not None: + cmd += ["-map", f"0:a:{audio_stream}"] + else: + cmd += ["-map", "0:a:0"] # default: first audio track + + cmd += [ + "-vn", + "-acodec", "pcm_s16le", + "-ar", "44100", + "-ac", "2", + str(tmp_path) + ] + + run(cmd) + return tmp_path + +def get_audio_duration(file_path: Path): + """ + Get the actual duration of the first audio stream by asking ffprobe + to count packets/frames if the stream metadata is missing. + Returns duration in seconds. + """ + + # As a last resort, use the format duration + result = subprocess.run([ + "ffprobe", + "-v", "error", + "-select_streams", "a:0", + "-show_entries", "stream_tags=DURATION", + "-of", "default=noprint_wrappers=1:nokey=1", + file_path + ], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True + ) + result = result.stdout.strip().split(":")[::-1] + # if DEBUG: print(f"audio duration: {result}") + res = 0 + for i in range(len(result)): + res += float(result[i])*(60**i) + return res + +def stick_together_audio(left_audio: Path, right_audio: Path, output_audio: Path) -> None: + """ + 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) + + # Use stem instead of prefix to create a temporary file + output_audio_tmp = output_audio.parent / (output_audio.stem + "_temp" + output_audio.suffix) + + run([ + "ffmpeg", + "-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) + ]) + + shutil.move(output_audio_tmp, output_audio) + +def get_audio_streams(file: Path) -> List[int]: + """ + Return a list of available audio stream indices (0-based for ffmpeg mapping). + + Example return: + [0, 1] -> corresponds to 0:a:0 and 0:a:1 + """ + cmd = [ + "ffprobe", + "-i", str(file), + "-select_streams", "a", + "-show_entries", "stream=index", + "-of", "json", + "-loglevel", "error" + ] + + result = run(cmd) + data = json.loads(result.stdout) + + # ffprobe gives absolute stream indices, but we want relative audio indices + streams = data.get("streams", []) + + # Convert to 0-based audio stream numbering + return list(range(len(streams))) + +def get_track_info(file: Path, audioline: int): + """Gets codec, bitrate, language, and title.""" + cmd = [ + "ffprobe", "-v", "error", "-select_streams", f"a:{audioline}", + "-show_entries", "stream=codec_name,bit_rate:stream_tags=language,title", + "-of", "json", str(file) + ] + result = subprocess.run(cmd, capture_output=True, text=True) + data = json.loads(result.stdout) + if "streams" in data and len(data["streams"]) > 0: + s = data["streams"][0] + return { + "codec": s.get("codec_name", "aac"), + "bitrate": s.get("bit_rate", "192k"), + "lang": s.get("tags", {}).get("language", "eng"), + "title": s.get("tags", {}).get("title", f"Track {audioline}") + } + return {"codec": "aac", "bitrate": "192k", "lang": "eng", "title": "Audio"} + +def shift_audio(file: Path, audioline: int, offset: float): + if not file.is_file(): return + + info = get_track_info(file, audioline) + # mkvmerge uses 0 for video, usually 1+ for audio. + # We need the MKV Toolnix Track ID. + mkv_id = audioline + 1 + + temp_mka = file.parent / "temp_shifted.mka" + output_mkv = file.parent / "temp_output.mkv" + + # --- STEP 1: Physical Shift via Filter --- + # We convert offset to milliseconds for the 'adelay' filter + ms = int(abs(offset) * 1000) + + if offset > 0: + # RIGHT SHIFT: Delay start, then trim the end to keep original length + # 'all' applies delay to all channels + filter_str = f"adelay={ms}|{ms}|{ms}|{ms}|{ms}|{ms},atrim=0,atrim=duration=base" # Simplified logic below + # For reliability, we use a simpler pad/trim approach: + filter_chain = f"adelay={ms}:all=1" + else: + # LEFT SHIFT: Skip the beginning + filter_chain = f"atrim=start={abs(offset)}" + + # Process to MKA + # We use the same codec and bitrate detected earlier + process_cmd = [ + "ffmpeg", "-y", "-i", str(file), + "-map", f"0:a:{audioline}", + "-af", filter_chain, + "-c:a", info['codec'], + "-b:a", str(info['bitrate']), + str(temp_mka) + ] + subprocess.run(process_cmd, check=True) + + # --- STEP 2: Replace in MKV via mkvmerge --- + # --audio-tracks !ID excludes the old track + # The new MKA is added at the end + merge_cmd = [ + "mkvmerge", "-o", str(output_mkv), + "--audio-tracks", f"!{mkv_id}", str(file), + "--language", f"0:{info['lang']}", + "--track-name", f"0:{info['title']}", + str(temp_mka) + ] + + subprocess.run(merge_cmd, check=True) + + # Cleanup and Replace + shutil.move(str(output_mkv), str(file)) + temp_mka.unlink() + print(f"Done: {file.name} updated physically.") + +def replace_with_shifted_mka(mkv_file: Path, mka_source: Path, audioline: int, offset: float): + """ + Physically shifts an existing MKA and replaces a track in the MKV. + - Codec: Re-encoded to Opus (to ensure physical shift compatibility) + - Metadata: Hardcoded to German / German (Opus Stereo) + """ + if not mkv_file.is_file() or not mka_source.is_file(): + print("Error: Input files not found.") + return + + # mkvmerge IDs: 0 is usually video, so audioline 0 is track 1, etc. + mkv_track_id = audioline + 1 + + with tempfile.TemporaryDirectory() as temp_dir: + temp_path = Path(temp_dir) + shifted_mka = temp_path / "shifted.mka" + output_mkv = temp_path / "final_output.mkv" + + # --- STEP 1: Physical Shift --- + # Positive offset: adelay (adds real silence packets) + # Negative offset: atrim (removes packets) + asetpts (resets timestamps) + ms = int(abs(offset) * 1000) + if offset > 0: + filter_chain = f"adelay={ms}:all=1" + else: + filter_chain = f"atrim=start={abs(offset)},asetpts=PTS-STARTPTS" + + print(f"Physically shifting {mka_source.name}...") + process_cmd = [ + "ffmpeg", "-y", "-i", str(mka_source), + "-af", filter_chain, + "-c:a", "libopus", # Set to Opus as per your track name + "-b:a", "160k", # Standard high-quality bitrate for stereo Opus + str(shifted_mka) + ] + subprocess.run(process_cmd, check=True, capture_output=True) + + # --- STEP 2: Replace track in MKV --- + # --audio-tracks !ID excludes the old line from the original MKV + print(f"Replacing track {audioline} in {mkv_file.name}...") + merge_cmd = [ + "mkvmerge", "-o", str(output_mkv), + "--audio-tracks", f"!{mkv_track_id}", str(mkv_file), + "--language", "0:ger", + "--track-name", "0:German (Opus Stereo)", + str(shifted_mka) + ] + + subprocess.run(merge_cmd, check=True, capture_output=True) + + # --- STEP 3: Cleanup and Overwrite --- + shutil.move(str(output_mkv), str(mkv_file)) + print(f"Success: Updated {mkv_file.name}") \ No newline at end of file diff --git a/main.py b/main.py new file mode 100644 index 0000000..b688457 --- /dev/null +++ b/main.py @@ -0,0 +1,207 @@ +import sys +from pathlib import Path +from typing import Optional +import os +import glob +import numpy as np +import subprocess +from utils import run +from audio import extract_audio_segment_file, get_audio_duration, stick_together_audio, replace_with_shifted_mka +from video import probe_video, find_temporal_offset, corelate_videos +from mkv import check_intro_outro_sync +from tqdm import tqdm + +DEBUG = False + +def mux_audio_into_output(Quality_Video, audio_file, output_dir): + """Mux trimmed audio into Quality_Video and place final file in 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}.mkv" + + run([ + "mkvmerge", + "-o", str(final_output), + str(Quality_Video), + "--language", "0:deu", # assign German language + "--track-name", "0:German (Opus Stereo)", + str(audio_file) + ]) + + return final_output + +def process_one(German_Video: Path, Quality_Video: Path, output_dir: Path, offset: Optional[int] = None) -> bool: + + print("probe_video") + # --- Probe video --- + fps, duration, _, vstart = probe_video(German_Video) + fps2, duration2, _, vstart2 = probe_video(Quality_Video) + + 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") + + print("Compute offset") + # --- Offset --- + if offset == None: + offset0 = corelate_videos(Quality_Video, German_Video, fps2) + else: + offset0 = offset + if offset0 > duration or offset0 > duration2: + raise ValueError("Offset needs to be smaller than the video durations") + print(f"Offset: {offset0}s") + + print("Extract pre beginning") + 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" + + # Clear old faulty files, ger is gearantied to be replaced + if os.path.isfile(quality_audio_file1): + os.remove(quality_audio_file1) + if os.path.isfile(quality_audio_file2): + os.remove(quality_audio_file2) + + ### Get the first part till german starts (if german does not start before file) + start2 = vstart2 + end2 = start2 + offset0 + if end2 - start2 > 0.05: + if DEBUG: print(f"Cutting audio ({quality_audio_file1}) (from timestamp {start2:.6f}s) to timestamp {end2:.6f}s") + try: + extract_audio_segment_file(Quality_Video, start2, end2, quality_audio_file1) + except subprocess.CalledProcessError as e: + raise BrokenPipeError( + f"{e}\n" + f"Trying to extract Audio from {Quality_Video} {start2}-{end2} " + f"to {quality_audio_file2} to have a duration of {end2-start2}, " + f"while the audio has a length of {get_audio_duration(Quality_Video)}" + ) + + print("Extract german audio") + + ### Get the German Audio + if offset0 < 0: + start = vstart - offset0 + else: + start = vstart + end = min(vstart + duration, start + duration2) + if DEBUG: print(f"Cutting audio ({german_audio_file}) (from timestamp {start:.6f}s) to timestamp {end:.6f}s") + try: + extract_audio_segment_file(German_Video, start, end, german_audio_file) + except subprocess.CalledProcessError as e: + raise BrokenPipeError( + f"{e}\n" + f"Trying to extract Audio from {German_Video} {start}-{end} " + f"to {german_audio_file} to have a duration of {end-start}, " + f"while the audio has a length of {get_audio_duration(German_Video)}" + ) + + print("Extract post end") + + ### Get the end from the first original audio if it is longer than the german part + start2 = vstart2 + end + # end2 = start2 + (duration2 - duration - (n1 / fps)) + end2 = vstart2 + get_audio_duration(Quality_Video) + if end2 - start2 > 0.5: + if DEBUG: print(f"Cutting audio ({quality_audio_file2}) (from timestamp {start2:.6f}s) to timestamp {end2:.6f}s") + try: + extract_audio_segment_file(Quality_Video, start2, end2, quality_audio_file2) + except subprocess.CalledProcessError as e: + raise BrokenPipeError( + f"{e}\n" + f"Trying to extract Audio from {Quality_Video} {start2}-{end2} " + f"to {quality_audio_file2} to have a duration of {end2-start2}, " + f"while the audio has a length of {get_audio_duration(Quality_Video)}" + ) + + print("Combine audio") + + # Combine the audio files + 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) + + print("Mux") + + # --- Mux trimmed audio --- + final_file = mux_audio_into_output(Quality_Video, german_audio_file, output_dir) + + print("Cleanup") + + 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("Final check") + try: + files_acceptable, op_ed_offset, stream_num = check_intro_outro_sync(final_file) + except: + new_name = final_file.with_name(f"[NO CHAPTER]_{final_file.name}") + final_file.rename(new_name) + if os.path.isfile(german_audio_file): + os.remove(path=german_audio_file) + return False + + if files_acceptable and isinstance(op_ed_offset, float) and isinstance(stream_num, int): + print(f"✔ {final_file.name} | offset={op_ed_offset:.4f}") + if np.abs(op_ed_offset) > (1/fps2) and os.path.isfile(german_audio_file): + replace_with_shifted_mka(final_file, german_audio_file, stream_num, op_ed_offset) + if os.path.isfile(german_audio_file): + os.remove(path=german_audio_file) + return True + else: + print(f"✖ FAILED: {final_file.name}") + if os.path.isfile(german_audio_file): + os.remove(path=german_audio_file) + new_name = final_file.with_name(f"[FAILED]_{final_file.name}") + final_file.rename(new_name) + return False + + +def main(): + output_dir = "./output_dir" + start_num = 1 + episode_num = 11 + fault_catch = [] + + episodes = range(start_num, episode_num + 1) + + for i in episodes: + german_pattern = f"Folder0/S01E{i:02}*.mkv" + if os.path.isfile(german_pattern): + German_Video = [german_pattern] + else: + German_Video = glob.glob(german_pattern) + quality_pattern = f"Folder1/S01E{i:02}*.mkv" + if os.path.isfile(quality_pattern): + Quality_Video = [quality_pattern] + else: + Quality_Video = glob.glob(quality_pattern) + # print(f"German_Video={German_Video}") + # print(f"Quality_Video={Quality_Video}") + + if len(German_Video) != 1: + raise ValueError(f"Error with German wildcard:\n{german_pattern}\n{German_Video}") + if len(Quality_Video) != 1: + raise ValueError(f"Error with Quality wildcard:\n{quality_pattern}\n{Quality_Video}") + if not process_one(Path(German_Video[0]), Path(Quality_Video[0]), Path(output_dir)): + fault_catch.append([Path(German_Video[0]), Path(Quality_Video[0])]) + if fault_catch: + print("\nFailed Videos:") + for ger, orig in fault_catch: + print(f"{ger} + {orig}") + +if __name__ == '__main__': + if len(sys.argv) == 3: + if not process_one(German_Video=Path(sys.argv[1]), Quality_Video=Path(sys.argv[2]), output_dir=Path("../output_dir")): + print(f"Video could not be correctly combined") + elif len(sys.argv) == 4: + if not process_one(German_Video=Path(sys.argv[1]), Quality_Video=Path(sys.argv[2]), output_dir=Path(sys.argv[3])): + print(f"Video could not be correctly combined") + else: + main() \ No newline at end of file diff --git a/mkv/__init__.py b/mkv/__init__.py new file mode 100644 index 0000000..65e8c98 --- /dev/null +++ b/mkv/__init__.py @@ -0,0 +1,9 @@ +# mkv/__init__.py + +from .chapter import ( + get_intro_outro, + get_chapters + ) +from .intro_outro import ( + check_intro_outro_sync +) \ No newline at end of file diff --git a/mkv/chapter.py b/mkv/chapter.py new file mode 100644 index 0000000..6909a63 --- /dev/null +++ b/mkv/chapter.py @@ -0,0 +1,67 @@ +# for import needed +import sys +from pathlib import Path + +sys.path.append(str(Path(__file__).resolve().parent.parent)) + +from utils import run +### Regular Imports +from typing import Tuple, List +import json + +def get_chapters(file: Path) -> List[Tuple[float, float]]: + """Return list of (start_ti +from .intro_outro importme, duration) for each chapter.""" + cmd = [ + "ffprobe", + "-i", str(file), + "-print_format", "json", + "-show_chapters", + "-loglevel", "error" + ] + + result = run(cmd) + data = json.loads(result.stdout) + + chapters = [] + for ch in data.get("chapters", []): + start = float(ch["start_time"]) + end = float(ch["end_time"]) + duration = end - start + chapters.append((start, duration)) + + return chapters + +def get_intro_outro(file: Path, length: float = 90.0) -> List[Tuple[float, float]]: + try: + chapters = get_chapters(Path(file)) + except: + raise AttributeError(f"{file} has no chapter Marks") + multiplier = 2 + min_len = length / multiplier + max_len = length * multiplier + filtered = [ + (start, dur) for start, dur in chapters + if min_len <= dur <= max_len + ] + while len(filtered) > 2 and multiplier <= 1.1: + multiplier -= 0.1 + min_len = length / multiplier + max_len = length * multiplier + filtered = [ + (start, dur) for start, dur in chapters + if min_len <= dur <= max_len + ] + if len(filtered) == 2: + return filtered + elif len(filtered) == 1: + ### Found only a Intro or Outro + raise ValueError(f"Could only find one intro or outro:\n{filtered}") + elif len(filtered) == 0: + raise ValueError(f"Found no outro or intro:\n{filtered}") + else: + raise ValueError(f"Found too many intros/outros:\n{filtered}") + +if __name__ == '__main__': + file = "/home/honney/projects/mkv/output_dir/08.mkv" + print(get_intro_outro(Path(file))) \ No newline at end of file diff --git a/mkv/check_intro_outro.py b/mkv/check_intro_outro.py new file mode 100644 index 0000000..8b19f71 --- /dev/null +++ b/mkv/check_intro_outro.py @@ -0,0 +1,18 @@ +import sys +import os +from pathlib import Path +from mkv import check_intro_outro_sync + +if __name__ == '__main__': + if len(sys.argv) == 2: + if os.path.isfile(sys.argv[1]): + boolean, floater, stream = check_intro_outro_sync(Path(sys.argv[1])) + if boolean: + print("Video is syncable!") + print(f"With offset of {floater} at audio_stream {stream}") + else: + print("Video is not syncable!") + else: + raise FileExistsError(f"File does not exists: {sys.argv[1]}") + else: + print("python check_intro.py /path/to/file") diff --git a/mkv/intro_outro.py b/mkv/intro_outro.py new file mode 100644 index 0000000..12a6cd8 --- /dev/null +++ b/mkv/intro_outro.py @@ -0,0 +1,55 @@ +from mkv.chapter import get_intro_outro +from audio.audio_utils import extract_audio_segment_tmp, get_audio_streams +from audio.audio_cross_correlation import get_audio_offset +from pathlib import Path +import numpy as np +from typing import Tuple + +def check_intro_outro_sync(file: Path, length: float = 90.0, uncertainty: float = 0.1, max_offset: float = 0.1) -> Tuple[bool, float|None, int|None]: + intro_outro = get_intro_outro(file, length) + if intro_outro[0][0] subprocess.CompletedProcess[str]: + """ + 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 \ No newline at end of file diff --git a/video/__init__.py b/video/__init__.py new file mode 100644 index 0000000..2a3e1fc --- /dev/null +++ b/video/__init__.py @@ -0,0 +1,5 @@ +# video/__init__.py + +from .video_utils import probe_video +from .temporal_cross_correlation import find_temporal_offset +from .scene_change_cross_correlation import corelate_videos \ No newline at end of file diff --git a/video/scene_change_cross_correlation.py b/video/scene_change_cross_correlation.py new file mode 100644 index 0000000..d423bd7 --- /dev/null +++ b/video/scene_change_cross_correlation.py @@ -0,0 +1,117 @@ +from .video_utils import probe_video + +import cv2 +import numpy as np +from scipy import signal as scipy_signal +from pathlib import Path + +def get_scene_change_signal(vid: Path) -> tuple[np.ndarray, float]: + """ + Reads a video, lowers resolution, and generates a signal array representing + scene changes. Returns the signal and the original FPS. + """ + cap = cv2.VideoCapture(str(vid)) + if not cap.isOpened(): + raise ValueError(f"Could not open video: {vid}") + + fps = cap.get(cv2.CAP_PROP_FPS) + signal_list = [] + + ret, prev_frame = cap.read() + if not ret: + return np.array([]), fps + + # Resize to 160x90 and convert to grayscale for faster, easier comparison + prev_gray = cv2.cvtColor(cv2.resize(prev_frame, (160, 90)), cv2.COLOR_BGR2GRAY) + + while True: + ret, curr_frame = cap.read() + if not ret: + break + + curr_gray = cv2.cvtColor(cv2.resize(curr_frame, (160, 90)), cv2.COLOR_BGR2GRAY) + + # Calculate absolute difference between the current frame and the previous one + diff = cv2.absdiff(prev_gray, curr_gray) + + # Square the differences. This ensures 80-90% similar frames output a very low number, + # but completely different frames (scene changes) output massive spikes. + score = np.mean(np.square(diff.astype(np.float32))) + signal_list.append(score) + + prev_gray = curr_gray + + cap.release() + + # The first frame has no previous frame to compare to, so we insert a 0 at the start + signal_array = np.insert(np.array(signal_list), 0, 0.0) + + return signal_array, fps + +def resample_signal(signal: np.ndarray, fps: float, normalized_fps: float) -> np.ndarray: + """ + Stretches or contracts the signal to match the target normalized_fps. + """ + if len(signal) == 0: + return signal + + # Calculate duration of the original signal in seconds + duration = len(signal) / fps + + # Calculate how many samples we should have at the new FPS + num_samples = int(duration * normalized_fps) + + # Use scipy's Fourier method to resample the signal + resampled = scipy_signal.resample(signal, num_samples) + + # Remove negative artifacts introduced by fourier resampling + return np.clip(resampled, 0, None) # type: ignore + +def corelate_videos(vid1: Path, vid2: Path, fps: float) -> float: + """ + Correlates two videos by finding the time offset between their scene changes. + Returns the offset in seconds. A positive offset means vid1 starts AFTER vid2. + """ + # 1. Get signals and original framerates + signal1, fps1 = get_scene_change_signal(vid1) + signal2, fps2 = get_scene_change_signal(vid2) + + # 2. Resample both signals to the target FPS + resampled_signal1 = resample_signal(signal1, fps1, fps) + resampled_signal2 = resample_signal(signal2, fps2, fps) + + # 3. Normalize signals (mean=0, std=1) to prevent overall brightness/contrast differences + # from throwing off the mathematical correlation + norm_sig1 = (resampled_signal1 - np.mean(resampled_signal1)) / (np.std(resampled_signal1) + 1e-8) + norm_sig2 = (resampled_signal2 - np.mean(resampled_signal2)) / (np.std(resampled_signal2) + 1e-8) + + # 4. Cross-correlate the signals + correlation = scipy_signal.correlate(norm_sig1, norm_sig2, mode='full') + lags = scipy_signal.correlation_lags(len(norm_sig1), len(norm_sig2), mode='full') + + # 5. Find the lag with the highest correlation peak + best_lag_idx = np.argmax(correlation) + best_lag_samples = lags[best_lag_idx] + + # 6. Convert lag in samples to lag in seconds + offset_time = best_lag_samples / fps + + return offset_time + +# --- Example Usage --- +if __name__ == "__main__": + video_path_1 = Path("/home/honney/projects/mkv/Maid-sama/11_org.mkv") + video_path_2 = Path("/home/honney/projects/mkv/Maid-sama/11_ger.mkv") + + target_fps, _, _, _ = probe_video(video_path_1) + + try: + offset = corelate_videos(video_path_1, video_path_2, target_fps) + print(f"Time offset: {offset:.3f} seconds") + print(f"Frame offset: {int(offset*target_fps)} frames") + if offset > 0: + print("Video 1 starts after Video 2.") + else: + print("Video 1 starts before Video 2.") + except Exception as e: + print(f"Error: {e}") \ No newline at end of file diff --git a/video/temporal_cross_correlation.py b/video/temporal_cross_correlation.py new file mode 100644 index 0000000..d139ab0 --- /dev/null +++ b/video/temporal_cross_correlation.py @@ -0,0 +1,93 @@ +# for import needed +import sys +from pathlib import Path + +sys.path.append(str(Path(__file__).resolve().parent.parent)) + +from utils import run +### Normal Imports: +from typing import Tuple, Optional +import numpy as np +import numpy.typing as npt +import cv2 +from scipy import signal +import os + +def get_video_signal(path: Path) -> Tuple[np.ndarray, float]: + 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 + + small_frame = cv2.resize(frame, (160, 90)) + gray = cv2.cvtColor(small_frame, cv2.COLOR_BGR2GRAY) + + if prev_gray is not None: + 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), fps + + +def resample_signal(sig: np.ndarray, original_fps: float, target_fps: float) -> npt.NDArray[np.float64]: + duration = len(sig) / original_fps + target_length = int(duration * target_fps) + resampled = signal.resample(sig, target_length) + + return np.asarray(resampled, dtype=np.float64) + + +def find_temporal_offset(video_a: Path, video_b: Path, target_fps: float=24) -> Tuple[int, float]: + + 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) + + correlation = signal.correlate(sig_a, sig_b, mode='full') + lags = signal.correlation_lags(len(sig_a), len(sig_b), mode='full') + + best_lag = lags[np.argmax(correlation)] + + # Convert lag to seconds + offset_seconds = best_lag / target_fps + + return best_lag, offset_seconds + +if __name__ == '__main__': + if len(sys.argv) == 3: + if not os.path.isfile(sys.argv[1]): + raise ValueError(f"{sys.argv[1]} is not a valid file path") + if not os.path.isfile(sys.argv[2]): + raise ValueError(f"{sys.argv[2]} is not a valid file path") + offset_f, offset_t = find_temporal_offset(Path(sys.argv[1]), Path(sys.argv[2])) + print(f"Offset between Videos:\n{offset_f} Frames\n{offset_t} s") + elif len(sys.argv) == 1: + file1 = "/home/honney/projects/mkv/Maid-sama/11_org.mkv" + file2 = "/home/honney/projects/mkv/Maid-sama/11_ger.mkv" + if not os.path.isfile(file1): + raise ValueError(f"{file1} is not a valid file path") + if not os.path.isfile(file2): + raise ValueError(f"{file2} is not a valid file path") + offset_f, offset_t = find_temporal_offset(Path(file1), Path(file2)) + print(f"Offset between Videos:\n{offset_f} Frames\n{offset_t} s") + else: + print("Correct usage:") \ No newline at end of file diff --git a/video/video_utils.py b/video/video_utils.py new file mode 100644 index 0000000..07774c7 --- /dev/null +++ b/video/video_utils.py @@ -0,0 +1,40 @@ +# for import needed +import sys +from pathlib import Path + +sys.path.append(str(Path(__file__).resolve().parent.parent)) + +from utils import run +### Normal Imports: + +def probe_video(path: Path) -> tuple[float, float, str, float]: + """ + Get video FPS, duration, codec, and stream start time. + + Args: + path: To Video File + + Returns: + fps, duration, codec, start_time + """ + import json + cmd = [ + "ffprobe", + "-v", "error", + "-print_format", "json", + "-show_streams", + "-show_format", + path + ] + data = json.loads(run(cmd).stdout) + + duration = float(data["format"]["duration"]) + for s in data["streams"]: + if s["codec_type"] == "video": + num, den = s["r_frame_rate"].split("/") + fps = float(num) / float(den) + codec = s["codec_name"] + start_time = float(s.get("start_time", 0)) + return fps, duration, codec, start_time + + raise RuntimeError("No video stream found") \ No newline at end of file