src/audio.py

"""Deterministic 16 kHz waveform -> log-mel features; standard-library/scipy implementation."""
import numpy as np
from scipy.io import wavfile
from scipy.signal import resample_poly, stft
from scipy.fft import dct

def to_float(y):
    """Scale PCM samples to float32 in [-1,1]; must run before mixing to mono, which drops the integer dtype."""
    y=np.asarray(y)
    if y.dtype==np.uint8: return (y.astype(np.float32)-128.)/128.
    if np.issubdtype(y.dtype,np.integer):
        return y.astype(np.float32)/max(abs(np.iinfo(y.dtype).min),np.iinfo(y.dtype).max)
    return y.astype(np.float32)

def read_audio(path):
    """(sample_rate, mono float32 waveform) from WAV via scipy, or FLAC/OGG/MP3/... via the optional soundfile package."""
    path=str(path)
    if path.lower().endswith(('.wav','.wave')):
        try:
            sr,y=wavfile.read(path); y=to_float(y)
        except ValueError:
            # Compressed or unusual WAV encodings scipy cannot parse; libsndfile often can.
            sr,y=_read_soundfile(path)
    else: sr,y=_read_soundfile(path)
    if y.ndim==2: y=y.mean(axis=1)
    return int(sr),y.astype(np.float32)

def _read_soundfile(path):
    try: import soundfile as sf
    except ImportError:
        raise RuntimeError(f"Reading {path} needs the soundfile package (pip install soundfile), "
                           "or convert it first: ffmpeg -i input.mp3 -ac 1 output.wav") from None
    y,sr=sf.read(path,dtype='float32',always_2d=False)
    return sr,y

SR=16000
DURATION=2.0
N_MELS=64
N_FFT=512
HOP=160

def _hz_to_mel(hz): return 2595.0*np.log10(1.0+np.asarray(hz)/700.0)
def _mel_to_hz(mel): return 700.0*(10.0**(np.asarray(mel)/2595.0)-1.0)
def mel_filterbank(sr=SR, n_fft=N_FFT, n_mels=N_MELS):
    points=_mel_to_hz(np.linspace(_hz_to_mel(0),_hz_to_mel(sr/2),n_mels+2))
    bins=np.floor((n_fft+1)*points/sr).astype(int)
    fb=np.zeros((n_mels,n_fft//2+1),dtype=np.float32)
    for m in range(1,n_mels+1):
        left,center,right=bins[m-1:m+2]
        if center>left:
            fb[m-1,left:center]=np.arange(left,center)-left
            fb[m-1,left:center]/=(center-left)
        if right>center:
            fb[m-1,center:right]=(right-np.arange(center,right))/(right-center)
    return fb

_FB=mel_filterbank()
def resample(y, sr):
    y=np.asarray(y,dtype=np.float32).reshape(-1)
    if sr != SR:
        import math
        g=math.gcd(int(sr),SR)
        y=resample_poly(y,SR//g,int(sr)//g).astype(np.float32)
    return y

def crop_offsets(n_samples, k):
    """k evenly spaced 2 s crop starts covering a 16 kHz waveform; k=1 gives the center crop."""
    n=int(SR*DURATION)
    if n_samples<=n or k<=1: return [None]
    return [int(round(o)) for o in np.linspace(0,n_samples-n,k)]

def prepare_waveform(y, sr, offset=None):
    y=resample(y,sr)
    n=int(SR*DURATION)
    if len(y)>n:
        # Center crop by default: deterministic inference and evaluation.
        start=(len(y)-n)//2 if offset is None else min(max(0,int(offset)),len(y)-n)
        y=y[start:start+n]
    elif len(y)<n:
        y=np.pad(y,(0,n-len(y)))
    peak=float(np.max(np.abs(y))) if len(y) else 0.
    if peak>1e-8: y=y/peak
    return y

def logmel(y, sr=SR, offset=None):
    y=prepare_waveform(y,sr,offset)
    _,_,z=stft(y,fs=SR,nperseg=N_FFT,noverlap=N_FFT-HOP,
               nfft=N_FFT,boundary='zeros',padded=True)
    power=(np.abs(z)**2).astype(np.float32)
    mel=_FB @ power
    feat=np.log(np.maximum(mel,1e-8))
    # Per-record normalization stabilizes recorder gain; retain a consistent transformation.
    feat=(feat-feat.mean())/(feat.std()+1e-5)
    return feat.astype(np.float32)[None,:,:]

def audio_quality(y):
    y=np.asarray(y,dtype=np.float32)
    if y.size==0: return False, ['empty_audio']
    rms=float(np.sqrt(np.mean(y*y)))
    peak=float(np.max(np.abs(y)))
    reasons=[]
    if rms<0.002: reasons.append('low_level_or_silence')
    if peak>0.999: reasons.append('possible_clipping')
    return len(reasons)==0, reasons