File size: 4,771 Bytes
dd2737e 69261fe dd2737e 69261fe dd2737e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 | """Numpy-only audio helpers for the FastWave AX650 SDK.
- WAV read/write (PCM16 / PCM24 / float32) without soundfile
- polyphase resampling (scipy.signal.resample_poly compatible)
- band one-hot, hann window, overlap-add
Only numpy is required on the board.
"""
import math
import numpy as np
SR_OUT = 48000
N_FFT = 1024
HOP = 256
FREQ = N_FFT // 2 + 1
CHUNK = 32768
# --------------------------------------------------------------------------- WAV
def read_wav(path):
"""Return (float32 mono in [-1, 1], sample_rate)."""
import wave
with wave.open(path, "rb") as w:
sr = w.getframerate()
n_ch = w.getnchannels()
sw = w.getsampwidth()
n = w.getnframes()
raw = w.readframes(n)
if sw == 2:
x = np.frombuffer(raw, dtype="<i2").astype(np.float32) / 32768.0
elif sw == 3:
b = np.frombuffer(raw, dtype=np.uint8).reshape(-1, 3).astype(np.int32)
v = b[:, 0] | (b[:, 1] << 8) | (b[:, 2] << 16)
v = np.where(v >= 1 << 23, v - (1 << 24), v)
x = v.astype(np.float32) / (1 << 23)
elif sw == 4:
x = np.frombuffer(raw, dtype="<i4").astype(np.float32) / (1 << 31)
else:
raise ValueError(f"unsupported sample width: {sw}")
if n_ch > 1:
x = x.reshape(-1, n_ch).mean(axis=1)
return np.ascontiguousarray(x, dtype=np.float32), sr
def write_wav(path, x, sr):
import wave
x = np.asarray(x, dtype=np.float32)
x = np.clip(x, -1.0, 1.0)
pcm = (x * 32767.0).astype("<i2")
with wave.open(path, "wb") as w:
w.setnchannels(1)
w.setsampwidth(2)
w.setframerate(int(sr))
w.writeframes(pcm.tobytes())
# --------------------------------------------------------------------- resampling
def _firwin(cutoff, numtaps, beta=5.0):
"""Lowpass FIR (same design as scipy.signal.firwin(..., window=('kaiser', beta)))."""
n = np.arange(numtaps)
alpha = (numtaps - 1) / 2.0
m = n - alpha
h = cutoff * np.sinc(cutoff * m)
h *= np.kaiser(numtaps, beta)
h /= h.sum()
return h
def resample_poly(x, up, down):
"""Polyphase rational resampling, scipy.signal.resample_poly compatible."""
x = np.asarray(x, dtype=np.float64)
g = math.gcd(int(up), int(down))
up, down = int(up) // g, int(down) // g
if up == down:
return x.astype(np.float32)
max_rate = max(up, down)
half = 10 * max_rate
h = (_firwin(1.0 / max_rate, 2 * half + 1) * up).astype(np.float64)
hl = len(h)
n_out = int(math.ceil(len(x) * up / down))
# polyphase evaluation: out[k] = sum_j h[j] * x[(half + k*down - j)/up]
base = half + np.arange(n_out, dtype=np.int64) * down
m0 = (base - hl + 1 + up - 1) // up
m0 = np.maximum(m0, 0)
acc = np.zeros(n_out, dtype=np.float64)
for t in range(hl // up + 2):
m = m0 + t
j = base - m * up
valid = (j >= 0) & (j < hl) & (m < len(x))
acc[valid] += h[j[valid]] * x[m[valid]]
return acc.astype(np.float32)
def resample_to(x, sr_in, sr_out):
if sr_in == sr_out:
return np.asarray(x, dtype=np.float32).copy()
return resample_poly(x, int(sr_out), int(sr_in))
def low_quality(x, sr_in, out_sr=SR_OUT):
"""Upsample the band-limited input to out_sr (same role as the training pipeline)."""
n_out = int(round(len(x) * out_sr / sr_in))
y = resample_to(x, sr_in, out_sr)
if len(y) > n_out:
y = y[:n_out]
elif len(y) < n_out:
y = np.pad(y, (0, n_out - len(y)))
return y
# ------------------------------------------------------------------------ misc
def band_onehot(sr_in, sr_out=SR_OUT, freq=FREQ):
"""(2, freq) float32 one-hot of usable frequency bins."""
hi = float(sr_in) / float(sr_out)
band = np.zeros(freq, dtype=np.int64)
band[: int(hi * freq)] = 1
# matches F.one_hot(band).transpose(1, -1): ch0 = high band, ch1 = low band
out = np.zeros((2, freq), dtype=np.float32)
out[0, band == 0] = 1.0
out[1, band == 1] = 1.0
return out
def hann(n):
return (0.5 - 0.5 * np.cos(2.0 * np.pi * np.arange(n) / n)).astype(np.float32)
def chunk_bounds(n, chunk, hop):
"""Start positions of overlapping chunks covering n samples."""
if n <= chunk:
return [0]
starts = list(range(0, n - chunk + 1, hop))
if starts[-1] + chunk < n:
starts.append(n - chunk)
return starts
def overlap_add(chunks, starts, n, chunk):
"""Hann-window overlap-add."""
w = hann(chunk)
acc = np.zeros(n + chunk, dtype=np.float64)
wsum = np.zeros(n + chunk, dtype=np.float64)
for c, s in zip(chunks, starts):
acc[s: s + chunk] += c * w
wsum[s: s + chunk] += w
wsum = np.maximum(wsum, 1e-8)
return (acc[:n] / wsum[:n]).astype(np.float32)
|