ZeroMQ Streaming

Overview

Getting radio samples out of CyberEther and into another program, or onto another computer, is the first step toward a remote receiver, a recording server, or a decoder written in another language. ZeroMQ is a messaging library that many such programs already use. It carries whole messages over TCP and offers ready-made patterns such as publish and subscribe.^[1] At 2 MS/s, the complex float32 samples the graph works with add up to 16 MB/s, or about 30 messages of 512 KiB every second.

Unlike a raw stream, this one keeps its labels. The ZeroMQ Transmitter sends a small header ahead of each buffer with its type, shape, sample rate, and center frequency, and the ZeroMQ Receiver puts them back on its output. The spectrum analyzer at the end of this flowgraph therefore shows MHz around the tuned frequency, just as if the radio fed it directly, where the UDP, TCP, and WebSocket examples can only show bins. Both blocks talk over the loopback address 127.0.0.1. Running it needs a USB radio supported by the SDR block, such as an RTL-SDR, and a free port 5555 on this machine.

How it works

The SDR block delivers buffers of 8 batches of 8,192 complex samples, 512 KiB of data every 32.8 ms, labeled with the sample rate and the tuned frequency. The ZeroMQ Transmitter, built on the libzmq library,^[2] sends each buffer as one message of two parts. The first is a JSON header of a few hundred bytes with a format version, a stream identifier, a sequence number, the data type, the shape, and the attributes. The second is the raw samples, real part then imaginary part. The library delivers all parts of a message or none of them,^[1] so a header never arrives without its samples.

On the wire, the ZeroMQ Message Transport Protocol runs over TCP and frames each part with its length.^[3] Publish sends every buffer to every subscriber and drops messages for one that falls too far behind, while Push hands each buffer to one puller in turn.^[4] The transmitter never waits, so a slow receiver can lose buffers but never stalls the graph. With Push, a buffer that no puller is ready for is counted in Dropped Buffers.

The ZeroMQ Receiver reads whole messages on a background thread into a ring buffer that holds four of its output buffers, or 131 ms of signal, and tracks which attributes came with which bytes. Each cycle copies out one full buffer with its matching sample rate, frequency, and other single-value attributes. Subscribe also checks the sequence numbers and counts any gap in Lost Messages.

What to look for

The analyzer shows the received samples as a trace above a waterfall, with the frequency axis in MHz. At the saved 96.9 MHz it runs from 95.9 to 97.9 MHz, and retuning the SDR block moves the axis with it, since the new frequency travels in the next header. Each station is a hump in the trace and a stripe in the waterfall at its real frequency, and each of the 8,192 bins is about 244 Hz wide. The trace averages about eight buffers, a quarter of a second, and the 512 waterfall rows cover about 2.1 seconds.

Throughput should read about 16 MB/s on both blocks, Peers should read 1 on each, and Buffer Loss and Lost Messages on the receiver should stay at 0.

Turning Send Attributes off shows what the header buys. The samples keep flowing, but the receiver logs that it is receiving raw buffers, and the axis falls back to bins because nothing tells the analyzer the rate or the frequency. That plain mode is what a program that reads every message part as samples expects. Setting Pattern on the receiver to Pull instead logs a warning on both ends that a peer disconnected during the handshake, because Publish only pairs with Subscribe and Push only with Pull.

For a deeper treatment of ZeroMQ sockets and patterns, see the sources below.

Going further

A few lines of Python outside CyberEther can subscribe to the same stream and read the header too, which is the starting point for a program of your own. Run the code below with Python, NumPy, and the Python bindings for ZeroMQ while the flowgraph runs. Peers on the transmitter goes to 2, and once per second the script prints the center frequency and sample rate from the header, the average power, and how many buffers the sequence numbers show as lost. At the saved settings it should read 96.900 MHz and 2.00 MS/s with 0 lost.

import json
import time

import numpy as np
import zmq

context = zmq.Context()
sock = context.socket(zmq.SUB)
sock.setsockopt(zmq.SUBSCRIBE, b"")
sock.connect("tcp://127.0.0.1:5555")

last = None
lost = 0
start = time.monotonic()

while True:
    *head, payload = sock.recv_multipart()
    header = json.loads(head[-1]) if head else {}
    x = np.frombuffer(payload, dtype=np.complex64)
    sequence = header.get("sequence")
    if last is not None and sequence is not None and sequence > last + 1:
        lost += sequence - last - 1
    last = sequence
    if time.monotonic() - start >= 1.0:
        attributes = header.get("attributes", {})
        rate = attributes.get("sampleRate", {}).get("value", 0.0)
        center = attributes.get("frequency", {}).get("value", 0.0)
        level = 10 * np.log10(np.mean(np.abs(x) ** 2) + 1e-20)
        print(f"{center / 1e6:.3f} MHz, {rate / 1e6:.2f} MS/s, {level:.1f} dBFS, {lost} lost")
        start = time.monotonic()

Batches and Samples on the receiver are independent of the radio. Setting Samples to 2,048 gives 2,048 bins over the same 2 MHz, and raising Buffer Multiplier to 16 keeps the ring at four sender buffers. Either end can listen, so a receiver on Bind can collect from several transmitters on Connect, and Push with two pullers gives each one every other buffer.

To reach another computer, set Address on the ZeroMQ Transmitter to 0.0.0.0 and point a ZeroMQ Receiver there at the address of this machine. The connection has no encryption or login, so anyone who can reach the port can read the stream.^[3]

The UDP Streaming, TCP Streaming, and WebSocket Streaming entries send the same samples without a header, the Spectrum Analyzer entry explains the display, and the block catalog covers every setting of both ZeroMQ blocks.

References