import unittest import numpy as np from scipy import signal from blurt_audio import OUTPUT_RATE, StreamingAudio def tone(rate, seconds, amplitude=0.1): t = np.arange(round(rate * seconds)) / rate return (amplitude * np.sin(2 * np.pi * 220 * t)).astype(np.float32) class StreamingAudioTests(unittest.TestCase): def collect(self, rate, audio, block): stream = StreamingAudio(rate, 1) chunks = [] for pos in range(0, len(audio), block): chunks.extend(stream.feed(audio[pos : pos + block])) chunks.extend(stream.finish()) return chunks def assert_coverage(self, chunks, expected): cursor = 0 for chunk in chunks: self.assertEqual(chunk.start, cursor) self.assertEqual(len(chunk.samples), chunk.end - chunk.start) cursor = chunk.end self.assertEqual(cursor, expected) def test_output_length_and_ranges_are_block_size_independent(self): for rate in (16_000, 44_100, 48_000): audio = tone(rate, 10.123) expected = round(len(audio) * OUTPUT_RATE / rate) reference = self.collect(rate, audio, len(audio)) split = self.collect(rate, audio, 997) self.assert_coverage(reference, expected) self.assert_coverage(split, expected) self.assertEqual([(c.start, c.end) for c in reference], [(c.start, c.end) for c in split]) def test_strong_pause_after_minimum_cuts_during_feed(self): audio = np.concatenate((tone(OUTPUT_RATE, 8.2), np.zeros(int(OUTPUT_RATE)))) stream = StreamingAudio(OUTPUT_RATE, 1) chunks = stream.feed(audio) self.assertEqual(len(chunks), 1) self.assertGreater(chunks[0].end, 8 * OUTPUT_RATE) self.assertLess(chunks[0].end, 9.2 * OUTPUT_RATE) chunks += stream.finish() self.assert_coverage(chunks, len(audio)) def test_shorter_pause_is_accepted_near_target(self): audio = np.concatenate((tone(OUTPUT_RATE, 20), np.zeros(int(.6 * OUTPUT_RATE)), tone(OUTPUT_RATE, 1))) chunks = self.collect(OUTPUT_RATE, audio, 777) self.assertGreaterEqual(len(chunks), 2) self.assertLess(chunks[0].end, int(20.7 * OUTPUT_RATE)) self.assert_coverage(chunks, len(audio)) def test_uninterrupted_audio_forces_bounded_chunks(self): audio = tone(OUTPUT_RATE, 92) chunks = self.collect(OUTPUT_RATE, audio, 1024) self.assertGreaterEqual(len(chunks), 3) self.assertTrue(all(c.end - c.start <= 45 * OUTPUT_RATE for c in chunks)) self.assert_coverage(chunks, len(audio)) def test_silence_is_retained_but_marked_not_speech(self): audio = np.zeros(10 * OUTPUT_RATE, dtype=np.float32) chunks = self.collect(OUTPUT_RATE, audio, 1000) self.assert_coverage(chunks, len(audio)) self.assertTrue(all(not c.speech for c in chunks)) def test_quiet_speech_is_preserved_and_uploaded(self): audio = tone(OUTPUT_RATE, 9, amplitude=2e-4) chunks = self.collect(OUTPUT_RATE, audio, 1234) self.assert_coverage(chunks, len(audio)) self.assertTrue(all(c.speech for c in chunks)) self.assertTrue(all(c.end - c.start >= 8 * OUTPUT_RATE for c in chunks[:-1])) def test_tiny_tail_and_checkpoint_are_non_mutating(self): stream = StreamingAudio(OUTPUT_RATE, 1) stream.feed(tone(OUTPUT_RATE, 0.031)) first = stream.checkpoint() second = stream.checkpoint() np.testing.assert_array_equal(first, second) chunks = stream.finish() self.assertEqual(len(chunks), 1) self.assert_coverage(chunks, round(0.031 * OUTPUT_RATE)) def test_multichannel_is_mixed_to_mono(self): mono = tone(48_000, 1) stereo = np.column_stack((mono, mono)) stream = StreamingAudio(48_000, 2) stream.feed(stereo) chunks = stream.finish() self.assert_coverage(chunks, 16_000) def test_samples_are_callback_size_independent(self): for rate in (16_000, 44_100, 48_000): audio = np.concatenate((tone(rate, 8.3), np.zeros(round(rate * .9)), tone(rate, 2.1))) large = self.collect(rate, audio, len(audio)) small = self.collect(rate, audio, 317) self.assertEqual([(c.start, c.end) for c in large], [(c.start, c.end) for c in small]) np.testing.assert_array_equal(np.concatenate([c.samples for c in large]), np.concatenate([c.samples for c in small])) def test_resampler_joins_agree_with_whole_clip_reference(self): rate = 44_100 source = tone(rate, 5.25) + 0.015 * np.sin( 2 * np.pi * 1700 * np.arange(round(rate * 5.25)) / rate ).astype(np.float32) chunks = self.collect(rate, source, 271) actual = np.concatenate([c.samples for c in chunks]).astype(np.float32) / 32767 reference = signal.resample_poly(source, 160, 441) sos = signal.butter(2, 70, btype="highpass", fs=OUTPUT_RATE, output="sos") zi = signal.sosfilt_zi(sos) * float(reference[0]) reference = signal.sosfilt(sos, reference, zi=zi)[0] rms = float(np.sqrt(np.mean(reference * reference))) peak = float(np.percentile(np.abs(reference), 99.9)) gain = min(10 ** (-20 / 20) / rms, 10 ** (-1 / 20) / peak, 10 ** (30 / 20)) reference = np.clip(reference * gain, -1, 1) self.assertLess(float(np.mean(np.abs(actual - reference))), 2e-4) for join in range(OUTPUT_RATE, len(actual), OUTPUT_RATE): self.assertLess(float(np.max(np.abs(actual[join - 16:join + 16] - reference[join - 16:join + 16]))), .003) def test_long_recording_discards_sealed_history(self): stream = StreamingAudio(OUTPUT_RATE, 1) emitted = [] block = tone(OUTPUT_RATE, 1) max_retained = 0 for _ in range(100): emitted.extend(stream.feed(block)) max_retained = max(max_retained, len(stream._resampled)) self.assertLessEqual(stream._pending_len, stream.rate + round(stream.rate * .05)) emitted.extend(stream.finish()) self.assertLessEqual(max_retained, 45 * OUTPUT_RATE) self.assert_coverage(emitted, 100 * OUTPUT_RATE) def test_highpass_state_is_continuous_across_resampler_batches(self): rate = 48_000 source = tone(rate, 4, amplitude=.03) + np.float32(.2) chunks = self.collect(rate, source, 113) actual = np.concatenate([c.samples for c in chunks]).astype(np.float32) # No repeated DC-step transient should appear at one-second batch joins. for join in range(OUTPUT_RATE, len(actual), OUTPUT_RATE): local = actual[join - 100:join + 100] baseline = actual[join - 500:join - 300] self.assertLess(abs(float(np.sqrt(np.mean(local * local))) - float(np.sqrt(np.mean(baseline * baseline)))), 100) if __name__ == "__main__": unittest.main()