| """Convert waveforms without changing their declared duration.""" |
| from __future__ import annotations |
| import math |
| import unicodedata |
| from dataclasses import dataclass |
| import numpy as np |
| from numpy.typing import NDArray |
| from scipy.signal import resample_poly |
| Waveform = NDArray[np.float32] |
| CLIPPING_THRESHOLD = 0.999 |
|
|
| @dataclass(frozen=True) |
| class Signal: |
| samples: int |
| rate_hz: int |
| peak: float |
| rms_dbfs: float | None |
| clipping_fraction: float |
|
|
| def text_key(text: str) -> str: |
| return ' '.join(unicodedata.normalize('NFC', text).split()) |
|
|
| def validate_waveform(waveform: Waveform, rate_hz: int) -> None: |
| if waveform.ndim != 1 or waveform.size == 0 or rate_hz <= 0: |
| message = f'invalid waveform shape={waveform.shape}, rate_hz={rate_hz}' |
| raise ValueError(message) |
| if not np.isfinite(waveform).all(): |
| offset = int(np.flatnonzero(~np.isfinite(waveform))[0]) |
| message = f'non-finite waveform value={waveform[offset]} at sample={offset}' |
| raise ValueError(message) |
|
|
| def measure_signal(waveform: Waveform, rate_hz: int) -> Signal: |
| validate_waveform(waveform, rate_hz) |
| rms = float(np.sqrt(np.mean(np.square(waveform, dtype=np.float64)))) |
| return Signal(waveform.size, rate_hz, float(np.abs(waveform).max()), 20 * math.log10(rms) if rms > 0 else None, float(np.mean(np.abs(waveform) >= CLIPPING_THRESHOLD))) |
|
|
| def resample_waveform(waveform: Waveform, rate_hz: int, target_rate_hz: int=24000) -> Waveform: |
| validate_waveform(waveform, rate_hz) |
| if target_rate_hz <= 0: |
| message = f'invalid target_rate_hz={target_rate_hz}' |
| raise ValueError(message) |
| divisor = math.gcd(rate_hz, target_rate_hz) |
| converted = np.asarray(resample_poly(waveform, target_rate_hz // divisor, rate_hz // divisor, window=('kaiser', 8.6)), dtype=np.float32) |
| expected = (waveform.size * target_rate_hz + rate_hz - 1) // rate_hz |
| if converted.size != expected: |
| message = f'resampling returned {converted.size} samples, expected {expected}' |
| raise ValueError(message) |
| validate_waveform(converted, target_rate_hz) |
| return converted |
|
|
