137 lines
4.1 KiB
Python
137 lines
4.1 KiB
Python
"""Portable audio helpers (no external programs needed).
|
|
|
|
Audio is handled as float32 numpy arrays shaped (frames, channels).
|
|
Decoding uses libsndfile (WAV, FLAC, OGG/Vorbis/Opus, MP3, AIFF, ...); other formats (M4A, WMA, ...)
|
|
fall back to an ffmpeg found on the system, if any. Resampling uses soxr.
|
|
"""
|
|
import io
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
|
|
import numpy as np
|
|
import soundfile as sf
|
|
|
|
READABLE = (".wav", ".flac", ".ogg", ".oga", ".opus", ".mp3", ".aif", ".aiff", ".au", ".w64", ".caf")
|
|
FFMPEG_ONLY = (".m4a", ".aac", ".wma", ".mp4", ".webm", ".mkv", ".ac3")
|
|
|
|
|
|
class AudioError(Exception):
|
|
pass
|
|
|
|
|
|
def _as2d(x):
|
|
x = np.asarray(x, dtype=np.float32)
|
|
return x[:, None] if x.ndim == 1 else x
|
|
|
|
|
|
def load(path):
|
|
"""Read an audio file -> (float32 array (n, ch), rate)."""
|
|
try:
|
|
x, sr = sf.read(path, dtype="float32", always_2d=True)
|
|
return x, sr
|
|
except (sf.LibsndfileError, RuntimeError, TypeError) as e:
|
|
ff = shutil.which("ffmpeg")
|
|
if not ff:
|
|
raise AudioError("Cannot read '%s' (%s). Supported: WAV, FLAC, OGG, OPUS, MP3, AIFF. "
|
|
"Convert the file to one of these, or install ffmpeg for other formats."
|
|
% (os.path.basename(path), e))
|
|
raw = subprocess.run([ff, "-v", "error", "-i", path, "-f", "wav", "-"], capture_output=True)
|
|
if raw.returncode:
|
|
raise AudioError("ffmpeg could not read '%s': %s" % (path, raw.stderr.decode(errors="replace")[-300:]))
|
|
x, sr = sf.read(io.BytesIO(raw.stdout), dtype="float32", always_2d=True)
|
|
return x, sr
|
|
|
|
|
|
def save_wav(path, x, rate):
|
|
sf.write(path, _as2d(x), rate, subtype="PCM_16")
|
|
|
|
|
|
def resample(x, rate, new_rate):
|
|
x = _as2d(x)
|
|
if rate == new_rate or len(x) == 0:
|
|
return x
|
|
import soxr
|
|
return soxr.resample(x, rate, new_rate, quality="HQ").astype(np.float32)
|
|
|
|
|
|
def to_channels(x, channels):
|
|
"""Up/down-mix. Mono is copied to every channel; extra channels get the average."""
|
|
x = _as2d(x)
|
|
cur = x.shape[1]
|
|
if cur == channels:
|
|
return x
|
|
if channels == 1:
|
|
return x.mean(axis=1, keepdims=True)
|
|
if cur == 1:
|
|
return np.repeat(x, channels, axis=1)
|
|
out = np.empty((len(x), channels), dtype=np.float32)
|
|
keep = min(cur, channels)
|
|
out[:, :keep] = x[:, :keep]
|
|
if channels > keep:
|
|
out[:, keep:] = x.mean(axis=1, keepdims=True)
|
|
return out
|
|
|
|
|
|
def trim(x, rate, start=None, end=None):
|
|
x = _as2d(x)
|
|
a = int(round((start or 0) * rate))
|
|
b = len(x) if end is None else int(round(end * rate))
|
|
return x[max(0, a):max(a, min(len(x), b))]
|
|
|
|
|
|
def gain(x, db):
|
|
return (_as2d(x) * (10 ** (db / 20.0))).astype(np.float32)
|
|
|
|
|
|
def peak_db(x):
|
|
p = float(np.max(np.abs(x))) if len(x) else 0.0
|
|
return -120.0 if p <= 0 else 20 * np.log10(p)
|
|
|
|
|
|
def rms_db(x):
|
|
r = float(np.sqrt(np.mean(np.square(x)))) if len(x) else 0.0
|
|
return -120.0 if r <= 0 else 20 * np.log10(r)
|
|
|
|
|
|
def normalize(x, peak=-1.0):
|
|
"""Scale so the highest peak is `peak` dBFS."""
|
|
p = peak_db(x)
|
|
return _as2d(x) if p <= -119 else gain(x, peak - p)
|
|
|
|
|
|
def match_loudness(x, reference):
|
|
"""Scale x to the RMS loudness of `reference` (limited so it does not clip)."""
|
|
g = rms_db(reference) - rms_db(x)
|
|
g = min(g, -0.3 - peak_db(x))
|
|
return gain(x, g)
|
|
|
|
|
|
def trim_silence(x, threshold_db=-50.0, pad=0.02, rate=48000):
|
|
x = _as2d(x)
|
|
lvl = np.max(np.abs(x), axis=1)
|
|
idx = np.nonzero(lvl > 10 ** (threshold_db / 20.0))[0]
|
|
if len(idx) == 0:
|
|
return x
|
|
p = int(pad * rate)
|
|
return x[max(0, idx[0] - p):min(len(x), idx[-1] + p)]
|
|
|
|
|
|
def conform(x, rate, new_rate, channels):
|
|
"""Resample + channel conversion + clip guard, ready for encoding."""
|
|
y = to_channels(resample(x, rate, new_rate), channels)
|
|
return np.clip(y, -1.0, 1.0)
|
|
|
|
|
|
def decode_wem(data: bytes):
|
|
"""Decode a game WEM -> (float32 array (n, ch), rate), using the game's codebooks."""
|
|
from .convert import wem_to_ogg
|
|
from .wem import Wem
|
|
ogg, _ = wem_to_ogg(Wem.parse(data))
|
|
x, sr = sf.read(io.BytesIO(ogg), dtype="float32", always_2d=True)
|
|
return x, sr
|
|
|
|
|
|
def duration(x, rate):
|
|
return len(x) / float(rate) if rate else 0.0
|