ZeroMQ-Schnittstelle

GNU Radio Companion bietet mit den ZeroMQ-Interfaces einen Satz von Blöcken, die einen Flowgraph über einen Netzwerksocket mit einem beliebigen externen Prozess verbinden — egal ob dieser ebenfalls in GNU Radio läuft oder ein völlig eigenständiges Programm ist. Das darunterliegende ZeroMQ ist eine leichtgewichtige Messaging-Bibliothek, die klassische Socket-Programmierung um fertige Kommunikationsmuster (Publish/Subscribe, Push/Pull, Request/Reply) ergänzt, ohne einen eigenen Broker-Prozess zu benötigen. Die einzelnen Blöcke sind im Block-Verzeichnis beschrieben, gruppiert nach ZMQ PUB, ZMQ SUB, ZMQ REQ, ZMQ REP und ZMQ PULL.

Im DCF77-Projekt wird genau dieses Muster genutzt: Der Flowgraph dekodiert die Zeitinformation und sendet sie per ZMQ PUB Message Sink an tcp://*:5555. Ein separates, unabhängig laufendes Python/Qt-Programm empfängt diese Nachrichten über einen SUB-Socket und zeigt die Uhrzeit grafisch an — siehe DCF77: Demodulation in GNU Radio.

1. Publish/Subscribe: das Grundmuster

Für GRC-Flowgraphs stehen zwei Varianten des PUB/SUB-Musters zur Verfügung, je nachdem, welche Art von Daten übertragen werden soll:

Das DCF77-Projekt nutzt die Message-Variante: Der Decoder-Block erzeugt keinen kontinuierlichen Datenstrom, sondern einmal pro erfolgreich dekodiertem 59-Sekunden-Rahmen eine einzelne formatierte Textnachricht. Details zu den einzelnen Blöcken (Parameter, Bind/Connect-Rollen) siehe ZMQ PUB Sink und ZMQ SUB Source im Block-Verzeichnis.

2. Parameter und Dimensionierung

3. Der externe Client: dcf77_display.py

Auf der Empfängerseite genügt eine minimale ZeroMQ-SUB-Verbindung — kein GNU-Radio-Import nötig, jede Sprache mit ZeroMQ-Bindings (Python, C++, JavaScript, …) kann als Client dienen. Das DCF77-Projekt nutzt dafür ein eigenständiges PyQt5-Programm:

import zmq
context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.connect("tcp://127.0.0.1:5555")
socket.setsockopt_string(zmq.SUBSCRIBE, "")  # alle Topics abonnieren

import zmq
context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.connect("tcp://127.0.0.1:5555")
socket.setsockopt_string(zmq.SUBSCRIBE, "")  # alle Topics abonnieren

Der leere String bei SUBSCRIBE bedeutet „alle Nachrichten empfangen, unabhängig vom Topic-Präfix" — ZeroMQ filtert PUB/SUB-Nachrichten normalerweise anhand eines Präfix-Strings (dem key-Parameter auf Senderseite), im einfachsten Fall wird dieser Mechanismus hier aber gar nicht genutzt.

Der Empfang selbst läuft nicht-blockierend in einem Qt-Timer, damit die Programmoberfläche parallel reagibel bleibt:

self.timer = QtCore.QTimer()
self.timer.timeout.connect(self.update_data)
self.timer.start(100)

def update_data(self):
    try:
        raw = self.socket.recv(zmq.NOBLOCK)
        text = raw.decode('utf-8', errors='ignore')
        p = text.split('|')  # Zeit|Bits|Datum|Zone|Schaltsek
        ...
    except zmq.Again:
        pass  # keine neue Nachricht in diesem Zyklus

self.timer = QtCore.QTimer()
self.timer.timeout.connect(self.update_data)
self.timer.start(100)

def update_data(self):
    try:
        raw = self.socket.recv(zmq.NOBLOCK)
        text = raw.decode('utf-8', errors='ignore')
        p = text.split('|')  # Zeit|Bits|Datum|Zone|Schaltsek
        ...
    except zmq.Again:
        pass  # keine neue Nachricht in diesem Zyklus

zmq.NOBLOCK sorgt dafür, dass recv() sofort mit zmq.Again abbricht, wenn gerade keine Nachricht vorliegt, statt den Qt-Event-Loop zu blockieren. Die 100 ms-Polling-Rate ist unabhängig vom timeout-Parameter des GRC-Blocks — sie bestimmt nur, wie oft die Qt-Anwendung selbst nachschaut, nicht wie GNU Radio intern wartet.

4. Stolperstein: PMT-Serialisierung bei der Message-Variante

Das erklärt eine Eigenheit im gezeigten Client-Code: raw.decode('utf-8', errors='ignore') interpretiert diese Header-Bytes (\x02, \x00, \x1f — alles gültige, aber nicht druckbare UTF-8-Codepoints) einfach mit als Text und hängt sie vor das erste Feld (p[0], die Uhrzeit). Da es sich um nicht darstellbare Steuerzeichen handelt, bleiben sie in der Qt-Anzeige unsichtbar — der Trick funktioniert hier also eher zufällig, weil das erste Feld nirgends auf exakte Gleichheit geprüft oder in eine Zahl umgewandelt wird.

Der robuste Weg, eine ZMQ-PUB-Message-Nachricht in Python zu lesen, nutzt stattdessen PMTs eigene Deserialisierung:

import pmt

raw = socket.recv(zmq.NOBLOCK)
msg = pmt.deserialize_str(raw)
text = pmt.symbol_to_string(msg)  # exakt "12:34:56|30|01.01.2026|MEZ|NEIN", ohne Header

import pmt

raw = socket.recv(zmq.NOBLOCK)
msg = pmt.deserialize_str(raw)
text = pmt.symbol_to_string(msg)  # exakt "12:34:56|30|01.01.2026|MEZ|NEIN", ohne Header

5. Weitere wichtige Aspekte

Kein GNU Radio auf der Empfängerseite nötig: Da ZeroMQ ein sprachunabhängiges Wire-Protokoll ist, kann der Client in beliebiger Sprache mit ZeroMQ-Bindings geschrieben werden — die Stream-Varianten übertragen dabei rohe Bytes im jeweiligen Item-Format (z. B. complex64 für IQ-Samples), die Message-Varianten das oben gezeigte PMT-Binärformat.

Mehrere Subscriber gleichzeitig: PUB/SUB ist ein 1-zu-n-Muster — beliebig viele SUB-Clients können sich gleichzeitig mit demselben PUB-Endpoint verbinden und erhalten alle dieselben Nachrichten, ohne dass der Flowgraph davon weiß oder zusätzliche Last durch weitere Zuhörer entsteht.

Kein Nachliefern verpasster Nachrichten: ZeroMQ PUB/SUB puffert nicht rückwirkend — ein Subscriber, der sich erst nach dem Senden verbindet, hat die vorherige Nachricht endgültig verpasst. Für das DCF77-Beispiel unproblematisch (die nächste Minute liefert ohnehin eine neue Nachricht), für Anwendungen mit seltenen, nicht wiederholbaren Ereignissen ggf. relevant.

dcf77-zeromq-display.zip — der komplette, hier gezeigte PyQt5-Client (dcf77_display.py), lauffähig gegen den ZMQ-PUB-Sink des DCF77-Flowgraphs.