Fix Get Lucky offset and improve alignment tool
Correct offset from 66.389 to 0.207 (was matching repeated section). The tool now generates stereo comparison WAV files (left=local, right=YouTube) for each candidate so you can verify by ear. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>main
Claude
parent
c673b3e746
commit
a72c52edf5
|
|
@ -15,7 +15,7 @@ export const SONGS: Song[] = [
|
|||
bgUrl: '/songs/get-lucky/bg.png',
|
||||
bannerUrl: '/songs/get-lucky/bn.png',
|
||||
youtubeVideoId: '5NV6Rdv1a3I',
|
||||
youtubeOffset: 66.389,
|
||||
youtubeOffset: 0.207,
|
||||
},
|
||||
{
|
||||
id: 'just-dance',
|
||||
|
|
|
|||
|
|
@ -2,12 +2,12 @@
|
|||
"""
|
||||
Find the time offset between a local audio file and a YouTube video's audio.
|
||||
|
||||
Uses cross-correlation to determine how many seconds into the YouTube audio
|
||||
the local audio begins. This offset is needed so the rhythm game chart
|
||||
(timed to the local audio) stays in sync when playing via YouTube.
|
||||
Uses cross-correlation to find candidate offsets, then generates stereo
|
||||
comparison WAV files (left ear = local, right ear = YouTube) so you can
|
||||
verify alignment by ear.
|
||||
|
||||
Usage:
|
||||
python find_offset.py <local_audio_path> <youtube_video_id>
|
||||
python find_offset.py <local_audio_path> <youtube_video_id> [--out-dir DIR]
|
||||
|
||||
Example:
|
||||
python find_offset.py ../../public/songs/get-lucky/audio.ogg 5NV6Rdv1a3I
|
||||
|
|
@ -17,9 +17,10 @@ Requirements:
|
|||
ffmpeg must be on PATH
|
||||
|
||||
Output:
|
||||
Prints the offset in seconds (how far into the YouTube video the local
|
||||
audio's time=0 corresponds to). A positive value means the YouTube video
|
||||
has extra content before the song starts.
|
||||
- Prints top candidate offsets ranked by correlation
|
||||
- Writes stereo WAV files for each candidate to --out-dir
|
||||
- Listen with headphones: left ear = local, right ear = YouTube
|
||||
- The file where both ears are in sync has the correct offset
|
||||
"""
|
||||
|
||||
import argparse
|
||||
|
|
@ -28,12 +29,15 @@ import os
|
|||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import wave
|
||||
|
||||
import numpy as np
|
||||
from scipy import signal
|
||||
|
||||
|
||||
SAMPLE_RATE = 16000 # downsample to 16kHz mono for fast correlation
|
||||
NUM_CANDIDATES = 5 # number of top peaks to output
|
||||
MIN_PEAK_GAP_S = 10 # minimum seconds between reported peaks
|
||||
|
||||
|
||||
def audio_to_pcm(path: str, sr: int = SAMPLE_RATE) -> np.ndarray:
|
||||
|
|
@ -68,42 +72,76 @@ def download_youtube_audio(video_id: str, out_dir: str) -> str:
|
|||
if result.returncode != 0:
|
||||
raise RuntimeError(f"yt-dlp failed: {result.stderr}")
|
||||
|
||||
# find the output file
|
||||
for f in os.listdir(out_dir):
|
||||
if f.startswith("yt_audio"):
|
||||
return os.path.join(out_dir, f)
|
||||
raise RuntimeError("yt-dlp produced no output file")
|
||||
|
||||
|
||||
def find_offset(local_pcm: np.ndarray, yt_pcm: np.ndarray, sr: int = SAMPLE_RATE) -> float:
|
||||
def find_candidates(
|
||||
local_pcm: np.ndarray,
|
||||
yt_pcm: np.ndarray,
|
||||
sr: int = SAMPLE_RATE,
|
||||
) -> list[tuple[float, float]]:
|
||||
"""
|
||||
Find the offset (in seconds) of local_pcm within yt_pcm using
|
||||
cross-correlation.
|
||||
|
||||
Returns the time in the YouTube audio where local audio's t=0 aligns.
|
||||
Find top candidate offsets using cross-correlation of the full local audio.
|
||||
Returns list of (offset_seconds, confidence) tuples.
|
||||
"""
|
||||
# Use only the first 30 seconds of local audio for speed
|
||||
max_samples = sr * 30
|
||||
local_chunk = local_pcm[:max_samples]
|
||||
|
||||
# Normalize both signals
|
||||
local_chunk = local_chunk / (np.max(np.abs(local_chunk)) + 1e-10)
|
||||
local_norm = local_pcm / (np.max(np.abs(local_pcm)) + 1e-10)
|
||||
yt_norm = yt_pcm / (np.max(np.abs(yt_pcm)) + 1e-10)
|
||||
|
||||
# Cross-correlate
|
||||
print("Computing cross-correlation...", file=sys.stderr)
|
||||
corr = signal.fftconvolve(yt_norm, local_chunk[::-1], mode="full")
|
||||
corr = signal.fftconvolve(yt_norm, local_norm[::-1], mode="full")
|
||||
corr_abs = np.abs(corr)
|
||||
|
||||
# The peak in corr tells us where local_chunk best aligns in yt_norm
|
||||
peak_index = np.argmax(np.abs(corr))
|
||||
# Extract top peaks with minimum separation
|
||||
candidates = []
|
||||
corr_work = corr_abs.copy()
|
||||
gap = sr * MIN_PEAK_GAP_S
|
||||
|
||||
# Convert index to lag: lag = peak_index - (len(local_chunk) - 1)
|
||||
lag = peak_index - (len(local_chunk) - 1)
|
||||
for _ in range(NUM_CANDIDATES):
|
||||
idx = np.argmax(corr_work)
|
||||
lag = idx - (len(local_norm) - 1)
|
||||
offset = lag / sr
|
||||
conf = float(corr_work[idx])
|
||||
if conf <= 0:
|
||||
break
|
||||
candidates.append((round(offset, 3), round(conf, 1)))
|
||||
lo = max(0, idx - gap)
|
||||
hi = min(len(corr_work), idx + gap)
|
||||
corr_work[lo:hi] = 0
|
||||
|
||||
offset_sec = lag / sr
|
||||
confidence = float(np.abs(corr[peak_index]))
|
||||
return candidates
|
||||
|
||||
return offset_sec, confidence
|
||||
|
||||
def write_comparison_wav(
|
||||
path: str,
|
||||
local_pcm: np.ndarray,
|
||||
yt_pcm: np.ndarray,
|
||||
offset: float,
|
||||
sr: int = SAMPLE_RATE,
|
||||
) -> None:
|
||||
"""Write a stereo WAV: left = local audio, right = YouTube at offset."""
|
||||
start = int(offset * sr)
|
||||
end = start + len(local_pcm)
|
||||
if end > len(yt_pcm):
|
||||
end = len(yt_pcm)
|
||||
yt_seg = yt_pcm[start:end]
|
||||
local_seg = local_pcm[: len(yt_seg)]
|
||||
|
||||
# Normalize to same peak level
|
||||
left = local_seg / (np.max(np.abs(local_seg)) + 1e-10) * 0.8
|
||||
right = yt_seg / (np.max(np.abs(yt_seg)) + 1e-10) * 0.8
|
||||
|
||||
l16 = (left * 32767).astype(np.int16)
|
||||
r16 = (right * 32767).astype(np.int16)
|
||||
stereo = np.column_stack([l16, r16]).flatten()
|
||||
|
||||
with wave.open(path, "w") as w:
|
||||
w.setnchannels(2)
|
||||
w.setsampwidth(2)
|
||||
w.setframerate(sr)
|
||||
w.writeframes(stereo.tobytes())
|
||||
|
||||
|
||||
def main():
|
||||
|
|
@ -112,6 +150,11 @@ def main():
|
|||
)
|
||||
parser.add_argument("local_audio", help="Path to the local audio file (e.g. audio.ogg)")
|
||||
parser.add_argument("youtube_id", help="YouTube video ID (e.g. 5NV6Rdv1a3I)")
|
||||
parser.add_argument(
|
||||
"--out-dir",
|
||||
default=None,
|
||||
help="Directory to write comparison WAV files (default: ./compare_<youtube_id>/)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--json", action="store_true",
|
||||
help="Output result as JSON (for scripting)",
|
||||
|
|
@ -122,43 +165,59 @@ def main():
|
|||
print(f"Error: {args.local_audio} not found", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
out_dir = args.out_dir or f"compare_{args.youtube_id}"
|
||||
os.makedirs(out_dir, exist_ok=True)
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
# Download YouTube audio
|
||||
yt_path = download_youtube_audio(args.youtube_id, tmp)
|
||||
|
||||
# Decode both to PCM
|
||||
print("Decoding local audio...", file=sys.stderr)
|
||||
local_pcm = audio_to_pcm(args.local_audio)
|
||||
print("Decoding YouTube audio...", file=sys.stderr)
|
||||
yt_pcm = audio_to_pcm(yt_path)
|
||||
|
||||
print(
|
||||
f"Local: {len(local_pcm)/SAMPLE_RATE:.1f}s, "
|
||||
f"YouTube: {len(yt_pcm)/SAMPLE_RATE:.1f}s",
|
||||
file=sys.stderr,
|
||||
)
|
||||
local_dur = len(local_pcm) / SAMPLE_RATE
|
||||
yt_dur = len(yt_pcm) / SAMPLE_RATE
|
||||
print(f"Local: {local_dur:.1f}s, YouTube: {yt_dur:.1f}s", file=sys.stderr)
|
||||
|
||||
# Find offset
|
||||
offset, confidence = find_offset(local_pcm, yt_pcm)
|
||||
# Find candidate offsets
|
||||
candidates = find_candidates(local_pcm, yt_pcm)
|
||||
|
||||
# Generate comparison WAVs
|
||||
print(f"\nWriting comparison files to {out_dir}/", file=sys.stderr)
|
||||
print(" Left ear = local audio, Right ear = YouTube audio\n", file=sys.stderr)
|
||||
|
||||
for offset, conf in candidates:
|
||||
label = f"{offset:.1f}".replace(".", "_").replace("-", "neg")
|
||||
filename = f"offset_{label}s.wav"
|
||||
filepath = os.path.join(out_dir, filename)
|
||||
write_comparison_wav(filepath, local_pcm, yt_pcm, offset)
|
||||
m, s = divmod(abs(offset), 60)
|
||||
print(
|
||||
f" {filename:<28s} offset={offset:>8.3f}s ({int(m)}:{s:05.2f}) "
|
||||
f"confidence={conf:.0f}",
|
||||
file=sys.stderr,
|
||||
)
|
||||
|
||||
if args.json:
|
||||
print(json.dumps({
|
||||
"youtube_id": args.youtube_id,
|
||||
"offset_seconds": round(offset, 3),
|
||||
"confidence": round(confidence, 2),
|
||||
"local_duration": round(len(local_pcm) / SAMPLE_RATE, 2),
|
||||
"youtube_duration": round(len(yt_pcm) / SAMPLE_RATE, 2),
|
||||
"candidates": [
|
||||
{"offset": off, "confidence": conf} for off, conf in candidates
|
||||
],
|
||||
"compare_dir": out_dir,
|
||||
"local_duration": round(local_dur, 2),
|
||||
"youtube_duration": round(yt_dur, 2),
|
||||
}, indent=2))
|
||||
else:
|
||||
print(f"\nResult:", file=sys.stderr)
|
||||
print(f" YouTube offset: {offset:.3f}s", file=sys.stderr)
|
||||
print(f" Confidence: {confidence:.1f}", file=sys.stderr)
|
||||
print(
|
||||
f" Interpretation: local audio t=0 corresponds to YouTube t={offset:.3f}s",
|
||||
f"\nListen to each file with headphones. The one where both "
|
||||
f"ears are in sync is the correct offset.",
|
||||
file=sys.stderr,
|
||||
)
|
||||
# Print just the offset to stdout for piping
|
||||
print(f"{offset:.3f}")
|
||||
# Print best candidate to stdout for piping
|
||||
if candidates:
|
||||
print(f"{candidates[0][0]:.3f}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
|
|
|||
Loading…
Reference in New Issue