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
269 lines
8.8 KiB
Python
269 lines
8.8 KiB
Python
"""Tests for the SCO (Subcarrier Oscillator) demodulator block."""
|
|
|
|
import math
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
try:
|
|
from gnuradio import blocks, gr
|
|
|
|
HAS_GNURADIO = True
|
|
except ImportError:
|
|
HAS_GNURADIO = False
|
|
|
|
from apollo.constants import (
|
|
SAMPLE_RATE_BASEBAND,
|
|
SCO_DEVIATION_PERCENT,
|
|
SCO_FREQUENCIES,
|
|
)
|
|
|
|
pytestmark = pytest.mark.skipif(not HAS_GNURADIO, reason="GNU Radio not installed")
|
|
|
|
|
|
def generate_sco_tone(
|
|
sco_number: int,
|
|
voltage: float,
|
|
n_samples: int,
|
|
sample_rate: float = SAMPLE_RATE_BASEBAND,
|
|
) -> np.ndarray:
|
|
"""Generate a synthetic SCO signal at a given sensor voltage.
|
|
|
|
Maps voltage (0-5V) linearly to frequency deviation:
|
|
0V -> center - 7.5%
|
|
2.5V -> center (nominal)
|
|
5V -> center + 7.5%
|
|
|
|
Args:
|
|
sco_number: SCO channel (1-9).
|
|
voltage: Simulated sensor input voltage (0-5V).
|
|
n_samples: Number of output samples.
|
|
sample_rate: Sample rate in Hz.
|
|
|
|
Returns:
|
|
Float array of the SCO tone at the appropriate frequency.
|
|
"""
|
|
center_freq = SCO_FREQUENCIES[sco_number]
|
|
deviation_hz = center_freq * (SCO_DEVIATION_PERCENT / 100.0)
|
|
|
|
# Map 0-5V to -deviation..+deviation
|
|
normalized = (voltage - 2.5) / 2.5 # -1.0 to +1.0
|
|
actual_freq = center_freq + normalized * deviation_hz
|
|
|
|
t = np.arange(n_samples, dtype=np.float64) / sample_rate
|
|
return np.cos(2.0 * math.pi * actual_freq * t).astype(np.float32)
|
|
|
|
|
|
class TestSCODemodInstantiation:
|
|
"""Test block creation and parameter validation."""
|
|
|
|
def test_all_channels(self):
|
|
"""Should instantiate for each valid SCO channel (1-9)."""
|
|
from apollo.sco_demod import sco_demod
|
|
|
|
for ch in range(1, 10):
|
|
demod = sco_demod(sco_number=ch)
|
|
assert demod is not None
|
|
assert demod.center_freq == SCO_FREQUENCIES[ch]
|
|
|
|
def test_invalid_channel_zero(self):
|
|
"""Channel 0 should raise ValueError."""
|
|
from apollo.sco_demod import sco_demod
|
|
|
|
with pytest.raises(ValueError, match="SCO number must be 1-9"):
|
|
sco_demod(sco_number=0)
|
|
|
|
def test_invalid_channel_ten(self):
|
|
"""Channel 10 should raise ValueError."""
|
|
from apollo.sco_demod import sco_demod
|
|
|
|
with pytest.raises(ValueError, match="SCO number must be 1-9"):
|
|
sco_demod(sco_number=10)
|
|
|
|
def test_deviation_property(self):
|
|
"""Deviation should be 7.5% of center frequency."""
|
|
from apollo.sco_demod import sco_demod
|
|
|
|
for ch in range(1, 10):
|
|
demod = sco_demod(sco_number=ch)
|
|
expected = SCO_FREQUENCIES[ch] * 0.075
|
|
assert abs(demod.deviation_hz - expected) < 0.01
|
|
|
|
def test_custom_sample_rate(self):
|
|
"""Should accept a custom sample rate."""
|
|
from apollo.sco_demod import sco_demod
|
|
|
|
demod = sco_demod(sco_number=1, sample_rate=10_240_000)
|
|
assert demod is not None
|
|
|
|
|
|
class TestSCODemodFunctional:
|
|
"""Functional tests with synthetic SCO tones."""
|
|
|
|
def test_midscale_voltage(self):
|
|
"""A 2.5V input (center frequency) should produce output near 2.5V."""
|
|
from apollo.sco_demod import sco_demod
|
|
|
|
tb = gr.top_block()
|
|
sample_rate = SAMPLE_RATE_BASEBAND
|
|
sco_ch = 5 # 52,500 Hz -- mid-range, well within Nyquist
|
|
|
|
# 200ms of signal
|
|
n_samples = int(sample_rate * 0.2)
|
|
tone = generate_sco_tone(sco_ch, voltage=2.5, n_samples=n_samples,
|
|
sample_rate=sample_rate)
|
|
|
|
src = blocks.vector_source_f(tone.tolist())
|
|
demod = sco_demod(sco_number=sco_ch, sample_rate=sample_rate)
|
|
snk = blocks.vector_sink_f()
|
|
|
|
tb.connect(src, demod, snk)
|
|
tb.run()
|
|
|
|
output = np.array(snk.data())
|
|
assert len(output) > 0, "Demodulator produced no output"
|
|
|
|
# Skip transients (first 50%), look at settled output
|
|
settled = output[len(output) // 2 :]
|
|
if len(settled) > 10:
|
|
mean_v = np.mean(settled)
|
|
# Should be near 2.5V (within 1V tolerance for FM demod settling)
|
|
assert 1.0 < mean_v < 4.0, (
|
|
f"SCO ch{sco_ch} at 2.5V input: mean output {mean_v:.2f}V, "
|
|
f"expected near 2.5V"
|
|
)
|
|
|
|
def test_low_voltage_below_midscale(self):
|
|
"""A 0V input should produce output below midscale."""
|
|
from apollo.sco_demod import sco_demod
|
|
|
|
tb = gr.top_block()
|
|
sample_rate = SAMPLE_RATE_BASEBAND
|
|
sco_ch = 5
|
|
|
|
n_samples = int(sample_rate * 0.2)
|
|
tone_low = generate_sco_tone(sco_ch, voltage=0.0, n_samples=n_samples,
|
|
sample_rate=sample_rate)
|
|
|
|
src = blocks.vector_source_f(tone_low.tolist())
|
|
demod = sco_demod(sco_number=sco_ch, sample_rate=sample_rate)
|
|
snk = blocks.vector_sink_f()
|
|
|
|
tb.connect(src, demod, snk)
|
|
tb.run()
|
|
|
|
output = np.array(snk.data())
|
|
settled = output[len(output) // 2 :]
|
|
if len(settled) > 10:
|
|
mean_v = np.mean(settled)
|
|
# Should be below 2.5V
|
|
assert mean_v < 2.5, (
|
|
f"SCO ch{sco_ch} at 0V input: mean output {mean_v:.2f}V, "
|
|
f"expected below 2.5V"
|
|
)
|
|
|
|
def test_high_voltage_above_midscale(self):
|
|
"""A 5V input should produce output above midscale."""
|
|
from apollo.sco_demod import sco_demod
|
|
|
|
tb = gr.top_block()
|
|
sample_rate = SAMPLE_RATE_BASEBAND
|
|
sco_ch = 5
|
|
|
|
n_samples = int(sample_rate * 0.2)
|
|
tone_high = generate_sco_tone(sco_ch, voltage=5.0, n_samples=n_samples,
|
|
sample_rate=sample_rate)
|
|
|
|
src = blocks.vector_source_f(tone_high.tolist())
|
|
demod = sco_demod(sco_number=sco_ch, sample_rate=sample_rate)
|
|
snk = blocks.vector_sink_f()
|
|
|
|
tb.connect(src, demod, snk)
|
|
tb.run()
|
|
|
|
output = np.array(snk.data())
|
|
settled = output[len(output) // 2 :]
|
|
if len(settled) > 10:
|
|
mean_v = np.mean(settled)
|
|
# Should be above 2.5V
|
|
assert mean_v > 2.5, (
|
|
f"SCO ch{sco_ch} at 5V input: mean output {mean_v:.2f}V, "
|
|
f"expected above 2.5V"
|
|
)
|
|
|
|
def test_monotonic_voltage_response(self):
|
|
"""Output voltage should increase monotonically with input voltage."""
|
|
from apollo.sco_demod import sco_demod
|
|
|
|
sample_rate = SAMPLE_RATE_BASEBAND
|
|
sco_ch = 6 # 70,000 Hz
|
|
n_samples = int(sample_rate * 0.2)
|
|
|
|
voltages = [0.0, 2.5, 5.0]
|
|
outputs = []
|
|
|
|
for v_in in voltages:
|
|
tb = gr.top_block()
|
|
tone = generate_sco_tone(sco_ch, voltage=v_in, n_samples=n_samples,
|
|
sample_rate=sample_rate)
|
|
src = blocks.vector_source_f(tone.tolist())
|
|
demod = sco_demod(sco_number=sco_ch, sample_rate=sample_rate)
|
|
snk = blocks.vector_sink_f()
|
|
|
|
tb.connect(src, demod, snk)
|
|
tb.run()
|
|
|
|
output = np.array(snk.data())
|
|
settled = output[len(output) // 2 :]
|
|
outputs.append(np.mean(settled) if len(settled) > 10 else float("nan"))
|
|
|
|
# Outputs should be monotonically increasing
|
|
assert outputs[0] < outputs[1] < outputs[2], (
|
|
f"Non-monotonic voltage response: "
|
|
f"V_in={voltages}, V_out={[f'{v:.2f}' for v in outputs]}"
|
|
)
|
|
|
|
def test_channel_9_highest_frequency(self):
|
|
"""SCO channel 9 (165 kHz) should still produce valid output."""
|
|
from apollo.sco_demod import sco_demod
|
|
|
|
tb = gr.top_block()
|
|
sample_rate = SAMPLE_RATE_BASEBAND
|
|
sco_ch = 9 # 165,000 Hz
|
|
|
|
n_samples = int(sample_rate * 0.2)
|
|
tone = generate_sco_tone(sco_ch, voltage=2.5, n_samples=n_samples,
|
|
sample_rate=sample_rate)
|
|
|
|
src = blocks.vector_source_f(tone.tolist())
|
|
demod = sco_demod(sco_number=sco_ch, sample_rate=sample_rate)
|
|
snk = blocks.vector_sink_f()
|
|
|
|
tb.connect(src, demod, snk)
|
|
tb.run()
|
|
|
|
output = np.array(snk.data())
|
|
assert len(output) > 0, "SCO ch9 demodulator produced no output"
|
|
|
|
def test_channel_1_lowest_frequency(self):
|
|
"""SCO channel 1 (14.5 kHz) should still produce valid output."""
|
|
from apollo.sco_demod import sco_demod
|
|
|
|
tb = gr.top_block()
|
|
sample_rate = SAMPLE_RATE_BASEBAND
|
|
sco_ch = 1 # 14,500 Hz
|
|
|
|
n_samples = int(sample_rate * 0.2)
|
|
tone = generate_sco_tone(sco_ch, voltage=2.5, n_samples=n_samples,
|
|
sample_rate=sample_rate)
|
|
|
|
src = blocks.vector_source_f(tone.tolist())
|
|
demod = sco_demod(sco_number=sco_ch, sample_rate=sample_rate)
|
|
snk = blocks.vector_sink_f()
|
|
|
|
tb.connect(src, demod, snk)
|
|
tb.run()
|
|
|
|
output = np.array(snk.data())
|
|
assert len(output) > 0, "SCO ch1 demodulator produced no output"
|