diff --git a/freedata_server/modem.py b/freedata_server/modem.py index 873a88c0..7c5346dd 100644 --- a/freedata_server/modem.py +++ b/freedata_server/modem.py @@ -71,6 +71,26 @@ class RF: self.AUDIO_STREAMING_CHUNK_SIZE = 2400 self.audio_out_queue = queue.Queue() + # Size of the block the RX DSP chain runs on, in 48 kHz samples. The input + # stream is opened with blocksize=0 (PortAudio picks the capture size per + # device), so the DSP must not rely on the captured block size: + # rx_audio_processing_worker re-blocks whatever the callback is handed + # to exactly this size, so a device or setting that delivers some other + # size cannot reach the DSP. 4800 (100 ms, 800 samples at 8 kHz) is the + # size the rest of the chain is built around: + # * codec2.resampler.resample48_to_8 asserts len % FDMDV_OS_48 (6) == 0 + # and raises AssertionError otherwise, + # * audio.calculate_fft pads/truncates to 800 samples at 8 kHz, so a + # shorter block is mostly zero padding and the spectrum, channel busy + # detection and audio_dbfs all degrade, + # * enqueue_streaming_audio_chunks zero-pads every block up to + # AUDIO_STREAMING_CHUNK_SIZE, so a shorter block streams mostly + # silence and emits one chunk per block regardless of size, + # * audio.normalize_audio (rx_auto_audio_level, on by default) + # normalizes per block, so a shorter block means a faster, jumpier AGC. + # Lowering this lowers RX latency, but needs those four addressed first. + self.RX_DSP_BLOCK_48K = 4800 + # Make sure our resampler will work assert (self.AUDIO_SAMPLE_RATE / self.modem_sample_rate) == codec2.api.FDMDV_OS_48 # type: ignore @@ -84,6 +104,11 @@ class RF: self.rx_audio_worker_running = False self.rx_audio_worker_thread = None self.rx_audio_dropped_blocks = 0 + # 48 kHz samples that have been captured but do not yet fill a whole + # RX_DSP_BLOCK_48K; they are carried over to the next captured block so the + # sample stream handed to the resampler stays gapless (its filter memory + # depends on that) and always has a valid length. + self.rx_audio_carry_48k = np.empty(0, dtype=np.int16) self.demodulator = demodulator.Demodulator(self.ctx) self.modulator = modulator.Modulator(self.ctx) @@ -179,14 +204,20 @@ class RF: self.resampler = codec2.resampler() # SoundDevice audio input stream - # blocksize=0 lets PortAudio deliver small blocks, keeping RX - # buffering delay low. latency=0.2 sets the ring depth explicitly: - # the default ("high") can negotiate as little as two periods on - # some devices (measured on snd-aloop, where a two period ring at - # 100 ms periods drops audio continuously), and with blocksize=0 - # alone the ring can come out as shallow as 40 ms. An explicit - # 200 ms request gives a deep ring of small periods on every - # device we measured (CM108 hardware and snd-aloop alike). + # blocksize=0 lets PortAudio pick the capture block size per device. + # This is deliberate and load-bearing: a fixed blocksize=4800 makes + # some virtual devices (measured on snd-aloop) starve/overflow -- + # a two period ring at 100 ms periods drops audio continuously and + # the modem is deaf. latency=0.2 sets the ring depth explicitly so + # the ring comes out deep (many small periods) on every device + # measured (CM108 hardware and snd-aloop alike). + # The capture block size is decoupled from the DSP block size: + # PortAudio may deliver any block length here (512/1024-class blocks + # are common on real hardware, and are NOT a multiple of codec2's + # FDMDV_OS_48 == 6, which resample48_to_8 asserts on). + # rx_audio_processing_worker re-blocks whatever arrives into exact + # RX_DSP_BLOCK_48K blocks, so no capture size can reach the DSP + # chain short or misaligned; see the note on that constant. self.sd_input_stream = sd.InputStream( channels=1, dtype="int16", @@ -199,6 +230,7 @@ class RF: self.sd_input_stream.start() # process RX audio off the real-time callback thread + self.rx_audio_carry_48k = np.empty(0, dtype=np.int16) # no stale audio across restarts self.rx_audio_worker_running = True self.rx_audio_worker_thread = threading.Thread( target=self.rx_audio_processing_worker, @@ -476,33 +508,70 @@ class RF: if indata is None: # shutdown sentinel break try: - audio_48k = np.frombuffer(indata, dtype=np.int16) - audio_8k = self.resampler.resample48_to_8(audio_48k) - - self.enqueue_streaming_audio_chunks(audio_8k, self.ctx.audio_rx_queue) - - if self.ctx.config_manager.config["AUDIO"].get("rx_auto_audio_level"): - audio_8k = audio.normalize_audio(audio_8k) - - audio_8k_level_adjusted = audio.set_audio_volume(audio_8k, self.rx_audio_level) - - if not self.ctx.state_manager.isTransmitting(): - audio.calculate_fft(audio_8k_level_adjusted, self.ctx.modem_fft, self.ctx.state_manager) - - length_audio_8k_level_adjusted = len(audio_8k_level_adjusted) - # Avoid buffer overflow by filling only if buffer for - # selected datachannel mode is not full - index = 0 - for mode in self.demodulator.MODE_DICT: - mode_data = self.demodulator.MODE_DICT[mode] - audiobuffer = mode_data["audio_buffer"] - decode = mode_data["decode"] - index += 1 - if audiobuffer: - if (audiobuffer.nbuffer + length_audio_8k_level_adjusted) > audiobuffer.size: - self.demodulator.buffer_overflow_counter[index] += 1 - self.ctx.event_manager.send_buffer_overflow(self.demodulator.buffer_overflow_counter) - elif decode: - audiobuffer.push(audio_8k_level_adjusted) + self.process_rx_audio_block(indata) except Exception as e: self.log.warning("[AUDIO EXCEPTION]", e=e) + + def process_rx_audio_block(self, indata) -> None: + """Re-blocks one captured audio block and runs the DSP chain on it. + + The input stream is free to hand the callback any block size, so the + captured samples are appended to rx_audio_carry_48k and the DSP chain is + run once per whole RX_DSP_BLOCK_48K available. Anything left over is + carried into the next captured block rather than being processed short: + a short block would fail codec2's "multiple of 6" resampler assertion and + silently degrade the FFT, streaming and AGC paths (see RX_DSP_BLOCK_48K). + + Args: + indata (np.ndarray): One captured 48 kHz int16 block, any length. + """ + captured_48k = np.frombuffer(indata, dtype=np.int16) + self.rx_audio_carry_48k = np.concatenate((self.rx_audio_carry_48k, captured_48k)) + + block = self.RX_DSP_BLOCK_48K + processed = 0 + while len(self.rx_audio_carry_48k) - processed >= block: + self.run_rx_audio_dsp(self.rx_audio_carry_48k[processed : processed + block]) + processed += block + + if processed: + # copy so the carry does not keep the whole concatenated block alive + self.rx_audio_carry_48k = self.rx_audio_carry_48k[processed:].copy() + + def run_rx_audio_dsp(self, audio_48k: np.ndarray) -> None: + """Runs the RX DSP chain on exactly one RX_DSP_BLOCK_48K of audio. + + Resamples to 8 kHz, feeds the audio streaming queue, applies the optional + auto level and the configured RX gain, updates the FFT, and pushes the + result into each decoding demodulator buffer that has room for it. + + Args: + audio_48k (np.ndarray): RX_DSP_BLOCK_48K 48 kHz int16 samples. + """ + audio_8k = self.resampler.resample48_to_8(audio_48k) + + self.enqueue_streaming_audio_chunks(audio_8k, self.ctx.audio_rx_queue) + + if self.ctx.config_manager.config["AUDIO"].get("rx_auto_audio_level"): + audio_8k = audio.normalize_audio(audio_8k) + + audio_8k_level_adjusted = audio.set_audio_volume(audio_8k, self.rx_audio_level) + + if not self.ctx.state_manager.isTransmitting(): + audio.calculate_fft(audio_8k_level_adjusted, self.ctx.modem_fft, self.ctx.state_manager) + + length_audio_8k_level_adjusted = len(audio_8k_level_adjusted) + # Avoid buffer overflow by filling only if buffer for + # selected datachannel mode is not full + index = 0 + for mode in self.demodulator.MODE_DICT: + mode_data = self.demodulator.MODE_DICT[mode] + audiobuffer = mode_data["audio_buffer"] + decode = mode_data["decode"] + index += 1 + if audiobuffer: + if (audiobuffer.nbuffer + length_audio_8k_level_adjusted) > audiobuffer.size: + self.demodulator.buffer_overflow_counter[index] += 1 + self.ctx.event_manager.send_buffer_overflow(self.demodulator.buffer_overflow_counter) + elif decode: + audiobuffer.push(audio_8k_level_adjusted) diff --git a/tests/test_rx_audio_callback.py b/tests/test_rx_audio_callback.py index 94420d1f..76310b6d 100644 --- a/tests/test_rx_audio_callback.py +++ b/tests/test_rx_audio_callback.py @@ -4,9 +4,15 @@ modem.RF.sd_input_audio_callback no longer runs the RX DSP inline; it copies the captured block onto rx_audio_in_queue and returns, and rx_audio_processing_worker drains that queue and runs the DSP (resample 48->8 kHz, FFT, demod-buffer push). +The stream is opened with blocksize=0, so the captured block size is whatever +PortAudio has available: device specific, variable, and not a multiple of +anything. rx_audio_processing_worker therefore re-blocks the captured stream to +RX_DSP_BLOCK_48K before any DSP runs on it. + These tests exercise that split directly with synthetic blocks, so they need no audio hardware (CI has none). They cover: - - the worker actually performs the relocated DSP on an enqueued block, and + - the worker actually performs the relocated DSP on an enqueued block, + - odd capture sizes are re-blocked rather than passed to the DSP short, and - the callback drops (and counts) instead of blocking when the queue is full, which is the real-time-safety property the change exists to provide. """ @@ -21,7 +27,12 @@ from freedata_server.context import AppContext from freedata_server import modem, codec2 CONFIG = "freedata_server/config.ini.example" -BLOCK_FRAMES = 4800 # one 48 kHz input block, matching sd.InputStream(blocksize=4800) + +# A capture size PortAudio really does hand us, and deliberately NOT a multiple of +# codec2's FDMDV_OS_48 (6) -- 512 % 6 == 2. Passing this straight to +# resample48_to_8 trips its "multiple of 6" assertion, which is what made the RX +# chain deaf (every block raising AssertionError) once blocksize=0 was used. +CAPTURE_FRAMES = 512 def _rf(): @@ -39,14 +50,14 @@ def _rf(): return rf -def _block(): +def _block(frames=CAPTURE_FRAMES): # sounddevice delivers indata as shape (frames, channels); int16 mono here. - return (np.random.randn(BLOCK_FRAMES, 1) * 3000).astype(np.int16) + return (np.random.randn(frames, 1) * 3000).astype(np.int16) class TestRxAudioCallbackWorkerSplit(unittest.TestCase): - def test_worker_processes_enqueued_block(self): - """A block handed to the callback is drained and DSP'd by the worker.""" + def test_worker_processes_enqueued_blocks(self): + """Blocks handed to the callback are drained and DSP'd by the worker.""" rf = _rf() rf.rx_audio_worker_running = True worker = threading.Thread(target=rf.rx_audio_processing_worker, daemon=True) @@ -54,7 +65,9 @@ class TestRxAudioCallbackWorkerSplit(unittest.TestCase): try: # status=None -> the block is enqueued (a truthy status is an # over/underflow and is dropped by the callback, unchanged by this PR). - rf.sd_input_audio_callback(_block(), BLOCK_FRAMES, None, None) + # Feed enough captured blocks to complete at least one DSP block. + for _ in range(rf.RX_DSP_BLOCK_48K // CAPTURE_FRAMES + 1): + rf.sd_input_audio_callback(_block(), CAPTURE_FRAMES, None, None) # The worker resamples to 8 kHz and feeds enqueue_streaming_audio_chunks, # which lands on ctx.audio_rx_queue -- our deterministic "DSP ran" signal. @@ -80,12 +93,76 @@ class TestRxAudioCallbackWorkerSplit(unittest.TestCase): self.assertTrue(rf.rx_audio_in_queue.full()) t0 = time.perf_counter() - rf.sd_input_audio_callback(_block(), BLOCK_FRAMES, None, None) + rf.sd_input_audio_callback(_block(), CAPTURE_FRAMES, None, None) elapsed = time.perf_counter() - t0 self.assertEqual(rf.rx_audio_dropped_blocks, 1, "a full queue must drop the block and count it") self.assertLess(elapsed, 0.05, "callback must not block on a full queue (real-time safety)") +class TestRxAudioReblocking(unittest.TestCase): + """The capture block size must never reach the DSP chain. + + blocksize=0 means PortAudio picks the size, and real devices hand back sizes + that are not multiples of codec2's FDMDV_OS_48 (6). resample48_to_8 asserts on + those, so process_rx_audio_block must accumulate instead of resampling short. + These call process_rx_audio_block directly (no worker thread) so a failure is + a raised exception rather than a swallowed, logged one. + """ + + def test_odd_capture_size_does_not_reach_the_resampler(self): + """A 512-frame capture (512 % 6 == 2) must not raise AssertionError.""" + rf = _rf() + # One short block: not enough for a DSP block, so it is carried, not resampled. + rf.process_rx_audio_block(_block(CAPTURE_FRAMES)) + self.assertEqual(len(rf.rx_audio_carry_48k), CAPTURE_FRAMES) + self.assertEqual(rf.ctx.audio_rx_queue.qsize(), 0, "a partial DSP block must not be processed short") + + def test_carry_reassembles_whole_dsp_blocks(self): + """Odd captures are accumulated into exact RX_DSP_BLOCK_48K blocks.""" + rf = _rf() + processed = [] + rf.run_rx_audio_dsp = lambda audio_48k: processed.append(len(audio_48k)) + + # A spread of sizes a real device might deliver, none a multiple of 6. + sizes = [512, 1024, 441, 512, 2048, 1024, 512, 940, 512, 1024] + for size in sizes: + rf.process_rx_audio_block(_block(size)) + + total = sum(sizes) + self.assertEqual( + processed, + [rf.RX_DSP_BLOCK_48K] * (total // rf.RX_DSP_BLOCK_48K), + "every DSP invocation must get exactly one whole block", + ) + self.assertEqual(len(rf.rx_audio_carry_48k), total % rf.RX_DSP_BLOCK_48K, "remainder must be carried over") + + def test_reblocking_preserves_the_sample_stream(self): + """No sample is dropped, duplicated or reordered by the re-blocking. + + The resampler's filter memory spans blocks, so the stream it sees has to be + the captured stream exactly. + """ + rf = _rf() + seen = [] + rf.run_rx_audio_dsp = lambda audio_48k: seen.append(np.array(audio_48k)) + + sizes = [700, 1300, 512, 4800, 441] + captured = [np.arange(s, dtype=np.int16).reshape(-1, 1) for s in sizes] + for block in captured: + rf.process_rx_audio_block(block) + + expected = np.concatenate([b.reshape(-1) for b in captured]) + got = np.concatenate(seen + [rf.rx_audio_carry_48k]) + np.testing.assert_array_equal(got, expected) + + def test_real_resampler_accepts_every_reblocked_block(self): + """End to end with the real codec2 resampler: odd captures, no assertion.""" + rf = _rf() + for size in (512, 1024, 441, 2048, 512, 1024, 512, 4800): + rf.process_rx_audio_block(_block(size)) # raises AssertionError if short + self.assertGreater(rf.ctx.audio_rx_queue.qsize(), 0, "DSP should have run on the reassembled blocks") + + if __name__ == "__main__": unittest.main()