Testen mit GRC

Dieses Dokument zeigt, was hinter der grafischen GRC-Oberfläche eigentlich passiert, wie man Blöcke ohne Bildschirm und Mausklicks ausführt, und wie man sich dafür passende Testsignale selbst baut — anhand eines eigenen, einfachen Beispielblocks, ohne Bezug zu einem bestimmten Funkprojekt.

• Voraussetzung: Python-Grundkenntnisse, numpy hilfreich
• Nicht vorausgesetzt: GRC-Erfahrung
• Werkzeuge: GNU Radio (Python-API), grcc

Was GRC eigentlich tut

Die grafische Oberfläche von GNU Radio Companion sieht aus wie ein eigenständiges Programm. Tatsächlich ist sie nur ein Editor für eine Textdatei — und genau das macht sie testbar.

Eine .grc-Datei ist eine YAML-Beschreibung von Blöcken und Verbindungen. Der Compiler grcc liest diese Datei und erzeugt daraus ein ganz gewöhnliches Python-Skript: jeder Block im Diagramm wird zu einer Zeile, die ein Objekt erzeugt, jede Verbindung (jedes „Kabel" im Bild) wird zu einem Aufruf von .connect(...). Ein Diagramm mit drei verbundenen Blöcken erzeugt ungefähr das hier:

# von grcc erzeugt, gekürzt
self.quelle = analog.sig_source_f(samp_rate, ...)
self.verstaerker = blocks.multiply_const_ff(2.0)
self.anzeige = qtgui.time_sink_f(...)

self.connect((self.quelle, 0), (self.verstaerker, 0))
self.connect((self.verstaerker, 0), (self.anzeige, 0))

# von grcc erzeugt, gekürzt
self.quelle = analog.sig_source_f(samp_rate, ...)
self.verstaerker = blocks.multiply_const_ff(2.0)
self.anzeige = qtgui.time_sink_f(...)

self.connect((self.quelle, 0), (self.verstaerker, 0))
self.connect((self.verstaerker, 0), (self.anzeige, 0))

Die Konsequenz: Alles, was in GRC per Maus zusammengeklickt wird, lässt sich genauso gut — und für Tests viel besser — direkt in Python schreiben. Man braucht die grafische Oberfläche nicht, um einen Flowgraph zu bauen oder auszuführen, nur um ihn zu zeichnen.

Werkzeuge einrichten

Drei kurze Prüfungen, bevor es losgeht — jede sollte ohne Fehlermeldung durchlaufen.

# 1) Lässt sich GNU Radio aus Python heraus ansprechen?
python3 -c "from gnuradio import gr; print(gr.version())"

# 2) Ist der GRC-Compiler vorhanden?
grcc --help

# 3) Funktioniert numpy im selben Python?
python3 -c "import numpy; print(numpy.__version__)"

# 1) Lässt sich GNU Radio aus Python heraus ansprechen?
python3 -c "from gnuradio import gr; print(gr.version())"

# 2) Ist der GRC-Compiler vorhanden?
grcc --help

# 3) Funktioniert numpy im selben Python?
python3 -c "import numpy; print(numpy.__version__)"

Für den Einstieg reicht eine einzige Installation über eine der gängigen GNU-Radio-Distributionen (z. B. radioconda, oder das Paket der eigenen Linux-Distribution). Empfohlene Ordnerstruktur für eigene Projekte, damit generierter und handgeschriebener Code sich nicht vermischen:

Der minimale Baustein: ein Flowgraph ohne Bildschirm

Für Tests wird nie die grafische Anzeige gebraucht, sondern immer dieselben vier Bausteine.

A. Der Container: gr.top_block()

Dasselbe Objekt, das auch eine kompilierte .grc-Datei im Hintergrund erzeugt — nur von Hand instanziiert statt gezeichnet. gr.top_block() ist dabei mehr als nur ein passiver Behälter für Blöcke: Es ist die Laufzeitumgebung des gesamten Flowgraphs und übernimmt drei Aufgaben gleichzeitig:

• Verbindungsverwaltung: tb.connect(...) registriert, welcher Ausgang mit welchem Eingang verdrahtet ist — das ist die programmatische Entsprechung der „Kabel" im GRC-Diagramm.
• Scheduler: Beim Start (run()/start()) legt der Top-Block für jeden angeschlossenen Block einen eigenen Thread an und alloziert die Puffer dazwischen. Er entscheidet laufend, wann welcher Block genug Eingabedaten hat und genug Platz im Ausgabepuffer, um seine work()-Funktion aufzurufen — Details dazu (Buffer, Flow Control, Backpressure) stehen im Abschnitt „Scheduler" in Daten in GRC.
• Lebenszyklus-Steuerung: run(), start(), stop() und wait() sind Methoden des Top-Blocks — er startet und stoppt alle Block-Threads koordiniert, nicht die einzelnen Blöcke selbst.

Für einfache Tests reicht es, sich den Top-Block als „das Objekt, das man ausführt" zu merken — die Scheduler-Mechanik dahinter wird erst relevant, wenn ein Flowgraph nicht wie erwartet terminiert oder Daten verliert.

from gnuradio import gr
tb = gr.top_block()

from gnuradio import gr
tb = gr.top_block()

B. Quellen ohne Antenne, Senken ohne Bildschirm

Für Tests werden reale Quellen (Antenne, Soundkarte) durch vector_source ersetzt — ein fertiges numpy-Array wird als Stream eingespeist. Auf der Gegenseite sammelt vector_sink die Ausgabe (inklusive aller Tags) einfach in einem Python-Array.

from gnuradio import blocks
import numpy as np

testdaten = np.array([...], dtype=np.float32)
quelle = blocks.vector_source_f(testdaten.tolist(), repeat=False)
senke = blocks.vector_sink_f()

from gnuradio import blocks
import numpy as np

testdaten = np.array([...], dtype=np.float32)
quelle = blocks.vector_source_f(testdaten.tolist(), repeat=False)
senke = blocks.vector_sink_f()

Für komplexe Signale (I/Q) gibt es dieselben Blöcke mit _c statt _f: vector_source_c, vector_sink_c.

Eigene Signalformen bauen

Das testdaten-Array oben ist bewusst leer gelassen — der eigentliche Reiz von vector_source liegt darin, dass man mit reinem numpy bauen kann, was immer man testen will, statt auf die in GRC per Klick verfügbaren Signalquellen (Sinus, Rechteck, Sägezahn, Rauschen) beschränkt zu sein. Der gemeinsame Nenner aller Signalformen ist eine Zeitachse:

samp_rate = 48000
dauer_s = 0.1
t = np.arange(int(samp_rate * dauer_s)) / samp_rate   # Zeitachse in Sekunden

samp_rate = 48000
dauer_s = 0.1
t = np.arange(int(samp_rate * dauer_s)) / samp_rate   # Zeitachse in Sekunden

Darauf aufbauend die gängigen Grundformen:

# Sinuston (reell, fuer *_f-Bloecke)
freq = 1000
sinus = np.sin(2 * np.pi * freq * t).astype(np.float32)

# Komplexer Sinuston / IQ-Signal (fuer *_c-Bloecke)
# -- entspricht dem, was ein SDR-Empfaenger als Basisbandsignal liefert
iq_sinus = np.exp(1j * 2 * np.pi * freq * t).astype(np.complex64)

# Rechteck (z.B. fuer FSK/ASK-Tests): Vorzeichen eines Sinus
rechteck = np.sign(np.sin(2 * np.pi * freq * t)).astype(np.float32)

# Weißes Rauschen mit definierter Standardabweichung
rng = np.random.default_rng(42)
rauschen = rng.normal(0, 0.1, len(t)).astype(np.float32)

# Sinuston (reell, fuer *_f-Bloecke)
freq = 1000
sinus = np.sin(2 * np.pi * freq * t).astype(np.float32)

# Komplexer Sinuston / IQ-Signal (fuer *_c-Bloecke)
# -- entspricht dem, was ein SDR-Empfaenger als Basisbandsignal liefert
iq_sinus = np.exp(1j * 2 * np.pi * freq * t).astype(np.complex64)

# Rechteck (z.B. fuer FSK/ASK-Tests): Vorzeichen eines Sinus
rechteck = np.sign(np.sin(2 * np.pi * freq * t)).astype(np.float32)

# Weißes Rauschen mit definierter Standardabweichung
rng = np.random.default_rng(42)
rauschen = rng.normal(0, 0.1, len(t)).astype(np.float32)

Komplexere Testsignale entstehen durch Kombinieren dieser Bausteine — Addition für Überlagerung (Signal + Rauschen), np.concatenate([...]) für Abschnitte, die zeitlich aufeinanderfolgen (z. B. Stille, dann Ton, dann wieder Stille), und Multiplikation für Modulation (z. B. ein Rechtecksignal, das die Amplitude eines Trägers moduliert, für einen ASK-Test):

# Drei Abschnitte hintereinander: Stille - Ton - Stille
stille = np.zeros(int(0.01 * samp_rate), dtype=np.float32)
ton = np.sin(2 * np.pi * freq * np.arange(int(0.02 * samp_rate)) / samp_rate).astype(np.float32)
testsignal = np.concatenate([stille, ton, stille])

# ASK: Rechteck-Hüllkurve moduliert einen Träger
traeger = np.sin(2 * np.pi * 5000 * t).astype(np.float32)
huellkurve = (np.sign(np.sin(2 * np.pi * 50 * t)) + 1) / 2   # 0/1 statt -1/1
ask_signal = (traeger * huellkurve).astype(np.float32)

# Drei Abschnitte hintereinander: Stille - Ton - Stille
stille = np.zeros(int(0.01 * samp_rate), dtype=np.float32)
ton = np.sin(2 * np.pi * freq * np.arange(int(0.02 * samp_rate)) / samp_rate).astype(np.float32)
testsignal = np.concatenate([stille, ton, stille])

# ASK: Rechteck-Hüllkurve moduliert einen Träger
traeger = np.sin(2 * np.pi * 5000 * t).astype(np.float32)
huellkurve = (np.sign(np.sin(2 * np.pi * 50 * t)) + 1) / 2   # 0/1 statt -1/1
ask_signal = (traeger * huellkurve).astype(np.float32)

C. Verbinden und ausführen

Drei Variablen kommen hier zusammen, die vorher an drei verschiedenen Stellen entstanden sind: quelle und senke sind die vector_source-/vector_sink-Instanzen aus Abschnitt B, mein_block ist eine Instanz des Blocks, den man eigentlich testen will (weiter unten am Beispiel des Pegelwächters, z. B. pegelwaechter_block.blk(low=0.3, high=0.7)) — der top_block selbst kennt zu diesem Zeitpunkt keinen davon, bis connect() sie miteinander bekannt macht.

tb.connect(quelle, mein_block, senke) ist eine Kurzschreibweise: Werden mehr als zwei Blöcke auf einmal übergeben, verkettet GNU Radio sie automatisch der Reihe nach — Ausgang 0 von quelle an Eingang 0 von mein_block, Ausgang 0 von mein_block an Eingang 0 von senke. Das entspricht genau den beiden „Kabeln", die man in GRC von Hand ziehen würde:

tb.connect((quelle, 0), (mein_block, 0))
tb.connect((mein_block, 0), (senke, 0))

tb.connect((quelle, 0), (mein_block, 0))
tb.connect((mein_block, 0), (senke, 0))

Diese Kurzform funktioniert nur, solange jeder Block genau einen Ein- und Ausgang hat (Port-Index 0). Sobald ein Block mehrere Ein- oder Ausgänge besitzt (z. B. ein Block mit separaten I/Q-Ausgängen oder ein Message-Port), muss man auf die explizite (block, port)-Schreibweise zurückgreifen, um festzulegen, welcher Port gemeint ist.

tb.connect(quelle, mein_block, senke)

# repeat=False an der Quelle -> run() kehrt automatisch zurück,
# sobald das Test-Array einmal komplett verarbeitet wurde
tb.run()

ergebnis = np.array(senke.data())

tb.connect(quelle, mein_block, senke)

# repeat=False an der Quelle -> run() kehrt automatisch zurück,
# sobald das Test-Array einmal komplett verarbeitet wurde
tb.run()

ergebnis = np.array(senke.data())

Für endliche Testsignale (das ist praktisch immer der Fall) reicht tb.run() — es blockiert, bis die Quelle erschöpft ist, und kehrt dann zurück. Nur für dauerhaft laufende Flowgraphs (z. B. mit einer echten, nie endenden Signalquelle) braucht man stattdessen tb.start(), eine Wartezeit, dann tb.stop(); tb.wait().

Eigenes Beispiel: ein Pegelwächter als Custom-Block

Die meisten eigenen GRC-Projekte enthalten irgendwann einen „Embedded Python Block" — einen selbst geschriebenen Baustein für Logik, die es fertig nicht gibt. In der GRC-Oberfläche fügt man diesen Blocktyp per Doppelklick ein und schreibt in einem Textfeld eine Python-Klasse hinein; GRC speichert diesen Code als Teil der .grc-Datei und exportiert ihn beim Kompilieren als eigene .py-Datei (Namensschema: <projektname>_epy_block_N.py).

Als durchgehendes Beispiel für den Rest dieses Dokuments: ein Pegelwächter — ein generischer Schmitt-Trigger, der aus einem reellen Signal einen 0/1-Logikpegel mit Hysterese macht (unabhängig von jedem konkreten Funkprojekt, aber strukturell typisch: er hat einen Zustand, der über mehrere Aufrufe hinweg erhalten bleibt).

import numpy as np
from gnuradio import gr

class blk(gr.sync_block):
    """Pegelwaechter mit Hysterese (Schmitt-Trigger).
    Gibt 1.0 aus, sobald der Eingang die obere Schwelle ueberschreitet,
    und bleibt bei 1.0, bis der Eingang unter die untere Schwelle faellt."""

    def __init__(self, low=0.3, high=0.7):
        gr.sync_block.__init__(
            self,
            name="Pegelwaechter",
            in_sig=[np.float32],
            out_sig=[np.float32]
        )
        self.low = low
        self.high = high
        self.state = 0.0

    def work(self, input_items, output_items):
        x = input_items[0]
        out = output_items[0]
        for i in range(len(x)):
            if x[i] >= self.high:
                self.state = 1.0
            elif x[i] <= self.low:
                self.state = 0.0
            out[i] = self.state
        return len(out)

import numpy as np
from gnuradio import gr

class blk(gr.sync_block):
    """Pegelwaechter mit Hysterese (Schmitt-Trigger).
    Gibt 1.0 aus, sobald der Eingang die obere Schwelle ueberschreitet,
    und bleibt bei 1.0, bis der Eingang unter die untere Schwelle faellt."""

    def __init__(self, low=0.3, high=0.7):
        gr.sync_block.__init__(
            self,
            name="Pegelwaechter",
            in_sig=[np.float32],
            out_sig=[np.float32]
        )
        self.low = low
        self.high = high
        self.state = 0.0

    def work(self, input_items, output_items):
        x = input_items[0]
        out = output_items[0]
        for i in range(len(x)):
            if x[i] >= self.high:
                self.state = 1.0
            elif x[i] <= self.low:
                self.state = 0.0
            out[i] = self.state
        return len(out)

Synthetische Testsignale erzeugen

Ein Testsignal unterscheidet sich von einer echten Aufnahme in genau einem Punkt: man kennt die richtige Antwort, bevor der Block sie liefert.

Ein einfacher Sprung mit Rauschen

Für den Pegelwächter reicht ein Signal, das zwischen zwei bekannten Pegeln springt, überlagert mit etwas Rauschen — genug, um Hysterese und Zustandsverhalten wirklich zu prüfen, statt nur den Idealfall:

rng = np.random.default_rng(42)          # fester Seed -> reproduzierbar

n = 2000
signal = np.zeros(n, dtype=np.float32)
signal[500:1000] = 1.0                   # bekannter "hoch"-Abschnitt
signal[1500:1800] = 1.0                  # zweiter bekannter Abschnitt

rauschen = rng.normal(0, 0.05, n).astype(np.float32)
testsignal = signal + rauschen

rng = np.random.default_rng(42)          # fester Seed -> reproduzierbar

n = 2000
signal = np.zeros(n, dtype=np.float32)
signal[500:1000] = 1.0                   # bekannter "hoch"-Abschnitt
signal[1500:1800] = 1.0                  # zweiter bekannter Abschnitt

rauschen = rng.normal(0, 0.05, n).astype(np.float32)
testsignal = signal + rauschen

Ground Truth explizit mitführen

Die eigentliche Testgrundlage ist nicht das Signal selbst, sondern eine separate, von Hand aufgeschriebene Erwartung, die nicht aus demselben Code abgeleitet wurde, der auch den Block unter Test bildet:

erwartung = np.zeros(n, dtype=np.float32)
erwartung[500:1000] = 1.0
erwartung[1500:1800] = 1.0
# erwartung ist bewusst identisch zu "signal" (ohne Rauschen) aufgebaut -
# das ist die Antwort, die der Pegelwaechter trotz Rauschen liefern soll

erwartung = np.zeros(n, dtype=np.float32)
erwartung[500:1000] = 1.0
erwartung[1500:1800] = 1.0
# erwartung ist bewusst identisch zu "signal" (ohne Rauschen) aufgebaut -
# das ist die Antwort, die der Pegelwaechter trotz Rauschen liefern soll

Kontrolliertes Signal-Rausch-Verhältnis

Um die Robustheit systematisch zu prüfen (nicht nur „funktioniert mit ein bisschen Rauschen"), das Rauschen über einen definierten Pegel statt eines geratenen Werts steuern:

snr_db = 6.0
rausch_amplitude = 10 ** (-snr_db / 20)   # bei Nutzsignal-Amplitude 1.0
rauschen = rng.normal(0, rausch_amplitude, n).astype(np.float32)

snr_db = 6.0
rausch_amplitude = 10 ** (-snr_db / 20)   # bei Nutzsignal-Amplitude 1.0
rauschen = rng.normal(0, rausch_amplitude, n).astype(np.float32)

Mit einer Schleife über mehrere snr_db-Werte lässt sich so herausfinden, ab welchem Rauschpegel der eigene Block anfängt, falsche Übergänge zu erkennen — eine viel härtere Prüfung als ein einzelner Testlauf.

Alles zusammen: der erste vollständige Test

Quelle, Block unter Test und Senke verbinden, Ergebnis gegen die Ground Truth vergleichen.

import numpy as np
from gnuradio import gr, blocks
import pegelwaechter_block            # eigene Datei mit der Klasse von oben

rng = np.random.default_rng(42)
n = 2000
signal = np.zeros(n, dtype=np.float32)
signal[500:1000] = 1.0
signal[1500:1800] = 1.0
testsignal = signal + rng.normal(0, 0.05, n).astype(np.float32)

tb = gr.top_block()
quelle = blocks.vector_source_f(testsignal.tolist(), repeat=False)
watcher = pegelwaechter_block.blk(low=0.3, high=0.7)
senke = blocks.vector_sink_f()

tb.connect(quelle, watcher, senke)
tb.run()

ergebnis = np.array(senke.data())
abweichungen = np.sum(ergebnis != signal)
print(f"Abweichende Samples: {abweichungen} von {n}")
assert abweichungen == 0, "Pegelwaechter weicht von der Ground Truth ab"

import numpy as np
from gnuradio import gr, blocks
import pegelwaechter_block            # eigene Datei mit der Klasse von oben

rng = np.random.default_rng(42)
n = 2000
signal = np.zeros(n, dtype=np.float32)
signal[500:1000] = 1.0
signal[1500:1800] = 1.0
testsignal = signal + rng.normal(0, 0.05, n).astype(np.float32)

tb = gr.top_block()
quelle = blocks.vector_source_f(testsignal.tolist(), repeat=False)
watcher = pegelwaechter_block.blk(low=0.3, high=0.7)
senke = blocks.vector_sink_f()

tb.connect(quelle, watcher, senke)
tb.run()

ergebnis = np.array(senke.data())
abweichungen = np.sum(ergebnis != signal)
print(f"Abweichende Samples: {abweichungen} von {n}")
assert abweichungen == 0, "Pegelwaechter weicht von der Ground Truth ab"

Einen ganzen, kompilierten Flowgraph headless testen

Sobald mehrere eigene Blöcke zu einem vollständigen .grc-Flowgraph zusammengesetzt sind (mit oder ohne Qt-Anzeige), lässt der sich genauso automatisiert prüfen — ganz ohne geöffnetes Fenster.

grcc -o build flowgraph.grc         # erzeugt build/flowgraph.py

# Ohne Bildschirm ausführen, nach fester Zeit beenden, Log prüfen:
export QT_QPA_PLATFORM=offscreen
python3 build/flowgraph.py > lauf.log 2>&1 &
PID=$!
sleep 10
kill -INT $PID; sleep 1; kill -9 $PID 2>/dev/null

grep -iE "error|traceback|exception" lauf.log   # muss leer bleiben

grcc -o build flowgraph.grc         # erzeugt build/flowgraph.py

# Ohne Bildschirm ausführen, nach fester Zeit beenden, Log prüfen:
export QT_QPA_PLATFORM=offscreen
python3 build/flowgraph.py > lauf.log 2>&1 &
PID=$!
sleep 10
kill -INT $PID; sleep 1; kill -9 $PID 2>/dev/null

grep -iE "error|traceback|exception" lauf.log   # muss leer bleiben

Dieses Muster — im Hintergrund starten, feste Zeit warten, sauber beenden, das Log auf Fehler durchsuchen — ist der einzige zuverlässige Weg, einen kompletten Flowgraph automatisiert zu prüfen, ohne jedes Mal eine grafische Oberfläche von Hand zu bedienen und wieder zu schließen.

Checkliste für den Einstieg

1. Erst prüfen, dass from gnuradio import gr und grcc --help ohne Fehler laufen — alles Weitere baut darauf auf.
2. Jeden eigenen Block als eigenständige Python-Klasse in einer eigenen Datei halten, nicht nur im GRC-Textfeld — dann lässt er sich direkt importieren und testen.
3. vector_source/vector_sink statt echter Hardware für jeden Test — kein Testlauf sollte eine Antenne oder Soundkarte brauchen.
4. Ground Truth getrennt vom Testsignal aufschreiben, nie aus demselben Rechenweg ableiten wie die Prüfung selbst.
5. Für jeden Zufallstest einen festen Seed verwenden (np.random.default_rng(seed)) — ein fehlgeschlagener Test muss exakt wiederholbar sein.
6. Sobald mehrere Blöcke zu einem echten Flowgraph zusammenkommen: headless mit Zeitbegrenzung laufen lassen und das Log auf Tracebacks prüfen, bevor an der Signalqualität gezweifelt wird.