SOLT-Serielles Schnittstellenprotokoll – Empfangen und Dekodieren von Daten
Einführung
Entwickler fordern häufig die Integration von SOLT-Geräten und die Bereitstellung einer API, scheuen sich aber gleichzeitig aufgrund mangelnder Erfahrung vor der Arbeit mit seriellen Schnittstellen. Dabei ist die Arbeit mit einer COM-Schnittstelle (seriell) eigentlich recht einfach und intuitiv. In manchen Fällen ist die Implementierung der Datenerfassung im Programm deutlich komfortabler und effizienter als die Einrichtung und Wartung eines Zwischenservers mit API.
Lasst uns herausfinden, wie das geht – es dauert nur 5-10 Minuten!
Der Knopf bzw. das Ruffeld ist ein Funksender, und das SOLT SR5-MPRT-Gerät dient als Empfänger/Modem. Das SR5-MPRT-Modem ist ein Gerät mit RS232-Schnittstelle und in zwei Versionen erhältlich: mit integriertem USB-Konverter und USB-A-Anschluss sowie mit DB9-Anschluss. Im Prinzip handelt es sich um identische Geräte mit denselben Protokollen.
Jedes Gerät (Fernbedienung, SOLT-Taste) verfügt über einen werkseitig programmierten, eindeutigen Code, der die eindeutige Identifizierung der Signalquelle ermöglicht. Dieser Code wird als remote_id bezeichnet und in einem Datenpaket übertragen (Beispiel: 41 31 42 32 43 33 („A1B2C3“)).
Neben der eindeutigen Geräte-ID enthält jede Fernbedienung eine oder mehrere Tasten, deren Drücken den entsprechenden Code sendet. Die Tastencodes bezeichnen verschiedene Aktionen:
Ruftasten |
Verhexen |
Abbrechen-Schaltflächen |
Verhexen |
| 1 | 0x14 | 2 | 0x24 |
| 3 | 0x28 | 4 | 0x34 |
| 5 | 0x38 | 6 | 0x44 |
| 7 | 0x48 | 8 | 0x54 |
| 9 | 0x58 | 10 | 0x64 |
| 11 | 0x68 | 12 | 0x74 |
| 13 | 0x78 | 14 | 0x84 |
| 15 - Abbrechen | 0xE4 | ||
| 16 – Vollständige Stornierung aller Anrufe | 0xF4 |
Die SB9-3XBK-Konsole verfügt beispielsweise über zwei Ruftasten (1 = 14 und 3 = 28) und eine Abbruchtaste (15 = E4).
Die SB7-1PBK-Konsole hat nur eine Ruftaste 1 = 14
SOLT-Protokollbeschreibung
Die Datenübertragung im SOLT-Protokoll erfolgt in Paketen mit einer festen Länge von 19 Byte . Das Paketformat ist wie folgt:
| Byte | Zweck | Beschreibung |
| 1 | Startsymbol (STX) | Definiert den Beginn des Pakets |
| 2-3 | Gedrückter Tastencode | Zwei ASCII-Zeichen zur Kennzeichnung der Schaltfläche |
| 4 | Separator - | Fester Trennsteg |
| 5-10 | Remote-ID | Eindeutige Gerätekennung (ASCII) |
| 11 | Separator - | Fester Trennsteg |
| 12-18 | Nutzerdaten | ASCII-Zeichenkette mit Benutzer-ID |
| 19 | Endsymbol (ETX) | Paketabschluss |
Beispiel eines Pakets in Hexadezimaldarstellung:
02 31 34 2D 41 31 42 32 43 33 2D 31 32 33 34 35 36 37 03
Wo:
- 02 (STX) - der Anfang des Pakets,
- 31 34 ("14") - Ruftastencode,
- 2D ("-") - das erste Trennzeichen,
- 41 31 42 32 43 33 ("A1B2C3") - Geräte-ID (Remote-ID),
- 2D ("-") - das zweite Trennzeichen,
- 31 32 33 34 35 36 37 ("1234567") - Benutzerdaten (diese werden bei der Registrierung der Taste im SR5-MPRT-Modem angegeben)
- 03 (ETX) - das Ende des Gehäusesymbols.
Beispiel eines Python-Programms
Der vollständige Code ist am Ende des Artikels angegeben.
Übrigens: Kopieren Sie einfach diesen gesamten Code in Ihren KI-Assistenten, und er wird ihn optimal an Ihre Entwicklungsumgebung anpassen.
Installation der erforderlichen Bibliotheken
Bevor Sie den Code ausführen, müssen Sie die Bibliotheken für die Arbeit mit dem COM-Port installieren. Verwenden Sie dazu folgenden Befehl:
pip install serial-asyncio
Diese Bibliotheken ermöglichen die Arbeit mit einer seriellen Schnittstelle im asynchronen Modus.
Speichern von Variablen und Porteinstellungen
Zu Beginn des Programms werden Konstanten für die Arbeit mit der seriellen Schnittstelle festgelegt:
import asyncio
import serial_asyncio
# COM port settings
SERIAL_PORT = "COM5" # Specify your port, eg "/dev/ttyUSB0" for Linux
BAUDRATE = 9600 # Baud rate
PACKET_LENGTH = 19 # Expected packet length
Diese Parameter definieren den Port, die Datenrate und die erwartete Paketlänge in Bytes.
SerialReader-Klasse
Zur Verarbeitung eingehender Daten wird die Klasse SerialReader verwendet, die die folgenden Methoden implementiert:
- data_received(self, data): Empfängt eingehende Daten, sammelt sie im Puffer und sendet sie zur Analyse weiter.
- process_packet(self, data): analysiert das empfangene Paket.
class SerialReader(asyncio.Protocol):
def __init__(self):
self.buffer = bytearray()
def data_received(self, data):
"""Process incoming data"""
self.buffer.extend(data)
while len(self.buffer) >= PACKET_LENGTH:
packet = self.buffer[:PACKET_LENGTH]
self.buffer = self.buffer[PACKET_LENGTH:]
self.process_packet(packet) # Daten an die Methode übergeben
def process_packet(self, data):
"""Datenpaket analysieren""
if len(data) != PACKET_LENGTH:
print("Ungültige Paketlänge")
zurückkehren
# Einzelne Teile des Pakets extrahieren
start_symbol = chr(data[0]) # Erstes Byte - Startsymbol
button_number = ''.join(chr(byte) for byte in data[1:3]).strip() # Bytes 2-3 - Buttonnummer
separator1 = chr(data[3]) # 4. Byte - Trennzeichen
remote_id = ''.join(chr(byte) for byte in data[4:10]).strip() # Bytes 5-10 - Remote-ID
separator2 = chr(data[10]) # 11. Byte - zweites Trennzeichen
user_info = ''.join(chr(byte) for byte in data[11:18]).strip() # Bytes 12-18 - Benutzerdaten
end_symbol = chr(data[18]) # 19. Byte - Endsymbol
geparste_daten = (
f'Startsymbol: {start_symbol}, Schaltflächennummer: {button_number}, '
f'Remote-ID: {remote_id}, Benutzerdaten: {user_info}, '
f'Trennzeichen: {separator1}, {separator2}, Endsymbol: {end_symbol}'
)
print(parsed_data)
Nachdem die grundlegenden Einstellungen und die Datenverarbeitungsklasse analysiert wurden, können Sie nun mit dem Öffnen des Ports und dem Empfangen von Daten fortfahren.
Öffnen eines COM-Ports und Empfangen von Daten
Dieses Beispiel verwendet die asynchrone Methode zur Arbeit mit dem COM-Port mithilfe der serial_asyncio-Bibliothek. Dieser Ansatz ermöglicht Folgendes:
- Daten in Echtzeit verarbeiten, ohne den Hauptdatenstrom zu blockieren.
- Arbeiten Sie mit mehreren Verbindungen oder führen Sie parallel andere Aufgaben aus.
- Vermeiden Sie es, das Programm während des Wartens auf Daten einzufrieren.
Wenn die Programmaufgabe jedoch einen einfacheren oder konsistenteren Ansatz erfordert, können Sie die übliche (synchrone) Methode mit der pyserial-Bibliothek verwenden. Dies ist beispielsweise der Fall, wenn Sie eine kleine Datenmenge ohne Hintergrundprozesse verarbeiten müssen.
Beim Start des Programms öffnet sich eine serielle Schnittstelle mit den angegebenen Parametern:
async def main():
"""Start asynchronous reading from the COM port"""
print(f'Opening port {SERIAL_PORT} at baud rate {BAUDRATE}')
loop = asyncio.get_running_loop()
transport, protocol = await serial_asyncio.create_serial_connection(
loop, SerialReader, SERIAL_PORT, BAUDRATE
)
try:
await asyncio.Event().wait() # Endloses Warten
außer asyncio.CancelledError:
passieren
Endlich:
transport.close()
print("Port geschlossen")
Paketanalyse
Nach dem Empfang der Daten wird das Datenpaket analysiert:
def process_packet(self, data):
"""Parse data packet"""
if len(data) != PACKET_LENGTH:
print("Invalid packet length")
return
# Extract individual parts of the packet
start_symbol = chr(data[0]) # First byte - start symbol
button_number = ''.join(chr(byte) for byte in data[1:3]).strip() # Bytes 2-3 - button number
separator1 = chr(data[3]) # 4th byte - separator
remote_id = ''.join(chr(byte) for byte in data[4:10]).strip() # Bytes 5-10 - remote ID
separator2 = chr(data[10]) # 11. Byte - zweites Trennzeichen
user_info = ''.join(chr(byte) for byte in data[11:18]).strip() # Bytes 12-18 - Benutzerdaten
end_symbol = chr(data[18]) # 19. Byte - Endsymbol
geparste_daten = (
f'Startsymbol: {start_symbol}, Schaltflächennummer: {button_number}, '
f'Remote-ID: {remote_id}, Benutzerdaten: {user_info}, '
f'Trennzeichen: {separator1}, {separator2}, Endsymbol: {end_symbol}'
)
print(parsed_data)
Vollständiger Beispielcode
# EXAMPLE OF ASYNCHRONOUS CONNECTION TO A SERIAL PORT
import asyncio
import serial_asyncio
# COM port settings
SERIAL_PORT = "COM5" # Specify your port, eg "/dev/ttyUSB0" for Linux
BAUDRATE = 9600 # Baud rate
PACKET_LENGTH = 19 # Expected packet length
class SerialReader(asyncio.Protocol):
def __init__(self):
self.buffer = bytearray()
def data_received(self, data):
"""Eingehende Daten verarbeiten""
self.buffer.extend(data)
while len(self.buffer) >= PACKET_LENGTH:
packet = self.buffer[:PACKET_LENGTH]
self.buffer = self.buffer[PACKET_LENGTH:]
self.process_packet(packet) # Daten an die Methode übergeben
def process_packet(self, data):
"""Datenpaket analysieren""
if len(data) != PACKET_LENGTH:
print("Ungültige Paketlänge")
zurückkehren
# Einzelne Teile des Pakets extrahieren
start_symbol = chr(data[0]) # Erstes Byte - Startsymbol
button_number = ''.join(chr(byte) for byte in data[1:3]).strip() # Bytes 2-3 - Buttonnummer
separator1 = chr(data[3]) # 4. Byte - Trennzeichen
remote_id = ''.join(chr(byte) for byte in data[4:10]).strip() # Bytes 5-10 - Remote-ID
separator2 = chr(data[10]) # 11. Byte - zweites Trennzeichen
user_info = ''.join(chr(byte) for byte in data[11:18]).strip() # Bytes 12-18 - Benutzerdaten
end_symbol = chr(data[18]) # 19. Byte - Endsymbol
geparste_daten = (
f'Startsymbol: {start_symbol}, Schaltflächennummer: {button_number}, '
f'Remote-ID: {remote_id}, Benutzerdaten: {user_info}, '
f'Trennzeichen: {separator1}, {separator2}, Endsymbol: {end_symbol}'
)
print(parsed_data)
async def main():
"Starte asynchrones Lesen vom COM-Port"
print(f'Öffne Port {SERIAL_PORT} mit Baudrate {BAUDRATE}')
loop = asyncio.get_running_loop()
transport, protocol = await serial_asyncio.create_serial_connection(
Schleife, SerialReader, SERIAL_PORT, BAUDRATE
)
versuchen:
await asyncio.Event().wait() # Endloses Warten
außer asyncio.CancelledError:
passieren
Endlich:
transport.close()
print("Port geschlossen")
if __name__ == "__main__":
asyncio.run(main())