Kivy-Frontpanel: Frequenzanzeige + Spektrum

Dieses Projekt ist der erste konkrete Umsetzungsschritt aus Ein Frontpanel für den USB-Empfänger: Frequenzanzeige und Spektrumdarstellung — zwei der QT-GUI-Elemente des Empfängers — wurden als eigenständige, wiederverwendbare Kivy-Widgets nachgebaut und an die echte Demodulationskette des Test-Empfängers angebunden. Die Spektrum-/Zeitsignalanzeige nutzt matplotlib statt eigenem Kivy-Zeichencode, zwei Buttons stimmen den Empfänger live um.

1. Architektur

Signalverarbeitung und Darstellung sind bewusst getrennt:

• spectrum_source.grc/.py — ein reiner GNU-Radio-Flowgraph ohne GUI (generate_options: no_gui), der dieselbe Demod-Kette wie ssb_usb_144300_file_test.grc verwendet: IQ-Testdatei → Freq Xlating FIR Filter (USB-Bandpass) → Realteil → AGC. Aus dieser Kette werden drei Probe Signal Vector-Blöcke gespeist (Details in Abschnitt 3). Wird von der Kivy-App importiert, nicht eigenständig ausgeführt.
• spectrum_widget.py — SpectrumWidget: matplotlib-Widget mit drei Achsen (Original-Spektrum, demoduliertes Spektrum, demoduliertes Zeitsignal). Kennt GNU Radio nicht, nur die Kivy-Properties orig_spectrum, demod_spectrum, demod_time.
• frequency_display.py — FrequencyDisplay: formatiert einen Hz-Wert als „xxx.xxxx MHz".
• main.py — die Kivy-App, die den Flowgraph startet, die drei Probes pollt und zwei Buttons zur Frequenzumstimmung bereitstellt.

Diese Trennung hat einen praktischen Vorteil für den Entwicklungsablauf: Die komplette Signalverarbeitung lässt sich zunächst ganz normal mit GUI in GRC entwickeln und testen — mit QT GUI Freq Sink, QT GUI Time Sink & Co. wie in jedem anderen Flowgraph. Erst wenn die Kette steht, wird generate_options von qt_gui auf no_gui umgestellt und die passenden Probe Signal Vector-Blöcke angehängt — der Flowgraph wird dadurch zu einer reinen, importierbaren Python-Klasse ohne eigenes Fenster. Ab diesem Punkt übernimmt Kivy vollständig die Optik; an der GNU-Radio-Seite ändert sich inhaltlich nichts mehr.

IQ-Datei → Freq Xlating FIR Filter → Realteil → AGC → demoduliertes Audio
   │                                                        │
   └─ Log Power FFT (Original) ─ Probe               ┌──────┴──────┐
                                                Log Power FFT   Stream to Vector
                                                  (Demod)         (Zeitsignal)
                                                    │                 │
                                                  Probe             Probe
                                                    ↑                 ↑
                                      (siehe Abschnitt 2: Kommunikation)
                                                    ↓
                              Kivy: Clock.schedule_interval → SpectrumWidget (matplotlib)

IQ-Datei → Freq Xlating FIR Filter → Realteil → AGC → demoduliertes Audio
   │                                                        │
   └─ Log Power FFT (Original) ─ Probe               ┌──────┴──────┐
                                                Log Power FFT   Stream to Vector
                                                  (Demod)         (Zeitsignal)
                                                    │                 │
                                                  Probe             Probe
                                                    ↑                 ↑
                                      (siehe Abschnitt 2: Kommunikation)
                                                    ↓
                              Kivy: Clock.schedule_interval → SpectrumWidget (matplotlib)

Verifiziert (GNU Radio 3.10.12.0, Kivy 2.3.1, matplotlib 3.10.1, tatsächlich ausgeführt und per Widget.export_to_png() geprüft):

Oben die Frequenzanzeige „0.5000 MHz" mit den Umstimm-Buttons, darunter die drei matplotlib-Achsen: das Original-IQ-Spektrum mit deutlichem Träger bei 500 kHz, das demodulierte NF-Spektrum und das demodulierte Zeitsignal als sauberer Ton über 1024 Samples.

2. Die Kommunikation zwischen GRC und Kivy im Detail

Das ist der eigentlich interessante Teil, weil er nicht offensichtlich ist: GNU Radio und Kivy laufen im selben Prozess, aber in völlig unabhängigen Thread-Welten — ein naiver Zugriff über die Thread-Grenze hinweg produziert schnell schwer nachvollziehbare Fehler.

Die drei Zugriffe über die Thread-Grenze aus den folgenden Abschnitten im Überblick: 1. das Auslesen des Spektrums per .level() (Abschnitt 2.2/2.3), 2. das einmalige Auslesen der Frequenz beim Start (Abschnitt 2.4) und 3. das Zurückschreiben einer neuen Frequenz per .set_center_freq() (Abschnitt 2.5, für spätere Schritte).

2.1 Ein Prozess, zwei unabhängige Thread-Welten

main.py startet einen einzigen Python-Prozess. Darin laufen nebeneinander:

• Kivys Hauptthread: die Kivy-Event-Loop (App.run()), zuständig für Rendering, Touch-Events und für das Auslösen aller Clock-Callbacks.
• GNU Radios Scheduler-Threads: tb.start() erzeugt für den Flowgraph eigene Worker-Threads (GNU Radio nutzt intern „Thread-per-Block"-Scheduling) — jeder Block bekommt seinen eigenen Thread, der fortlaufend work() aufruft, sobald Daten anliegen. Der Throttle-Block bremst das Ganze auf Echtzeittempo.

Beide Threadwelten laufen völlig unabhängig und wissen im Grunde nichts voneinander — sie teilen sich nur denselben Prozessspeicher.

2.2 Die Brücke: Probe Signal Vector

Alle drei Probe-Blöcke (blocks_probe_signal_vx_0/1/2) sind genau für diesen Zweck gebaut — ihre C++-Dokumentation beschreibt sie wörtlich als „Sink that allows a vector of samples to be grabbed from Python". Jeder hat keinen Ausgang: Sobald er von seinem vorgeschalteten Block (Log Power FFT bzw. Stream to Vector) einen neuen Vektor bekommt, speichert er nur dessen aktuellen Inhalt intern ab und verwirft den Rest. Die einzige nach außen sichtbare Methode ist .level() — sie liefert eine Kopie des zuletzt gespeicherten Vektors zurück. Aufrufbar ist sie von jedem Thread aus, weil genau das der vorgesehene Verwendungszweck ist (dieselbe Technik nutzt bereits analog.probe_avg_mag_sqrd_f im Frontpanel-Artikel für das S-Meter).

.level() liefert dabei immer den kompletten Vektor auf einmal zurück, nicht sequenziell einzelne Samples — kein Iterator, kein Generator. Das genaue Format je Probe ist in Abschnitt 3.1 beschrieben. Da GNU Radio die FFT-Frames bzw. Zeitsignal-Schnappschüsse deutlich öfter erzeugt, als Kivy pollt, überschreibt jeder Probe-Block seinen internen Puffer zwischen zwei Kivy-Updates unter Umständen mehrfach — ältere Frames gehen dabei verloren, es zählt immer nur der zuletzt gespeicherte. Für eine Live-Anzeige ist das gewollt, für eine lückenlose Aufzeichnung jedes einzelnen Frames wäre der Ansatz ungeeignet.

2.3 Wie Kivy tatsächlich zugreift

Clock.schedule_interval(self._update, 1.0 / UPDATE_HZ)

def _update(self, _dt):
    self.spectrum.orig_spectrum = list(self.tb.blocks_probe_signal_vx_0.level())
    self.spectrum.demod_spectrum = list(self.tb.blocks_probe_signal_vx_1.level())
    self.spectrum.demod_time = list(self.tb.blocks_probe_signal_vx_2.level())

Clock.schedule_interval(self._update, 1.0 / UPDATE_HZ)

def _update(self, _dt):
    self.spectrum.orig_spectrum = list(self.tb.blocks_probe_signal_vx_0.level())
    self.spectrum.demod_spectrum = list(self.tb.blocks_probe_signal_vx_1.level())
    self.spectrum.demod_time = list(self.tb.blocks_probe_signal_vx_2.level())

Clock.schedule_interval sorgt dafür, dass _update() garantiert im Kivy-Hauptthread aufgerufen wird — Kivy ruft alle Clock-Callbacks aus seiner eigenen Event-Loop heraus auf, nie aus einem Fremdthread. Die drei .level()-Aufrufe sind dabei ganz gewöhnliche, synchrone Python-Methodenaufrufe — kein Socket, keine Warteschlange, keine Serialisierung. Sie funktionieren nur deshalb sicher, weil probe_signal_vx von GNU-Radio-Seite dafür gebaut wurde, aus einem fremden Thread heraus gelesen zu werden.

2.4 Die Frequenz beim Start: noch einfacher, kein Probe nötig

Für die anfängliche Frequenzanzeige braucht es nicht einmal einen Probe-Block. Jede GRC-Variable wird im generierten Code zu einem ganz gewöhnlichen Instanzattribut:

# aus spectrum_source.py, von grcc generiert
self.carrier_freq = carrier_freq = 500000
...
def set_carrier_freq(self, carrier_freq):
    self.carrier_freq = carrier_freq
    self.freq_xlating_fir_filter_xxx_0.set_center_freq(self.carrier_freq)

# aus spectrum_source.py, von grcc generiert
self.carrier_freq = carrier_freq = 500000
...
def set_carrier_freq(self, carrier_freq):
    self.carrier_freq = carrier_freq
    self.freq_xlating_fir_filter_xxx_0.set_center_freq(self.carrier_freq)

main.py liest das beim Start direkt aus: self.tb.carrier_freq. Für einen Wert, der sich nicht kontinuierlich von selbst ändert, reicht ein simpler Attributzugriff — kein Polling, kein Probe-Block nötig. Erst für sich laufend ändernde Werte (wie die Spektren und das Zeitsignal, 20-mal pro Sekunde neu) braucht es die in Abschnitt 2.2/2.3 beschriebene Probe-Technik.

2.5 Die andere Richtung: Kivy steuert GNU Radio

Genau diese generierte set_carrier_freq()-Methode nutzt die App auch umgekehrt: Die beiden Umstimm-Buttons rufen sie direkt im on_press-Callback auf (der ebenfalls im Kivy-Hauptthread feuert):

# aus main.py
def _step_frequency(self, delta_hz):
    new_freq = self.tb.carrier_freq + delta_hz
    self.tb.set_carrier_freq(new_freq)
    self.freq_display.frequency_hz = new_freq

# aus main.py
def _step_frequency(self, delta_hz):
    new_freq = self.tb.carrier_freq + delta_hz
    self.tb.set_carrier_freq(new_freq)
    self.freq_display.frequency_hz = new_freq

set_carrier_freq() setzt dabei nicht nur das Instanzattribut, sondern ruft — weil carrier_freq als center_freq-Parameter des Freq Xlating FIR Filter verwendet wird — automatisch auch freq_xlating_fir_filter_xxx_0.set_center_freq() mit auf (siehe generierter Code oben, von grcc ohne weiteres Zutun erzeugt). Ein Tastendruck stimmt damit den laufenden Empfänger tatsächlich live um. Auch das bleibt ein einfacher, synchroner Methodenaufruf — GRC generiert für jede Variable automatisch passende set_*/get_*-Methoden auf der Flowgraph-Klasse, unabhängig davon, ob der Aufruf von Kivy, Qt oder einem beliebigen anderen Python-Code kommt.

2.6 Lebenszyklus

def build(self):
    self.tb = spectrum_source()
    self.tb.start()   # nicht blockierend - startet Scheduler-Threads und kehrt sofort zurueck
    ...

def on_stop(self):
    self.tb.stop()
    self.tb.wait()    # wartet, bis alle Blockthreads sauber beendet sind

def build(self):
    self.tb = spectrum_source()
    self.tb.start()   # nicht blockierend - startet Scheduler-Threads und kehrt sofort zurueck
    ...

def on_stop(self):
    self.tb.stop()
    self.tb.wait()    # wartet, bis alle Blockthreads sauber beendet sind

tb.start() ist nicht blockierend — genau deshalb kann es aus build() heraus aufgerufen werden, ohne Kivys eigene Event-Loop (die erst danach mit app.run() startet) zu blockieren. Ohne das saubere stop()/wait()-Paar in on_stop() würden die GNU-Radio-Threads beim Schließen des Fensters weiterlaufen.

3. Datenformat, Verdrahtung und matplotlib-Anzeige

Die Anzeige wurde bewusst nicht mehr mit eigenem Kivy-Zeichencode (Balken, Linien über kivy.graphics) umgesetzt, sondern mit matplotlib über kivy_garden.matplotlib — Achsenbeschriftung, Gitternetz, automatische Skalierung und die vertraute matplotlib-API kommen dadurch geschenkt.

3.1 Verdrahtung im Flowgraph

spectrum_source.grc verzweigt die Empfänger-Kette an zwei Stellen, um Original- und demoduliertes Signal getrennt zu erfassen:

blocks_file_source_0 → blocks_throttle2_0 ─┬─────────────────────────────────────────────┐
                                            │                                              │
                                            ▼                                              ▼
                                   fft_logpwrfft_x_0                        freq_xlating_fir_filter_xxx_0
                                   (Original-IQ, complex)                    (USB-Bandpass, center_freq=carrier_freq)
                                            │                                              │
                                            ▼                                              ▼
                                blocks_probe_signal_vx_0                       blocks_complex_to_real_0
                                (Original-Spektrum)                                        │
                                                                                            ▼
                                                                                  analog_agc_xx_0
                                                                                            │
                                                                    ┌───────────────────────┴───────────────────────┐
                                                                    ▼                                               ▼
                                                          fft_logpwrfft_x_1                            blocks_stream_to_vector_0
                                                          (demoduliert, float)                          (time_len=1024 Samples)
                                                                    │                                               │
                                                                    ▼                                               ▼
                                                       blocks_probe_signal_vx_1                        blocks_probe_signal_vx_2
                                                       (Demod-Spektrum)                                (Demod-Zeitsignal)

blocks_file_source_0 → blocks_throttle2_0 ─┬─────────────────────────────────────────────┐
                                            │                                              │
                                            ▼                                              ▼
                                   fft_logpwrfft_x_0                        freq_xlating_fir_filter_xxx_0
                                   (Original-IQ, complex)                    (USB-Bandpass, center_freq=carrier_freq)
                                            │                                              │
                                            ▼                                              ▼
                                blocks_probe_signal_vx_0                       blocks_complex_to_real_0
                                (Original-Spektrum)                                        │
                                                                                            ▼
                                                                                  analog_agc_xx_0
                                                                                            │
                                                                    ┌───────────────────────┴───────────────────────┐
                                                                    ▼                                               ▼
                                                          fft_logpwrfft_x_1                            blocks_stream_to_vector_0
                                                          (demoduliert, float)                          (time_len=1024 Samples)
                                                                    │                                               │
                                                                    ▼                                               ▼
                                                       blocks_probe_signal_vx_1                        blocks_probe_signal_vx_2
                                                       (Demod-Spektrum)                                (Demod-Zeitsignal)

Der Freq Xlating FIR Filter ist exakt derselbe Block, der auch im echten Empfänger die Empfangsfrequenz festlegt — hier heißt die zugehörige Variable carrier_freq statt freq_tune, weil die Testdatei ein festes IQ-Signal statt eines RTL-SDR-Streams liefert, das Prinzip ist aber identisch (siehe Abschnitt 2.4/2.5).

3.2 Format der gelieferten Daten

Bei beiden Log Power FFT-Blöcken ist shift='True' gesetzt — dadurch liegt Index 0 des zurückgegebenen Vektors bereits bei der niedrigsten Frequenz (−samp_rate/2) und der letzte Index bei der höchsten, statt in der für FFTs nativen Reihenfolge (0, positive, dann negative Frequenzen). Die Frequenzachse lässt sich dadurch mit einer einzigen linearen Formel aus dem Index berechnen, ganz ohne np.fft.fftshift:

# aus spectrum_widget.py
n = len(orig)
freqs_khz = (np.arange(n) - n / 2) * (self.iq_samp_rate / n) / 1000.0

# aus spectrum_widget.py
n = len(orig)
freqs_khz = (np.arange(n) - n / 2) * (self.iq_samp_rate / n) / 1000.0

Für das Zeitsignal gibt es keine FFT und damit auch kein shift — der Index entspricht direkt der zeitlichen Abfolge der demodulierten Samples, blocks_stream_to_vector_0 gruppiert dafür einfach time_len aufeinanderfolgende Samples zu einem Vektor.

3.3 Original und demoduliert getrennt, mit Frequenzsteuerung

SpectrumWidget zeichnet drei matplotlib-Achsen untereinander statt einer gemeinsamen Darstellung — Original- und demoduliertes Spektrum haben unterschiedliche Bandbreiten (iq_samp_rate = 1,92 MHz vs. audio_rate = 48 kHz) und sind dadurch getrennt besser lesbar:

# aus spectrum_widget.py
self.ax_orig.set_title("Original-Spektrum (IQ, vor Demodulation)")
self.ax_orig.set_xlabel("Frequenz [kHz]")

self.ax_demod.set_title("Demoduliertes Spektrum (NF-Audio)")
self.ax_demod.set_xlabel("Frequenz [Hz]")

self.ax_time.set_title("Demoduliertes Zeitsignal (nach Demodulation)")
self.ax_time.set_xlabel(f"Sample n (0 … {n-1}, entspricht 0 … {duration_ms:.1f} ms bei {audio_rate:.0f} Hz)")

# aus spectrum_widget.py
self.ax_orig.set_title("Original-Spektrum (IQ, vor Demodulation)")
self.ax_orig.set_xlabel("Frequenz [kHz]")

self.ax_demod.set_title("Demoduliertes Spektrum (NF-Audio)")
self.ax_demod.set_xlabel("Frequenz [Hz]")

self.ax_time.set_title("Demoduliertes Zeitsignal (nach Demodulation)")
self.ax_time.set_xlabel(f"Sample n (0 … {n-1}, entspricht 0 … {duration_ms:.1f} ms bei {audio_rate:.0f} Hz)")

Die beiden Umstimm-Buttons demonstrieren dabei anschaulich, warum die Trennung von Original- und Demod-Spektrum sinnvoll ist: Im Original-Spektrum bleibt der Träger unabhängig von der Einstellung immer an derselben Stelle sichtbar (er kommt ja unverändert aus der Datei), während das demodulierte Spektrum bei Verstimmung sofort zusammenbricht, weil der USB-Bandpass am Träger vorbeizeigt.

Verifiziert — Vergleich vor/nach einer Verstimmung um +15 kHz (per Button-Callback simuliert, _step_frequency(15000)):

Das Original-Spektrum zeigt den Träger unverändert bei 500 kHz (blau, oben) — die Verstimmung wirkt sich hier nicht aus, weil dieser Pfad vor dem Freq Xlating FIR Filter abgegriffen wird. Das demodulierte Spektrum (grün, Mitte) zeigt dagegen nur noch die beiden Filterflanken statt eines sauberen Tons in der Mitte, und die Zeitsignal-Amplitude (orange, unten) bricht von ±0,4 auf ±0,008 ein — der USB-Bandpass filtert jetzt vorwiegend Rauschen statt Nutzsignal heraus.

3.4 Eine Kivy-Falle: self.canvas ist reserviert

Beim Einbinden von FigureCanvasKivyAgg als Attribut eines Kivy-Widgets liegt ein naheliegender Name nahe: self.canvas. Genau das führt aber zu einem kryptischen Fehler, weil jedes Kivy-Widget bereits eine eingebaute canvas-Property besitzt (die internen Grafikbefehle des Widgets selbst) — eine Zuweisung self.canvas = FigureCanvasKivyAgg(...) überschreibt diese stillschweigend:

# falsch - self.canvas ist bereits von Kivy belegt
self.canvas = FigureCanvasKivyAgg(self.figure)
self.add_widget(self.canvas)
# → AttributeError: 'FigureCanvasKivyAgg' object has no attribute 'add'

# falsch - self.canvas ist bereits von Kivy belegt
self.canvas = FigureCanvasKivyAgg(self.figure)
self.add_widget(self.canvas)
# → AttributeError: 'FigureCanvasKivyAgg' object has no attribute 'add'

Der Fehler tritt erst beim add_widget()-Aufruf auf, weil Kivy intern versucht, das eigene Grafik-Canvas (jetzt fälschlich das matplotlib-Objekt) um das neue Kind-Widget zu erweitern. Die Lösung ist ein Attributname ohne Kollision:

# richtig
self.fig_canvas = FigureCanvasKivyAgg(self.figure)
self.add_widget(self.fig_canvas)

# richtig
self.fig_canvas = FigureCanvasKivyAgg(self.figure)
self.add_widget(self.fig_canvas)

4. Ausführen

pip install kivy kivy_garden.matplotlib
python3 main.py

pip install kivy kivy_garden.matplotlib
python3 main.py

5. Nächste Schritte

Weitere QT-Elemente des Empfängers, die sich nach demselben Pull-Prinzip nachbauen ließen:

• Lautstärke-/Squelch-Regler als Kivy Slider, dessen on_value per set_volume()/set_squelch_threshold() (Abschnitt 2.5) direkt in den Flowgraph zurückschreibt.
• LED-Indikator für den Squelch-Status — Polling wie beim Spektrum, nur mit einem einzelnen Bool-Wert statt eines Vektors.
• S-Meter als Schwesterwidget von SpectrumWidget, gespeist über analog.probe_avg_mag_sqrd_f statt probe_signal_vf (aggregierter Einzelwert statt ganzem Spektrum) — exakt die im Frontpanel-Artikel für die nativen QT-Widgets gezeigte Probe-Technik, nur mit Kivy statt Dial Gauge als Anzeige.