1import numpy as np
3from cymsim.decimation import DECIMATION_SPAN, StreamingDecimator, decimation_kernel, decimation_taps
6def test_kernel_is_normalised_linear_phase_and_attenuates_the_stopband():
7 m, fs = 54, 44100
8 h = decimation_kernel(m)
9 W = DECIMATION_SPAN
10 assert len(h) == W * m and np.isclose(h.sum(), 1.0)
11 d = np.arange(-W * m // 2, W * m // 2)
12 assert np.allclose(h[1:], h[::-1][:-1]) # symmetric about d = 0 (odd length ignored)
13 f = np.array([1000.0, 15000.0, 18000.0, 22050.0, 24000.0, 30000.0])
14 H = np.abs(np.exp(-2j * np.pi * np.outer(f / (fs * m), d)) @ h)
15 assert H[0] > 0.999 and H[1] > 0.99 and H[2] > 0.95 # passband
16 assert H[3] < 0.1 and H[4] < 1e-3 and H[5] < 1e-4 # transition and stopband
19def test_polyphase_layout_matches_the_kernel():
20 m = 10
21 h, taps = decimation_kernel(m), decimation_taps(m)
22 W = DECIMATION_SPAN
23 assert taps.shape == (W, m)
24 for w, kk in [(0, 0), (7, 3), (W - 1, m - 1)]:
25 d = kk + (W // 2 - 1 - w) * m
26 assert taps[w, kk] == h[d + W * m // 2]
27 assert np.isclose(taps.sum(), 1.0)
30def test_streaming_decimator_matches_direct_convolution_for_any_chunking():
31 m = 7
32 h = decimation_kernel(m)
33 W = DECIMATION_SPAN
34 rng = np.random.default_rng(0)
35 x = rng.standard_normal(m * 400)
36 xz = np.concatenate([np.zeros(W * m // 2), x]) # rest before t = 0
37 n_final = (len(x) - W * m // 2) // m + 1
38 direct = np.array([xz[j * m:j * m + W * m] @ h for j in range(n_final)])
39 dec = StreamingDecimator(m)
40 out, i = [], 0
41 for L in [1, 3, 50, 999, 7, 300, 2000]:
42 out.append(dec.push(x[i:i + L]))
43 i += L
44 out = np.concatenate(out)
45 assert len(out) == n_final
46 assert np.abs(out - direct).max() < 1e-12
47 # resuming from the saved tail continues the stream exactly
48 dec2 = StreamingDecimator(m, tail=dec.tail, n_seen=dec.n_seen)
49 more = rng.standard_normal(m * 50)
50 a = dec.push(more)
51 b = dec2.push(more)
52 assert len(a) == 50 and np.array_equal(a, b)