V 1.0 - Only Prints

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

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("./examples/high_quality_episode.mkv")
+    video_path_2 = Path("./examples/dubbed_episode.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}")

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 = "./examples/high_quality_episode.mkv"
+        file2 = "./examples/dubbed_episode.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:")

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")