54 lines
1.5 KiB
Python
54 lines
1.5 KiB
Python
# 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 |