Your first GNU Radio flowgraph

So far every lesson has processed whole numpy arrays at once. A radio does not work like that. Samples arrive continuously from an ADC, and the processing must keep up, block by block, for as long as the radio runs. GNU Radio is an open-source software-radio framework built around that streaming model. This lesson is for readers who know the numpy links of lessons 6 to 8 and want to build the same kind of link in GNU Radio. We write the flowgraph in Python, run it without any graphical interface, count bit errors, and then see how the same graph looks in GNU Radio Companion. Everything here was run on GNU Radio 3.10.12.

Blocks, streams and the top block

A GNU Radio program is a flowgraph: a set of blocks connected by streams. Each block does one job, such as filtering, adding noise or mapping bits to symbols. A stream carries items of one type from an output port to an input port. The common item types are bytes, 32-bit floats, and complex values made of two 32-bit floats, which GNU Radio calls gr_complex and numpy calls complex64. Block names end in a suffix that gives the types: _bc takes bytes and produces complex, _ccf takes and produces complex with float filter taps, _cc is complex in and out.

The container is a gr.top_block. You create blocks, connect them with connect, and call run(). The scheduler passes buffers of items from block to block. According to the GNU Radio API reference, run() starts the graph and waits for it to finish, and a flowgraph finishes when all its blocks report that they are done, or when something calls stop(). For a simulation that means the graph needs a source that ends. A blocks.vector_source_* with repeat=False ends after its data, and a blocks.head passes a fixed number of items and then stops. Anything downstream of a finished source finishes too. A blocks.vector_sink_* collects whatever reaches it so you can read it as a list afterwards.

Most blocks fall into three groups by rate. A sync block produces one output item per input item, like an adder. A decimator produces one output per D inputs. An interpolator produces I outputs per input. A stream has no sample-rate setting of its own. The rate is a number you track yourself, and in a pure simulation the graph runs as fast as the CPU allows. Only a hardware source or sink, tied to a real sample clock, sets the pace in time.

The link we will build

The flowgraph sends random bits through a QPSK modulator, a root-raised-cosine (RRC) pulse-shaping filter, additive white Gaussian noise, a matched RRC filter and a decision device, and back to bits. It is the lesson 6 link with QPSK instead of BPSK, built from GNU Radio blocks.

Each block in the Python code has a counterpart in GNU Radio Companion (GRC), the graphical editor, named in the diagram. The rates are nominal. With a symbol rate of 125 kbaud and 8 samples per symbol the filtered stream represents 1 MHz of complex bandwidth, but the simulation itself runs at whatever speed your CPU manages.

The flowgraph in Python

The block below is complete. The first line, # gnuradio, marks it as a flowgraph for our code checker. The noise level follows the same unit-energy convention as lesson 6. The RRC taps are scaled so their squares sum to 1, so each QPSK symbol carries Es = 1 and each bit Eb = 0.5. The complex noise variance per sample is then N0, and analog.noise_source_c takes its standard deviation as the amplitude.

# gnuradio
import numpy as np
from scipy.special import erfc
from gnuradio import gr, blocks, analog, digital, filter
from gnuradio.filter import firdes

sps = 8                     # samples per symbol
sym_rate = 125e3            # symbol rate, so the sample rate is 1 MHz
samp_rate = sps * sym_rate
n_bits = 200_000
ebn0_db = 6.0

# Gray-mapped QPSK: symbol index -> point. Indices 0..3 carry two bits each.
points = [(1 + 1j) / np.sqrt(2), (-1 + 1j) / np.sqrt(2), (1 - 1j) / np.sqrt(2), (-1 - 1j) / np.sqrt(2)]
const = digital.constellation_calcdist(points, [], 4, 1, digital.constellation.NO_NORMALIZATION)

# Root-raised-cosine taps, roll-off 0.35, 11 symbols long, scaled to unit energy.
ntaps = 11 * sps + 1
rrc = np.array(firdes.root_raised_cosine(1.0, samp_rate, sym_rate, 0.35, ntaps))
rrc /= np.sqrt(np.sum(rrc**2))
delay_sym = (ntaps - 1) // sps          # two filters, each (ntaps - 1) / 2 samples, = 11 symbols

# With unit-energy pulses each symbol carries Es = 1, so Eb = 0.5. Complex noise variance per sample = N0.
n0 = 0.5 / 10 ** (ebn0_db / 10)

rng = np.random.default_rng(9)
bits = rng.integers(0, 2, n_bits).astype(np.uint8)

tb = gr.top_block()
src = blocks.vector_source_b(bits.tolist(), False)          # one bit per byte, finite, so the graph ends
pack = blocks.pack_k_bits_bb(2)                             # 2 bits -> one symbol index 0..3
mapper = digital.chunks_to_symbols_bc(points)               # index -> complex point
tx_filt = filter.interp_fir_filter_ccf(sps, rrc.tolist())   # upsample by 8 and pulse-shape
noise = analog.noise_source_c(analog.GR_GAUSSIAN, float(np.sqrt(n0)), 42)
add = blocks.add_cc()                                       # channel: signal + noise
rx_filt = filter.fir_filter_ccf(sps, rrc.tolist())          # matched filter, keep 1 sample per symbol
decoder = digital.constellation_decoder_cb(const.base())    # nearest point -> index 0..3
unpack = blocks.unpack_k_bits_bb(2)                         # index -> 2 bits
sym_sink = blocks.vector_sink_c()                           # matched-filter output, for the SNR estimate
bit_sink = blocks.vector_sink_b()

tb.connect(src, pack, mapper, tx_filt)
tb.connect(tx_filt, (add, 0))
tb.connect(noise, (add, 1))
tb.connect(add, rx_filt, decoder, unpack, bit_sink)
tb.connect(rx_filt, sym_sink)
tb.run()

rx_bits = np.array(bit_sink.data(), dtype=np.uint8)
rx_sym = np.array(sym_sink.data())
skip = 2 * delay_sym                                       # filter delay in bits (2 bits per symbol)
n = min(rx_bits.size - skip, bits.size)
errors = int(np.sum(rx_bits[skip : skip + n] != bits[:n]))
ber = errors / n
ber_theory = 0.5 * erfc(np.sqrt(10 ** (ebn0_db / 10)))

tx_sym = np.array(points)[bits[: 2 * (n // 2)].reshape(-1, 2) @ [2, 1]]
err_vec = rx_sym[delay_sym : delay_sym + tx_sym.size] - tx_sym
snr_db = 10 * np.log10(1 / np.mean(np.abs(err_vec) ** 2))

print(f"sample rate {samp_rate / 1e6:.1f} MHz, symbol rate {sym_rate / 1e3:.0f} kbaud, filter delay {delay_sym} symbols")
print(f"{n} bits compared, {errors} errors")
print(f"BER {ber:.2e} at Eb/N0 = {ebn0_db} dB (theory {ber_theory:.2e})")
print(f"measured Es/N0 after the matched filter: {snr_db:.2f} dB (expected {ebn0_db + 10 * np.log10(2):.2f} dB)")

sample rate 1.0 MHz, symbol rate 125 kbaud, filter delay 11 symbols
199978 bits compared, 474 errors
BER 2.37e-03 at Eb/N0 = 6.0 dB (theory 2.39e-03)
measured Es/N0 after the matched filter: 9.01 dB (expected 9.01 dB)

The bit error rate is 2.37 × 10⁻³ at an Eb/N0 of 6 dB, against 2.39 × 10⁻³ from theory. With 474 errors counted, that difference is well within statistical spread. The symbol SNR measured after the matched filter is 9.01 dB, exactly the 6 dB Eb/N0 plus 3.01 dB for two bits per symbol. That check matters more than the BER. It confirms that the noise amplitude, the filter scaling and the sampling instant are all what we intended.

Four details in the code are worth reading closely.

The delay. Each RRC filter has 89 taps and delays the signal by 44 samples. Two of them delay it by 88 samples, which is exactly 11 symbols at 8 samples per symbol. Because the total is a whole number of symbols, the decimating filter’s output lands on the symbol peaks with no extra alignment. The received bit stream starts 22 bits late, so the code skips 22 bits before comparing. If you change the number of taps so that the total delay is not a multiple of 8, the matched filter samples between peaks and the BER rises.

The constellation. digital.chunks_to_symbols_bc maps symbol index i to the i-th point of a list. digital.constellation_decoder_cb does the reverse: it returns the index of the nearest point. We build the decoder’s constellation with digital.constellation_calcdist from the same list, with normalisation switched off so GNU Radio does not rescale our points. The order of the list is the bit mapping, here Gray coded so that neighbouring points differ in one bit.

Bits and symbols. blocks.pack_k_bits_bb(2) turns two one-bit bytes into one byte holding a value from 0 to 3, first bit most significant. blocks.unpack_k_bits_bb(2) undoes it after the decoder. The rate halves and doubles accordingly.

Termination. The noise source never ends, but the adder cannot produce output once the filtered signal stops arriving. When the vector source runs out, every block downstream of it finishes, and run() returns.

A BER curve from a class

GRC-generated code puts the flowgraph in a class derived from gr.top_block, with the blocks as attributes. The next block does the same by hand and sweeps Eb/N0 from 0 dB to 10 dB, building a fresh graph for each point. It uses 200 000 bits up to 7 dB and more bits above that, where errors are rare.

# gnuradio
import numpy as np
from scipy.special import erfc
from gnuradio import gr, blocks, analog, digital, filter
from gnuradio.filter import firdes

SPS = 8
POINTS = [(1 + 1j) / np.sqrt(2), (-1 + 1j) / np.sqrt(2), (1 - 1j) / np.sqrt(2), (-1 - 1j) / np.sqrt(2)]
NTAPS = 11 * SPS + 1
RRC = np.array(firdes.root_raised_cosine(1.0, SPS, 1.0, 0.35, NTAPS))
RRC /= np.sqrt(np.sum(RRC**2))


class QpskLink(gr.top_block):
    """Bits -> QPSK -> RRC -> AWGN -> matched filter -> decisions -> bits, all headless."""

    def __init__(self, bits: np.ndarray, ebn0_db: float, seed: int):
        gr.top_block.__init__(self, "QPSK link")
        n0 = 0.5 / 10 ** (ebn0_db / 10)                       # Es = 1, Eb = 0.5
        const = digital.constellation_calcdist(POINTS, [], 4, 1, digital.constellation.NO_NORMALIZATION)
        self.src = blocks.vector_source_b(bits.tolist(), False)
        self.pack = blocks.pack_k_bits_bb(2)
        self.mapper = digital.chunks_to_symbols_bc(POINTS)
        self.tx_filt = filter.interp_fir_filter_ccf(SPS, RRC.tolist())
        self.noise = analog.noise_source_c(analog.GR_GAUSSIAN, float(np.sqrt(n0)), seed)
        self.add = blocks.add_cc()
        self.rx_filt = filter.fir_filter_ccf(SPS, RRC.tolist())
        self.decoder = digital.constellation_decoder_cb(const.base())
        self.unpack = blocks.unpack_k_bits_bb(2)
        self.sink = blocks.vector_sink_b()
        self.connect(self.src, self.pack, self.mapper, self.tx_filt, (self.add, 0))
        self.connect(self.noise, (self.add, 1))
        self.connect(self.add, self.rx_filt, self.decoder, self.unpack, self.sink)


rng = np.random.default_rng(9)
skip = 2 * (NTAPS - 1) // SPS                                  # filter delay in bits
print("Eb/N0 dB, bits, errors, BER sim, BER theory")
for k, ebn0_db in enumerate(range(0, 11)):
    n_bits = 200_000 if ebn0_db < 8 else 2_000_000 if ebn0_db < 10 else 8_000_000   # more bits where errors are rare
    bits = rng.integers(0, 2, n_bits).astype(np.uint8)
    tb = QpskLink(bits, ebn0_db, seed=100 + k)
    tb.run()
    rx = np.array(tb.sink.data(), dtype=np.uint8)
    n = min(rx.size - skip, n_bits)
    errors = int(np.sum(rx[skip : skip + n] != bits[:n]))
    theory = 0.5 * erfc(np.sqrt(10 ** (ebn0_db / 10)))
    print(f"{ebn0_db}, {n}, {errors}, {errors / n:.2e}, {theory:.2e}")

Eb/N0 dB, bits, errors, BER sim, BER theory
0, 199978, 15852, 7.93e-02, 7.86e-02
1, 199978, 11152, 5.58e-02, 5.63e-02
2, 199978, 7554, 3.78e-02, 3.75e-02
3, 199978, 4517, 2.26e-02, 2.29e-02
4, 199978, 2495, 1.25e-02, 1.25e-02
5, 199978, 1177, 5.89e-03, 5.95e-03
6, 199978, 520, 2.60e-03, 2.39e-03
7, 199978, 162, 8.10e-04, 7.73e-04
8, 1999978, 376, 1.88e-04, 1.91e-04
9, 1999978, 75, 3.75e-05, 3.36e-05
10, 7999978, 29, 3.63e-06, 3.87e-06

The simulation follows theory across the range. At 10 dB, 8 million bits gave 29 errors, a BER of 3.63 × 10⁻⁶ against 3.87 × 10⁻⁶. Rare-event points like this need many bits: with 100 errors counted, the relative standard deviation of the estimate is about 1/√100 = 10 %, so aim for at least that many.

Probe a block before you trust it

GNU Radio’s channels.channel_model bundles noise, a frequency offset, a timing offset (epsilon) and a multipath FIR (taps) into one block, and the GNU Radio API reference describes it as a basic channel simulator. It is convenient, but it does more than add noise. Before putting an unfamiliar block into a BER test, feed it an impulse and feed it silence.

# gnuradio
import numpy as np
from gnuradio import gr, blocks, channels

# Probe channels.channel_model with an impulse and with silence before trusting it in a link.
x = np.zeros(64, dtype=np.complex64)
x[10] = 1.0
tb = gr.top_block()
src = blocks.vector_source_c(x.tolist(), False)
chan = channels.channel_model(noise_voltage=0.0, frequency_offset=0.0, epsilon=1.0, taps=[1.0], noise_seed=0)
sink = blocks.vector_sink_c()
tb.connect(src, chan, sink)
tb.run()
y = np.array(sink.data())
print(f"in: {x.size} samples, impulse at index 10; out: {y.size} samples, peak at index {int(np.argmax(np.abs(y)))}")

tb = gr.top_block()
src = blocks.vector_source_c([0j] * 100_000, False)
chan = channels.channel_model(noise_voltage=0.5, frequency_offset=0.0, epsilon=1.0, taps=[1.0], noise_seed=7)
sink = blocks.vector_sink_c()
tb.connect(src, chan, sink)
tb.run()
w = np.array(sink.data())
print(f"noise_voltage 0.5: complex noise variance {np.var(w):.3f}, per rail {np.var(w.real):.3f}")

in: 64 samples, impulse at index 10; out: 56 samples, peak at index 7
noise_voltage 0.5: complex noise variance 0.248, per rail 0.126

Two things show up. In GNU Radio 3.10.12 the block returned 56 samples for 64 in, with the impulse moved from index 10 to index 7. Its internal resampler, which implements the timing offset, shifts the stream and holds back a few samples. In our filtered link that 3-sample shift would move the matched filter off the symbol peaks, so you would need to realign the stream, for example with a blocks.delay of 3 samples to cancel the shift. Second, noise_voltage is the standard deviation of the complex noise: 0.5 gave a total variance of 0.248, split evenly between I and Q. The first flowgraph used analog.noise_source_c and blocks.add_cc instead, because that keeps the timing exact and makes the noise level explicit.

The same graph in GNU Radio Companion

GRC is a drag-and-drop editor that saves a flowgraph as a .grc file, which is YAML. Each block in the palette is described by a .block.yml file shipped with GNU Radio. That file holds the block’s label, its parameters and a make template: the Python call GRC writes into the generated script. In GNU Radio 3.10.12 the Channel Model block’s template produces channels.channel_model(...), and the Add block produces blocks.add_vcc(1), the vector form of the adder with vector length 1.

Python call in this lesson	GRC block label
blocks.vector_source_b	Vector Source
blocks.pack_k_bits_bb	Pack K Bits
digital.chunks_to_symbols_bc	Chunks to Symbols
filter.interp_fir_filter_ccf	Interpolating FIR Filter
firdes.root_raised_cosine	RRC Filter Taps (a variable block)
analog.noise_source_c	Noise Source
blocks.add_cc	Add
filter.fir_filter_ccf	Decimating FIR Filter
digital.constellation_decoder_cb	Constellation Decoder
blocks.unpack_k_bits_bb	Unpack K Bits
blocks.vector_sink_b	Vector Sink
channels.channel_model	Channel Model

Every GRC flowgraph has an Options block. Its generate_options setting chooses what kind of script GRC writes: in the example flowgraphs shipped with GNU Radio 3.10.12 the values are qt_gui for a graphical application, no_gui for a headless script and hb for a hierarchical block. Variables such as samp_rate become attributes of the generated class. A flowgraph built in GRC for a lab bench usually ends in GUI sinks, a spectrum or constellation display. For automated tests, keep those out and end in vector sinks as we did, or generate with no_gui.

You will also see a Throttle block in many GRC simulations. The API reference says it is intended only for GUI applications that have no other rate-limiting block, and that it does not control the rate precisely. A headless test that should finish quickly leaves it out. A flowgraph that talks to an SDR never needs it, because the radio’s sample clock sets the pace. The next lesson records samples from such a radio and stores them in a standard file format.

Exercise

1. Change ntaps to 11 * sps (an even count). Explain what happens to the delay and the BER, and fix the alignment without changing the taps.
2. Replace the noise source and adder in the first block with channels.channel_model, add blocks.delay(gr.sizeof_gr_complex, 3) after it, and confirm that the BER returns to the theoretical value.
3. Give channels.channel_model a normalised frequency_offset of 0.001. Collect the matched-filter output with a vector sink and describe what happens to the constellation over time.

Summary

• A GNU Radio flowgraph is blocks connected by typed streams inside a gr.top_block. run() returns once every block is done, so a simulation needs a finite source or a blocks.head.
• Sample rates are bookkeeping. Interpolators and decimators change them, and in simulation the graph runs as fast as the CPU allows.
• Our QPSK link measured 2.37 × 10⁻³ at 6 dB Eb/N0 against 2.39 × 10⁻³ in theory, and the measured symbol SNR matched the setting to 0.01 dB.
• Probe unfamiliar blocks with an impulse and with silence. channels.channel_model shifted our stream by 3 samples and scales noise by its complex standard deviation.
• GRC writes Python from per-block YAML templates, so every GRC block has a Python call you can use directly.

Further reading

• GNU Radio Manual and C++ API Reference: the official reference for every block, including gr::top_block and gr::channels::channel_model. At the time of writing the hosted manual is built from GNU Radio 3.10.9.1.
• GNU Radio v3.10.12.0 release: the release used for every block on this page.
• J. G. Proakis and M. Salehi, Digital Communications, 5th edition, McGraw-Hill, 2008: see the chapter “Optimum Receivers for AWGN Channels” for the QPSK error rate the simulation is checked against.