gnuradio-fm-radio-receiver/fm_radio.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# SPDX-License-Identifier: GPL-3.0
#
# GNU Radio Python Flow Graph
# Title: FM Radio Receiver

from gnuradio import analog
from gnuradio import audio
from gnuradio import filter
from gnuradio.filter import firdes
from gnuradio import gr
from gnuradio.fft import window
import sys
import signal
from argparse import ArgumentParser
from gnuradio.eng_arg import eng_float, intx
from gnuradio import eng_notation
import osmosdr
import time


class fm_radio(gr.top_block):
    def __init__(self):
        gr.top_block.__init__(self, "FM Radio Receiver", catch_exceptions=True)

        # Variables
        self.samp_rate = samp_rate = 2560000
        self.channel_width_hertz = channel_width_hertz = 250000
        self.audio_sample_rate_hertz = audio_sample_rate_hertz = 48000

        # Blocks
        self.rtlsdr_source_0 = osmosdr.source(args="numchan=" + str(1) + " " + "")
        self.rtlsdr_source_0.set_time_unknown_pps(osmosdr.time_spec_t())
        self.rtlsdr_source_0.set_sample_rate(samp_rate)
        self.rtlsdr_source_0.set_center_freq((98.3 * 1e6), 0)
        self.rtlsdr_source_0.set_freq_corr(0, 0)
        self.rtlsdr_source_0.set_dc_offset_mode(0, 0)
        self.rtlsdr_source_0.set_iq_balance_mode(0, 0)
        self.rtlsdr_source_0.set_gain_mode(False, 0)
        self.rtlsdr_source_0.set_gain(20, 0)
        self.rtlsdr_source_0.set_if_gain(20, 0)
        self.rtlsdr_source_0.set_bb_gain(20, 0)
        self.rtlsdr_source_0.set_antenna("", 0)
        self.rtlsdr_source_0.set_bandwidth(0, 0)
        self.rational_resampler_xxx_0 = filter.rational_resampler_fff(
            interpolation=audio_sample_rate_hertz,
            decimation=channel_width_hertz,
            taps=[],
            fractional_bw=0,
        )
        self.low_pass_filter_0 = filter.fir_filter_ccf(
            (int(samp_rate / channel_width_hertz)),
            firdes.low_pass(2, samp_rate, 100000, 10000, window.WIN_KAISER, 6.76),
        )
        self.audio_sink_0 = audio.sink(audio_sample_rate_hertz, "", True)
        self.analog_wfm_rcv_0 = analog.wfm_rcv(
            quad_rate=channel_width_hertz,
            audio_decimation=1,
        )

        # Connections
        self.connect((self.analog_wfm_rcv_0, 0), (self.rational_resampler_xxx_0, 0))
        self.connect((self.low_pass_filter_0, 0), (self.analog_wfm_rcv_0, 0))
        self.connect((self.rational_resampler_xxx_0, 0), (self.audio_sink_0, 0))
        self.connect((self.rtlsdr_source_0, 0), (self.low_pass_filter_0, 0))


def main(top_block_cls=fm_radio, options=None):
    tb = top_block_cls()

    def sig_handler(sig=None, frame=None):
        tb.stop()
        tb.wait()

        sys.exit(0)

    signal.signal(signal.SIGINT, sig_handler)
    signal.signal(signal.SIGTERM, sig_handler)

    tb.start()

    try:
        input("Press Enter to quit: ")
    except EOFError:
        pass
    tb.stop()
    tb.wait()


if __name__ == "__main__":
    main()