Probe-Blöcke im Überblick
Acht Blöcke, die einen Wert aus dem laufenden Flowgraph nach Python (oder nach außen) durchreichen, statt ihn selbst anzuzeigen — mit grundlegend unterschiedlichen Zugriffsmechanismen und Rückgabeformaten.
Alle „Probe"-Blöcke haben eines gemeinsam: Sie hängen als Sackgasse (kein oder kaum ein Ausgang) an einem Signalpfad und machen dessen aktuellen Zustand für eigenen Python-Code zugänglich — genau das Muster, das im Kivy-Frontpanel-Projekt für Spektrum, Zeitsignal und Frequenzanzeige verwendet wird. Sie unterscheiden sich aber deutlich darin, wie der Wert abgeholt wird und was genau er enthält.
1. Zwei grundlegend verschiedene Zugriffsmuster
| Muster | Blöcke | Wie der Wert ankommt |
|---|---|---|
| Pull (synchroner Methodenaufruf) | Probe Signal, Probe Signal Vector, Probe Avg Mag^2, Probe Density | eigener Code ruft eine Getter-Methode (.level(), .density()) auf und bekommt den zuletzt gespeicherten Wert zurück |
| Push (asynchrone PMT-Nachricht) | Probe Rate | der Block sendet selbstständig in festen Abständen eine Nachricht an einen verbundenen Message-Port |
| Extern über ControlPort | Ctrlport Probe, Ctrlport Complex Probe | der Wert wird per Thrift-RPC nach außen exportiert, für einen separaten ControlPort-Client — kein Python-Methodenaufruf im selben Prozess |
| Meta-Block | Function Probe | kein eigener Signaleingang — pollt stattdessen periodisch eine beliebige Python-Methode eines anderen Blocks und schreibt das Ergebnis in eine GRC-Variable |
Das Pull-Muster ist dasjenige, das im Kivy-Projekt verwendet wird (siehe dort Abschnitt 2.2/2.3): Clock.schedule_interval ruft .level() synchron aus dem Kivy-Hauptthread ab, sicher über die Thread-Grenze hinweg, weil die Probe-Blöcke genau dafür gebaut sind.
2. Probe Signal
Der einfachste Probe-Block: ein einzelner Skalarwert, keine Mittelung, keine Vektorisierung. .level() liefert exakt das zuletzt verarbeitete Sample zurück — sobald ein neues Sample eintrifft, wird der alte Wert überschrieben.
| Parameter | Bedeutung |
|---|---|
type (Input Type) |
Complex/Float/Int/Short/Byte — bestimmt sowohl den erwarteten Eingangsdatentyp als auch den Python-Klassennamen (probe_signal_c/f/i/s/b) |
Inhalt der Variable: ein einzelner Python-Skalar (float, complex oder int, je nach type) — kein Array, kein Vektor.
from gnuradio import gr, blocks
tb = gr.top_block()
src = blocks.vector_source_f(list(range(1, 21)), repeat=False) # 1.0 ... 20.0
probe = blocks.probe_signal_f()
tb.connect(src, probe)
tb.run()
print(type(probe.level()), probe.level())
# <class 'float'> 20.0 -- exakt das letzte verarbeitete Sample
from gnuradio import gr, blocks
tb = gr.top_block()
src = blocks.vector_source_f(list(range(1, 21)), repeat=False) # 1.0 ... 20.0
probe = blocks.probe_signal_f()
tb.connect(src, probe)
tb.run()
print(type(probe.level()), probe.level())
# <class 'float'> 20.0 -- exakt das letzte verarbeitete Sample
3. Probe Signal Vector
Dieselbe Idee wie Probe Signal, aber für vektorwertige Eingänge (z. B. den Ausgang eines Log Power FFT- oder Stream to Vector-Blocks). .level() liefert immer den kompletten zuletzt empfangenen Vektor auf einmal zurück — genau der Block, der im Kivy-Frontpanel für Spektrum und Zeitsignal verwendet wird.
| Parameter | Bedeutung |
|---|---|
type |
Complex/Float/Int/Short/Byte, wie bei Probe Signal |
vlen (Vector Length) |
Länge eines Vektors — muss zur vlen des Eingangs passen (z. B. fft_size bei einem Log Power FFT) |
Inhalt der Variable: eine Python-Liste (bzw. ein listenartiges Objekt) mit genau vlen Elementen — der jeweils zuletzt vollständig eingetroffene Vektor. Ältere Vektoren werden beim nächsten Eintreffen vollständig überschrieben, nicht angehängt.
from gnuradio import gr, blocks
tb = gr.top_block()
data = [float(i) for i in range(24)] # 0.0 ... 23.0
src = blocks.vector_source_f(data, repeat=False)
s2v = blocks.stream_to_vector(gr.sizeof_float, 8)
probe = blocks.probe_signal_vf(8)
tb.connect(src, s2v, probe)
tb.run()
print(type(probe.level()), len(probe.level()), list(probe.level()))
# <class 'list'> 8 [16.0, 17.0, 18.0, 19.0, 20.0, 21.0, 22.0, 23.0]
# -- die letzte vollstaendige 8er-Gruppe, nicht der Gesamtstream
from gnuradio import gr, blocks
tb = gr.top_block()
data = [float(i) for i in range(24)] # 0.0 ... 23.0
src = blocks.vector_source_f(data, repeat=False)
s2v = blocks.stream_to_vector(gr.sizeof_float, 8)
probe = blocks.probe_signal_vf(8)
tb.connect(src, s2v, probe)
tb.run()
print(type(probe.level()), len(probe.level()), list(probe.level()))
# <class 'list'> 8 [16.0, 17.0, 18.0, 19.0, 20.0, 21.0, 22.0, 23.0]
# -- die letzte vollstaendige 8er-Gruppe, nicht der Gesamtstream
Genau dieser Rückgabetyp ist der Grund, warum spectrum_widget.py im Kivy-Projekt die Werte explizit mit np.asarray(..., dtype=np.float32) in ein NumPy-Array umwandelt, statt sie direkt an matplotlib zu übergeben: Kivy-Properties (ListProperty) sind selbst „beobachtbare" Python-Listen mit eigener Callback-Maschinerie — reicht man sie ungewandelt an matplotlib durch, das intern copy.copy() auf den Werten aufruft, kollidiert das mit genau dieser Kivy-internen Beobachtung und führt zu einer RecursionError.
4. Probe Avg Mag^2
Anders als die beiden vorherigen Blöcke liefert dieser keinen Rohwert, sondern einen mit einem Einpol-IIR-Filter exponentiell geglätteten Mittelwert von |x|² — dieselbe Technik, die im Frontpanel-Artikel für das S-Meter verwendet wird.
| Parameter | Bedeutung |
|---|---|
type |
Complex, Float oder „Complex→Float" (komplexer Eingang, aber float-typisierter zusätzlicher Ausgang) |
threshold (Threshold (dB)) |
Schwelle für den optionalen zweiten Ausgang (1.0/0.0, je nachdem ob der geglättete Pegel darüber liegt) — wirkt sich nicht auf .level() aus |
alpha (Alpha) |
Glättungsfaktor des Einpol-IIR-Filters; kleiner = träger, glatter; größer = schneller, verrauschter |
Inhalt der Variable: ein einzelner float — der geglättete Mittelwert der linearen Leistung (|x|²), nicht in dB, obwohl der Parameter „Threshold (dB)" heißt. Für eine dB-Anzeige muss selbst noch 10·log10(...) gebildet werden.
from gnuradio import gr, blocks, analog
tb = gr.top_block()
amplitude = 0.5
src = analog.sig_source_f(48000, analog.GR_COS_WAVE, 1000, amplitude, 0)
head = blocks.head(gr.sizeof_float, 48000) # 1 Sekunde
probe = analog.probe_avg_mag_sqrd_f(0, 2e-3)
tb.connect(src, head, probe)
tb.run()
print(probe.level())
# 0.12488172224428769
# erwartet: amplitude^2 / 2 = 0.25 / 2 = 0.125 (Mittelwert von cos^2 ueber eine volle Periode)
from gnuradio import gr, blocks, analog
tb = gr.top_block()
amplitude = 0.5
src = analog.sig_source_f(48000, analog.GR_COS_WAVE, 1000, amplitude, 0)
head = blocks.head(gr.sizeof_float, 48000) # 1 Sekunde
probe = analog.probe_avg_mag_sqrd_f(0, 2e-3)
tb.connect(src, head, probe)
tb.run()
print(probe.level())
# 0.12488172224428769
# erwartet: amplitude^2 / 2 = 0.25 / 2 = 0.125 (Mittelwert von cos^2 ueber eine volle Periode)
5. Probe Rate
Misst nicht den Signalinhalt, sondern den tatsächlichen Durchsatz (Samples pro Sekunde) am Eingang — nützlich, um zu prüfen, ob ein Flowgraph wirklich in Echtzeit läuft. Anders als alle bisherigen Blöcke liefert er den Wert nicht über .level(), sondern als asynchrone PMT-Nachricht am Message-Port rate.
| Parameter | Bedeutung |
|---|---|
type / vlen |
Datentyp und Vektorlänge des Eingangs (nur zur Bestimmung der Item-Größe, kein Einfluss auf das Ergebnisformat) |
mintime (Min Update Time (ms)) |
Mindestabstand zwischen zwei Nachrichten |
alpha (Update Alpha) |
Glättungsfaktor für den mitgelieferten rate_avg-Wert |
name (Name) |
frei wählbarer Bezeichner, taucht als name-Feld in jeder Nachricht wieder auf |
Inhalt der Nachricht: kein Skalar und keine Liste, sondern ein PMT-Dictionary mit drei Feldern — name (der konfigurierte Bezeichner), rate_now (aktuell gemessene Rate) und rate_avg (geglättete Rate, gesteuert über alpha). Zum Empfang braucht es einen verbundenen Message-Sink oder einen eigenen Message-Handler (message_port_register_in + set_msg_handler), kein einfacher Attributzugriff.
Throttle auf 48 000 Samples/s begrenzt):
tb.msg_connect(probe, "rate", msg_sink, "print")
...
# ******* MESSAGE DEBUG PRINT ********
# ((name . meine_rate) (rate_now . 47991.8) (rate_avg . 47991.8))
# ************************************
# ******* MESSAGE DEBUG PRINT ********
# ((name . meine_rate) (rate_now . 47350.7) (rate_avg . 47895.6))
# ************************************
tb.msg_connect(probe, "rate", msg_sink, "print")
...
# ******* MESSAGE DEBUG PRINT ********
# ((name . meine_rate) (rate_now . 47991.8) (rate_avg . 47991.8))
# ************************************
# ******* MESSAGE DEBUG PRINT ********
# ((name . meine_rate) (rate_now . 47350.7) (rate_avg . 47895.6))
# ************************************
Die gemessene Rate liegt wie erwartet nahe an den konfigurierten 48 000 Samples/s des Throttle-Blocks.
6. Probe Density
Misst die Dichte von Einsen in einem entpackten Bitstrom (ein Byte pro Bit, Wert 0 oder 1) — z. B. um zu prüfen, ob ein Taktrückgewinnungs- oder Entscheider-Block im Mittel eine plausible 50/50-Verteilung liefert.
| Parameter | Bedeutung |
|---|---|
alpha (Alpha) |
Glättungsfaktor des Einpol-IIR-Filters, wie bei Probe Avg Mag^2 |
probe_rate (Probe Rate) |
im GRC-Dialog vorhanden, wird aber vom generierten Code nicht verwendet — der Aufruf lautet schlicht digital.probe_density_b(alpha). Ein totes GUI-Feld, keine Funktionsänderung. |
Inhalt der Variable: ein einzelner float zwischen 0 und 1 — der geglättete Anteil der als „1" eingetroffenen Bytes. Wichtig: Der Eingang muss ein entpackter Bitstrom sein (jedes Byte trägt genau ein Bit, Wert 0 oder 1), keine gepackten Datenbytes.
from gnuradio import gr, blocks, digital
tb = gr.top_block()
pattern = [1, 0] * 5000 # 50 % Einsen
src = blocks.vector_source_b(pattern, repeat=False)
probe = digital.probe_density_b(0.01)
tb.connect(src, probe)
tb.run()
print(probe.density()) # 0.4974874371859307
tb = gr.top_block()
pattern = ([1] * 8 + [0] * 2) * 1000 # 80 % Einsen
src = blocks.vector_source_b(pattern, repeat=False)
probe = digital.probe_density_b(0.01)
tb.connect(src, probe)
tb.run()
print(probe.density()) # 0.791880026659935
from gnuradio import gr, blocks, digital
tb = gr.top_block()
pattern = [1, 0] * 5000 # 50 % Einsen
src = blocks.vector_source_b(pattern, repeat=False)
probe = digital.probe_density_b(0.01)
tb.connect(src, probe)
tb.run()
print(probe.density()) # 0.4974874371859307
tb = gr.top_block()
pattern = ([1] * 8 + [0] * 2) * 1000 # 80 % Einsen
src = blocks.vector_source_b(pattern, repeat=False)
probe = digital.probe_density_b(0.01)
tb.connect(src, probe)
tb.run()
print(probe.density()) # 0.791880026659935
Beide gemessenen Werte liegen wie erwartet nahe an den tatsächlichen Bitanteilen (0,5 bzw. 0,8) — die kleine Abweichung stammt vom Einschwingverhalten des Glättungsfilters bei alpha=0,01.
7. Ctrlport Probe & Ctrlport Complex Probe
Beide Blöcke exportieren Werte nicht in den eigenen Python-Prozess, sondern über GNU Radios ControlPort — einen in den Flowgraph eingebetteten Thrift-RPC-Server. Gedacht für externe Monitoring-Werkzeuge, die sich von außen (auch über das Netzwerk) auf den laufenden Flowgraph aufschalten, nicht für den Zugriff aus demselben Skript heraus.
| Parameter | Bedeutung |
|---|---|
type (nur Ctrlport Probe) |
Datentyp des Eingangs (Complex/Float/Int/Short/Byte) — Ctrlport Complex Probe ist fest auf Complex spezialisiert |
name / desc |
Bezeichner und Beschreibung, unter denen der Wert im ControlPort-Verzeichnis erscheint |
len (nur Ctrlport Probe) |
Anzahl der auf einmal übertragenen Samples |
disp_mask (Display Mask, nur Ctrlport Probe) |
Hinweis an den ControlPort-Client, als was der Wert dargestellt werden soll: Constellation, Time, PSD, Spectrogram oder Raster |
Inhalt der Variable: kein direkt in Python auslesbarer Wert — die Daten liegen im ControlPort-Verzeichnis des laufenden Prozesses und müssen von einem separaten Thrift-Client abgeholt werden. Das setzt voraus, dass ControlPort im Options-Block aktiviert ist (zusätzliche Thrift-Abhängigkeit). Für das Kivy-Projekt war das bewusst nicht die gewählte Lösung: Der direkte .level()-Zugriff im selben Prozess (Abschnitt 2/3) kommt ohne zusätzliche Server-Infrastruktur aus und war für eine lokale Desktop-App die einfachere Wahl. ControlPort lohnt sich dort, wo mehrere unabhängige Prozesse oder Rechner denselben laufenden Flowgraph beobachten sollen.
8. Function Probe
Ein Sonderfall: kein Signalverarbeitungsblock mit Ein-/Ausgang, sondern ein GRC-Variablenblock. Er pollt in einem eigenen Hintergrundthread periodisch eine beliebige Python-Methode — typischerweise .level() eines der obigen Probe-Blöcke — und schreibt das Ergebnis in eine ganz normale GRC-Variable.
| Parameter | Bedeutung |
|---|---|
block_id (Block ID) |
ID des Blocks, dessen Methode aufgerufen werden soll (leer = die Flowgraph-Klasse selbst) |
function_name (Function Name) |
Name der aufzurufenden Methode, z. B. level |
function_args (Function Args) |
Argumente für den Aufruf, meist leer |
poll_rate (Poll Rate (Hz)) |
wie oft pro Sekunde der Wert abgefragt wird |
value (Initial Value) |
Startwert der Variable, bevor der erste Poll durchgelaufen ist |
Inhalt der Variable: exakt das, was die aufgerufene Methode zurückgibt — bei einem Probe-Signal-Block also ein float, bei einem Probe-Signal-Vector-Block eine Liste. Der generierte Code startet dafür einen threading.Thread, der in einer Endlosschleife self.<block_id>.<function_name>(<function_args>) aufruft, das Ergebnis der Variable zuweist und danach 1.0 / poll_rate Sekunden schläft.