File size: 2,125 Bytes
3699b0a
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""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