ITA2 Decoder Block

Erstellung eines eigenen, wiederverwertbaren GnuRadio Blocks für den ITA2 Decoder.

Im Rahmen des DDK9-Projekts war die RTTY-Demodulation mit einem ITA2-Decoder eine zentrale Aufgabe. Der erste Ansatz, einen Embedded-Python-Block zu verwenden, hat sich als unflexibel erwiesen. Deshalb wurde der Ansatz verfolgt, einen wiederverwendbaren eigenen Block zu entwickeln. Die hier vorgestellte Vorgehensweise beschreibt die Erstellung dieses Blocks und kann als Vorlage für die Entwicklung weiterer eigener Blöcke dienen.

Der ITA2-Decoder ist ein synchroner GNU-Radio-Block (Out-of-Tree), der serielle Bitströme (0 und 1) in lesbaren Text (ASCII/UTF-8) nach dem ITA2-Standard (Baudot-Code) dekodiert. Seine Besonderheit ist die implementierte Schwungrad-Synchronisation (Flywheel-Sync) sowie ein asynchrones SNR-Gate, die ihn robust gegenüber Fading, Rauschen und Phasenverschiebungen machen, wie sie typischerweise auf Kurzwelle auftreten.

Funktionsprinzip (Theory of Operation)

Normale RTTY-Decoder arbeiten im sogenannten HUNT-Modus: Sie warten auf ein Startbit (0), lesen 5 Datenbits, prüfen das Stoppbit (1) und fangen wieder von vorne an. Geht das Start- oder Stoppbit im Rauschen verloren, spuckt der Decoder fehlerhaften Text aus.

Dieser Block nutzt erweiterte Logiken:

Flywheel-Sync (Schwungrad): Sobald der Decoder eine konfigurierbare Anzahl an korrekten Zeichen am Stück fehlerfrei empfangen hat (lock_threshold), wechselt er in den LOCK-Modus. In diesem Zustand ignoriert der Block kurzzeitig fehlende Start- oder Stoppbits und tastet stur im Rhythmus der Baudrate weiter ab. Erst wenn zu viele Fehler am Stück auftreten (unlock_threshold), fällt er in den HUNT-Modus zurück.
1.5-Stoppbit-Kompensation: Im IDLE-Zustand erfordert das Flywheel mindestens zwei aufeinanderfolgende Mark-Bits (1), bevor es erneut feuert. Dies verhindert, dass der Decoder bei RTTY-Signalen mit 1.5 Stoppbits (z.B. 45.45 Baud) kontinuierlich um ein halbes Bit aus dem Takt gerät.
Unshift on Space (UOS): Eine optionale Fehlerkorrektur. Da ITA2 zwischen Buchstaben (LETTERS) und Zahlen (FIGURES) umschaltet, führt ein verpasstes Umschaltsignal zu langem Kauderwelsch. Ist UOS aktiv, erzwingt jedes empfangene Leerzeichen einen automatischen Wechsel zurück in den Buchstaben-Modus.
SNR-Gate: Um zu verhindern, dass bei Signalverlust reines Rauschen als wirrer Text dekodiert wird, kann der Block über einen Message-Port stummgeschaltet (gegatet) werden.

Parameter (Konfiguration)

Diese Werte werden beim Erstellen des Blocks in GRC (oder via Python) übergeben:

Parameter	Typ	Standardwert	Beschreibung
Lock Threshold	int	5	Anzahl aufeinanderfolgender, fehlerfreier Frames (gültiges Start+Stoppbit), um den Flywheel-LOCK zu aktivieren.
Unlock Threshold	int	8	Anzahl aufeinanderfolgender Framing-Fehler (fehlendes Stoppbit), bevor der Decoder den LOCK verliert und wieder aktiv nach Startbits sucht (HUNT).
Unshift on Space	bool	False	Wenn aktiv (True), schaltet der Decoder bei Empfang eines Leerzeichens (`0b00100`) automatisch von FIGURES zurück auf LETTERS. Sehr nützlich bei schlechten Empfangsbedingungen.
Gate Default	bool	True	Legt fest, ob der Decoder beim Starten des Flowgraphs geöffnet (True) oder geschlossen (False) ist.

Schnittstellen (Ports)

Der Block nutzt eine Kombination aus synchronen Datenströmen (für Bits) und asynchronen Message-Ports (für Steuerungsdaten und Text).

Eingänge

in (Stream, np.int8): Nimmt den geslicten Bitstrom auf. Die Werte müssen zwingend als harte Bits (0 oder 1) anliegen (z.B. von einem Binary-Slicer-Block).
gate (Message, PMT): Erwartet einen PMT-Boolean (pmt.PMT_T oder pmt.PMT_F).
- False: Blockiert die Verarbeitung. Der interne Zustand wird resettet, Flywheel wird deaktiviert.
- True: Setzt die Verarbeitung fort.

Ausgänge

text (Message, PMT): Gibt die dekodierten Zeichen als PMT-Strings aus. Zusätzlich werden Systemmeldungen wie [Flywheel: LOCK] oder [Flywheel: HUNT] sowie ein Bell-Character <BEL> (akustisches Signal) ausgegeben.

Die Zustandsmaschine (State Machine)

Der Block durchläuft intern für jedes Zeichen drei Zustände:

IDLE: Wartet auf ein Startbit (0). Fängt Rauschen und Pausen (Dauerton 1) ab.
DATA: Liest exakt 5 Bits ein, berechnet den Dezimalwert des Zeichens (0–31), wertet die ITA2-Tabelle (LETS oder FIGS) aus und publiziert das Zeichen über den Message-Port.
WAIT_STOP: Prüft das nächste Bit.
- Ist es eine 1 (korrektes Stoppbit), wird der „Good Frame"-Zähler erhöht.
- Ist es eine 0 (Framing-Fehler), wird der Fehler-Zähler erhöht. Ab 3 Fehlern am Stück (im HUNT-Modus) erzwingt der Block einen Hard-Reset des Zeichens.

Best Practices für den GNU Radio Companion (GRC)

Um diesen Block in einem Flowgraph optimal zu nutzen, sollte folgende Kette aufgebaut werden:

Demodulation: z.B. Quadrature Demod oder AFSK-Demodulator (liefert Float-Werte).
Clock Recovery: z.B. Symbol Sync oder M&M Clock Recovery, um exakt ein Sample pro Symbol (Bit) zu erhalten.
Slicing: Ein Binary Slicer wandelt die analogen Floats (+1.0 / -1.0) in harte Nullen und Einsen (int8).
Baudot Decoder: Dieser Block. (Der Input wird mit dem Slicer verbunden.)
Ausgabe: Verbinde den text-Message-Ausgang mit einem Message-Debug-Block (um den Text im Terminal zu sehen) oder mit einer QT-GUI-Message-Edit-Box (um den Text im grafischen UI anzuzeigen).

Implementierung

import numpy as np
import pmt
from gnuradio import gr

class ita2_decoder(gr.sync_block):
    """
    Allgemeingueltiger ITA2/Baudot-Decoder mit Flywheel-Sync und Message-Output.
    """
    # Standard-Tabellen (könnten bei Bedarf auch als Parameter übergeben werden)
    LETS = {0b00100: ' ', 0b10111: 'Q', 0b10011: 'W', 0b00001: 'E', 0b01010: 'R',
            0b10000: 'T', 0b10101: 'Y', 0b00111: 'U', 0b00110: 'I', 0b11000: 'O',
            0b10110: 'P', 0b00011: 'A', 0b00101: 'S', 0b01001: 'D', 0b01101: 'F',
            0b11010: 'G', 0b10100: 'H', 0b01011: 'J', 0b01111: 'K', 0b10010: 'L',
            0b10001: 'Z', 0b11101: 'X', 0b01110: 'C', 0b11110: 'V', 0b11001: 'B',
            0b01100: 'N', 0b11100: 'M', 0b01000: '\r', 0b00010: '\n'}

    FIGS = {0b00100: ' ', 0b10111: '1', 0b10011: '2', 0b00001: '3', 0b01010: '4',
            0b10000: '5', 0b10101: '6', 0b00111: '7', 0b00110: '8', 0b11000: '9',
            0b10110: '0', 0b00011: '-', 0b00101: "'", 0b01001: '$', 0b01101: '!',
            0b11010: '&', 0b10100: '#', 0b01011: '\a', 0b01111: '(', 0b10010: ')',
            0b10001: '+', 0b11101: '/', 0b01110: ':', 0b11110: ';', 0b11001: '?',
            0b01100: ',', 0b11100: '.', 0b01000: '\r', 0b00010: '\n'}

    def __init__(self, lock_threshold=5, unlock_threshold=8, unshift_on_space=False, gate_default=True):
        gr.sync_block.__init__(self,
            name='Baudot Decoder (Flywheel)',
            in_sig=[np.int8],
            out_sig=None) # Reine Message-basierte Ausgabe

        # Parameter registrieren
        self.lock_threshold = lock_threshold
        self.unlock_threshold = unlock_threshold
        self.unshift_on_space = unshift_on_space

        # Ports
        self.message_port_register_in(pmt.intern('gate'))
        self.set_msg_handler(pmt.intern('gate'), self._handle_gate)
        self.message_port_register_out(pmt.intern('text'))

        # Interner Zustand
        self._gate_open = bool(gate_default)
        self._reset()
        self._good = 0
        self._miss = 0
        self._flywheel = False

    def _emit(self, s):
        # Nachricht via Message-Port publizieren
        self.message_port_pub(pmt.intern('text'), pmt.to_pmt(s))

    def _handle_gate(self, msg):
        try:
            if pmt.is_pair(msg): msg = pmt.cdr(msg)
            new_open = bool(pmt.to_python(msg))
        except Exception: return
        if self._gate_open and not new_open:
            self._reset()
            self._flywheel = False
            self._good = 0
            self._miss = 0
        self._gate_open = new_open

    def _reset(self):
        self.state = 'IDLE'
        self.bit_buffer = []
        self.shift_state = 'LETTERS'
        self.framing_errors = 0
        self._fw_used = False
        self._idle_marks = 0

    def work(self, input_items, output_items):
        if not self._gate_open:
            return len(input_items[0])

        for bit in input_items[0]:
            bit = int(bit)

            if self.state == 'IDLE':
                if bit == 0:
                    self._fw_used = False
                    self._idle_marks = 0
                    self.state = 'DATA'
                    self.bit_buffer = []
                else:
                    self._idle_marks += 1
                    if self._flywheel and not self._fw_used and self._idle_marks >= 2:
                        self._fw_used = True
                        self._idle_marks = 0
                        self.state = 'DATA'
                        self.bit_buffer = []

            elif self.state == 'DATA':
                self.bit_buffer.append(bit)
                if len(self.bit_buffer) == 5:
                    code = sum(b << i for i, b in enumerate(self.bit_buffer))

                    if code == 0b11111:
                        self.shift_state = 'LETTERS'
                        self.framing_errors = 0
                    elif code == 0b11011:
                        self.shift_state = 'FIGURES'
                        self.framing_errors = 0
                    elif code == 0b00000:
                        pass
                    else:
                        # Unshift on Space Logik integrieren
                        if code == 0b00100 and self.unshift_on_space:
                            self.shift_state = 'LETTERS'

                        table = self.LETS if self.shift_state == 'LETTERS' else self.FIGS
                        ch = table.get(code)
                        if ch is not None:
                            self.framing_errors = max(0, self.framing_errors - 1)
                            if ch == '\a': self._emit('<BEL>')
                            else: self._emit(ch)

                    self.state = 'WAIT_STOP'

            elif self.state == 'WAIT_STOP':
                if bit == 1:
                    self.state = 'IDLE'
                    self._good += 1
                    self._miss = 0
                    self._idle_marks = 0
                    self.framing_errors = 0
                    if (not self._flywheel) and self._good >= self.lock_threshold:
                        self._flywheel = True
                        self._emit('[Flywheel: LOCK]\n')
                else:
                    self.framing_errors += 1
                    self._good = 0
                    self._miss += 1
                    if self.framing_errors >= 3:
                        self._reset()
                    else:
                        self.state = 'IDLE'
                    if self._flywheel and self._miss >= self.unlock_threshold:
                        self._flywheel = False
                        self._miss = 0
                        self._fw_used = False
                        self.shift_state = 'LETTERS'
                        self._emit('[Flywheel: HUNT]\n')

        return len(input_items[0])

import numpy as np
import pmt
from gnuradio import gr

class ita2_decoder(gr.sync_block):
    """
    Allgemeingueltiger ITA2/Baudot-Decoder mit Flywheel-Sync und Message-Output.
    """
    # Standard-Tabellen (könnten bei Bedarf auch als Parameter übergeben werden)
    LETS = {0b00100: ' ', 0b10111: 'Q', 0b10011: 'W', 0b00001: 'E', 0b01010: 'R',
            0b10000: 'T', 0b10101: 'Y', 0b00111: 'U', 0b00110: 'I', 0b11000: 'O',
            0b10110: 'P', 0b00011: 'A', 0b00101: 'S', 0b01001: 'D', 0b01101: 'F',
            0b11010: 'G', 0b10100: 'H', 0b01011: 'J', 0b01111: 'K', 0b10010: 'L',
            0b10001: 'Z', 0b11101: 'X', 0b01110: 'C', 0b11110: 'V', 0b11001: 'B',
            0b01100: 'N', 0b11100: 'M', 0b01000: '\r', 0b00010: '\n'}

    FIGS = {0b00100: ' ', 0b10111: '1', 0b10011: '2', 0b00001: '3', 0b01010: '4',
            0b10000: '5', 0b10101: '6', 0b00111: '7', 0b00110: '8', 0b11000: '9',
            0b10110: '0', 0b00011: '-', 0b00101: "'", 0b01001: '$', 0b01101: '!',
            0b11010: '&', 0b10100: '#', 0b01011: '\a', 0b01111: '(', 0b10010: ')',
            0b10001: '+', 0b11101: '/', 0b01110: ':', 0b11110: ';', 0b11001: '?',
            0b01100: ',', 0b11100: '.', 0b01000: '\r', 0b00010: '\n'}

    def __init__(self, lock_threshold=5, unlock_threshold=8, unshift_on_space=False, gate_default=True):
        gr.sync_block.__init__(self,
            name='Baudot Decoder (Flywheel)',
            in_sig=[np.int8],
            out_sig=None) # Reine Message-basierte Ausgabe

        # Parameter registrieren
        self.lock_threshold = lock_threshold
        self.unlock_threshold = unlock_threshold
        self.unshift_on_space = unshift_on_space

        # Ports
        self.message_port_register_in(pmt.intern('gate'))
        self.set_msg_handler(pmt.intern('gate'), self._handle_gate)
        self.message_port_register_out(pmt.intern('text'))

        # Interner Zustand
        self._gate_open = bool(gate_default)
        self._reset()
        self._good = 0
        self._miss = 0
        self._flywheel = False

    def _emit(self, s):
        # Nachricht via Message-Port publizieren
        self.message_port_pub(pmt.intern('text'), pmt.to_pmt(s))

    def _handle_gate(self, msg):
        try:
            if pmt.is_pair(msg): msg = pmt.cdr(msg)
            new_open = bool(pmt.to_python(msg))
        except Exception: return
        if self._gate_open and not new_open:
            self._reset()
            self._flywheel = False
            self._good = 0
            self._miss = 0
        self._gate_open = new_open

    def _reset(self):
        self.state = 'IDLE'
        self.bit_buffer = []
        self.shift_state = 'LETTERS'
        self.framing_errors = 0
        self._fw_used = False
        self._idle_marks = 0

    def work(self, input_items, output_items):
        if not self._gate_open:
            return len(input_items[0])

        for bit in input_items[0]:
            bit = int(bit)

            if self.state == 'IDLE':
                if bit == 0:
                    self._fw_used = False
                    self._idle_marks = 0
                    self.state = 'DATA'
                    self.bit_buffer = []
                else:
                    self._idle_marks += 1
                    if self._flywheel and not self._fw_used and self._idle_marks >= 2:
                        self._fw_used = True
                        self._idle_marks = 0
                        self.state = 'DATA'
                        self.bit_buffer = []

            elif self.state == 'DATA':
                self.bit_buffer.append(bit)
                if len(self.bit_buffer) == 5:
                    code = sum(b << i for i, b in enumerate(self.bit_buffer))

                    if code == 0b11111:
                        self.shift_state = 'LETTERS'
                        self.framing_errors = 0
                    elif code == 0b11011:
                        self.shift_state = 'FIGURES'
                        self.framing_errors = 0
                    elif code == 0b00000:
                        pass
                    else:
                        # Unshift on Space Logik integrieren
                        if code == 0b00100 and self.unshift_on_space:
                            self.shift_state = 'LETTERS'

                        table = self.LETS if self.shift_state == 'LETTERS' else self.FIGS
                        ch = table.get(code)
                        if ch is not None:
                            self.framing_errors = max(0, self.framing_errors - 1)
                            if ch == '\a': self._emit('<BEL>')
                            else: self._emit(ch)

                    self.state = 'WAIT_STOP'

            elif self.state == 'WAIT_STOP':
                if bit == 1:
                    self.state = 'IDLE'
                    self._good += 1
                    self._miss = 0
                    self._idle_marks = 0
                    self.framing_errors = 0
                    if (not self._flywheel) and self._good >= self.lock_threshold:
                        self._flywheel = True
                        self._emit('[Flywheel: LOCK]\n')
                else:
                    self.framing_errors += 1
                    self._good = 0
                    self._miss += 1
                    if self.framing_errors >= 3:
                        self._reset()
                    else:
                        self.state = 'IDLE'
                    if self._flywheel and self._miss >= self.unlock_threshold:
                        self._flywheel = False
                        self._miss = 0
                        self._fw_used = False
                        self.shift_state = 'LETTERS'
                        self._emit('[Flywheel: HUNT]\n')

        return len(input_items[0])

ITA2 Decoder Block

Funktionsprinzip (Theory of Operation)

Parameter (Konfiguration)

Schnittstellen (Ports)

Die Zustandsmaschine (State Machine)

Best Practices für den GNU Radio Companion (GRC)

Implementierung

Verarbeitung im Block

Stream Tags

Testen mit GRC

DDK9-RTTY-Demodulation