gr-apollo/src/apollo/fm_downlink_receiver.py
Ryan Malloy 7d48398551 Add FM downlink mode: carrier blocks, convenience wrappers, and loopback demo
FM mode now has the same three-layer architecture as PM mode:
- fm_mod/fm_demod for carrier-level FM modulation
- fm_signal_source/fm_downlink_receiver convenience wrappers
- fm_loopback_demo.py verifying round-trip SCO voltage recovery

Includes GRC YAML for all 4 blocks and doc updates across
blocks reference, SCO guide, and signal architecture pages.
2026-02-24 10:18:42 -07:00

95 lines
3.1 KiB
Python

"""
Apollo FM Downlink Receiver -- top-level hierarchical block for FM mode.
Combines FM carrier demodulation with per-channel SCO demodulation:
complex baseband -> fm_demod -> sco_demod(ch1) -> output[0]
-> sco_demod(ch2) -> output[1]
-> sco_demod(chN) -> output[N-1]
Input: complex baseband samples at 5.12 MHz
Output: N streaming float outputs, one per SCO channel (recovered 0-5V voltage)
Unlike usb_downlink_receiver (which outputs PDU messages), this block uses
streaming float outputs because SCO telemetry is continuous analog data,
not discrete frames.
For finer control over individual channel parameters, use fm_demod and
sco_demod blocks directly.
Reference: IMPLEMENTATION_SPEC.md sections 2.3, 4.3
"""
from gnuradio import gr
from apollo.constants import FM_CARRIER_DEVIATION_HZ, SAMPLE_RATE_BASEBAND, SCO_FREQUENCIES
from apollo.fm_demod import fm_demod
from apollo.sco_demod import sco_demod
class fm_downlink_receiver(gr.hier_block2):
"""Apollo FM downlink receiver -- complex baseband to recovered SCO voltages.
Inputs:
complex -- baseband IQ samples at sample_rate (default 5.12 MHz)
Outputs:
float[0..N-1] -- recovered sensor voltage per SCO channel (0.0 to 5.0 V)
Output ordering matches the channels list: output 0 = channels[0], etc.
"""
def __init__(
self,
channels: list[int] | None = None,
sample_rate: float = SAMPLE_RATE_BASEBAND,
carrier_pll_bw: float = 0.02,
fm_deviation_hz: float = FM_CARRIER_DEVIATION_HZ,
):
if channels is None:
channels = [1, 5, 9]
n_channels = len(channels)
gr.hier_block2.__init__(
self,
"apollo_fm_downlink_receiver",
gr.io_signature(1, 1, gr.sizeof_gr_complex),
gr.io_signature(n_channels, n_channels, gr.sizeof_float),
)
self._channels = list(channels)
self._sample_rate = sample_rate
# Validate channels
for ch in self._channels:
if ch not in SCO_FREQUENCIES:
raise ValueError(
f"SCO channel {ch} invalid. Valid: {sorted(SCO_FREQUENCIES.keys())}"
)
# Stage 1: FM carrier demodulator
self.fm = fm_demod(
carrier_pll_bw=carrier_pll_bw,
fm_deviation_hz=fm_deviation_hz,
sample_rate=sample_rate,
)
self.connect(self, self.fm)
# Stage 2: Per-channel SCO demodulators
self._sco_demods = {}
for idx, ch in enumerate(self._channels):
demod = sco_demod(sco_number=ch, sample_rate=sample_rate)
self._sco_demods[ch] = demod
# fm_demod output -> sco_demod -> hier output[idx]
self.connect(self.fm, demod, (self, idx))
@property
def channels(self) -> list[int]:
"""SCO channel numbers being decoded."""
return list(self._channels)
def get_sco_demod(self, channel: int) -> sco_demod:
"""Access a specific SCO demodulator for runtime inspection."""
return self._sco_demods[channel]