#!/usr/bin/env python3 """ Apollo USB Downlink Demo — generate and decode synthetic telemetry. Demonstrates the full gr-apollo demod chain: 1. Generate a synthetic USB baseband signal with known PCM frames 2. Feed it through usb_downlink_receiver (all-in-one block) 3. Print decoded frames as they arrive Usage: uv run python examples/usb_downlink_demo.py """ import numpy as np from gnuradio import blocks, gr from apollo.constants import SAMPLE_RATE_BASEBAND from apollo.usb_downlink_receiver import usb_downlink_receiver from apollo.usb_signal_gen import generate_usb_baseband def main(): np.random.seed(42) known_payload = bytes(np.random.randint(0, 256, size=124, dtype=np.uint8)) print("Generating 5-frame synthetic USB baseband signal...") signal, frame_bits = generate_usb_baseband( frames=5, frame_data=[known_payload] * 5, snr_db=30.0, # 30 dB SNR — moderate noise ) print(f" Signal: {len(signal)} samples, {len(signal)/SAMPLE_RATE_BASEBAND:.3f}s") print(f" Frames: {len(frame_bits)} x {len(frame_bits[0])} bits") print("\nBuilding flowgraph: usb_downlink_receiver...") tb = gr.top_block() src = blocks.vector_source_c(signal.tolist()) receiver = usb_downlink_receiver(output_format="scaled") snk = blocks.message_debug() tb.connect(src, receiver) tb.msg_connect(receiver, "frames", snk, "store") print("Running flowgraph...") tb.run() n_frames = snk.num_messages() print(f"\nReceived {n_frames} frame(s) on 'frames' port") if n_frames > 0: print("\nFirst frame metadata:") import pmt msg = snk.get_message(0) meta = pmt.car(msg) fid = pmt.to_long(pmt.dict_ref(meta, pmt.intern("frame_id"), pmt.from_long(-1))) conf = pmt.to_long(pmt.dict_ref(meta, pmt.intern("sync_confidence"), pmt.from_long(-1))) print(f" frame_id: {fid}") print(f" sync_confidence: {conf}") else: print("No frames decoded (PLL may need more settling time)") if __name__ == "__main__": main()