Ryan Malloy
0ee7ff0ad7
Complete signal processing pipeline from complex baseband to decoded PCM telemetry, verified against the 1965 NAA Study Guide (A-624): Core demod (Phase 1): - PM demodulator with carrier PLL recovery - 1.024 MHz subcarrier extractor (bandpass + downconvert) - BPSK demodulator with Costas loop + symbol sync - Convenience hier_block2 combining subcarrier + BPSK PCM frame processing (Phase 2): - 32-bit frame sync with Hamming distance correlator - SEARCH/VERIFY/LOCKED state machine, complement-on-odd handling - Frame demultiplexer (128-word, A/D voltage scaling) - AGC downlink decoder (15-bit word reassembly from DNTM1/DNTM2) Voice and analog (Phase 3): - 1.25 MHz FM voice subcarrier demod to 8 kHz audio - SCO demodulator for 9 analog sensor channels (14.5-165 kHz) Virtual AGC integration (Phase 4): - TCP bridge client with auto-reconnect and channel filtering - DSKY uplink encoder (VERB/NOUN/DATA command sequences) Top-level receiver (Phase 5): - usb_downlink_receiver hier_block2: one block, complex in, PDUs out - 14 GRC block YAML definitions for GNU Radio Companion - Example scripts for signal analysis and full-chain demo Infrastructure: - constants.py with all timing/frequency/frame parameters - protocol.py for sync word + AGC packet encode/decode - Synthetic USB signal generator for testing - 222 tests passing, ruff lint clean
67 lines
2.1 KiB
Python
67 lines
2.1 KiB
Python
"""Tests for the BPSK demodulator block."""
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
try:
|
|
from gnuradio import blocks, gr
|
|
|
|
HAS_GNURADIO = True
|
|
except ImportError:
|
|
HAS_GNURADIO = False
|
|
|
|
from apollo.constants import PCM_HIGH_BIT_RATE, SAMPLE_RATE_BASEBAND
|
|
|
|
pytestmark = pytest.mark.skipif(not HAS_GNURADIO, reason="GNU Radio not installed")
|
|
|
|
|
|
class TestBPSKDemod:
|
|
"""Test BPSK demodulation with synthetic signals."""
|
|
|
|
def test_clean_bpsk_recovery(self):
|
|
"""Known BPSK signal should recover original bits."""
|
|
from apollo.bpsk_demod import bpsk_demod
|
|
|
|
tb = gr.top_block()
|
|
sample_rate = SAMPLE_RATE_BASEBAND
|
|
symbol_rate = PCM_HIGH_BIT_RATE
|
|
sps = sample_rate / symbol_rate
|
|
n_bits = 200
|
|
|
|
# Generate known bit pattern
|
|
np.random.seed(99)
|
|
bits = np.random.randint(0, 2, n_bits)
|
|
nrz = 2.0 * bits - 1.0 # map 0→-1, 1→+1
|
|
|
|
# Upsample to sample rate (rectangular pulse shaping)
|
|
samples_per_bit = int(sps)
|
|
baseband = np.repeat(nrz, samples_per_bit).astype(np.complex64)
|
|
|
|
src = blocks.vector_source_c(baseband.tolist())
|
|
demod = bpsk_demod(
|
|
symbol_rate=symbol_rate,
|
|
sample_rate=sample_rate,
|
|
loop_bw=0.045,
|
|
)
|
|
snk = blocks.vector_sink_b()
|
|
|
|
tb.connect(src, demod, snk)
|
|
tb.run()
|
|
|
|
output = np.array(snk.data())
|
|
# Allow for sync-up time; check that most bits match after settling
|
|
if len(output) > 50:
|
|
# Find the best alignment between output and expected bits
|
|
best_match = 0
|
|
for offset in range(min(50, len(output))):
|
|
end = min(len(output) - offset, len(bits))
|
|
matches = np.sum(output[offset : offset + end] == bits[:end])
|
|
best_match = max(best_match, matches / end if end > 0 else 0)
|
|
assert best_match > 0.7, f"Bit recovery rate too low: {best_match:.2%}"
|
|
|
|
def test_block_instantiation(self):
|
|
from apollo.bpsk_demod import bpsk_demod
|
|
|
|
demod = bpsk_demod()
|
|
assert demod is not None
|