Protocole de port série SOLT - Réception et décodage de données
Introduction
Les développeurs demandent souvent l'intégration de périphériques SOLT et la mise à disposition d'une API, mais hésitent à utiliser un port série par manque d'expérience. Pourtant, l'utilisation d'un port COM (série) est en réalité assez simple et intuitive. Dans certains cas, implémenter soi-même la lecture des données dans le programme est bien plus pratique et efficace que de configurer et de maintenir un serveur intermédiaire avec une API.
Voyons comment faire – cela ne prendra que 5 à 10 minutes !
Le bouton ou panneau d'appel sert d'émetteur radio, et le dispositif SOLT SR5-MPRT fait office de récepteur-modem. Le modem SR5-MPRT est un appareil doté d'une interface RS232, disponible en deux versions : avec convertisseur USB intégré et interface de connexion USB A, ou sans convertisseur, avec interface de connexion DB9. En réalité, il s'agit d'appareils identiques utilisant les mêmes protocoles.
Chaque appareil (télécommande, bouton SOLT) possède un code unique programmé en usine, permettant d'identifier précisément la source du signal. Ce code, appelé remote_id, est transmis dans un paquet de données (exemple : 41 31 42 32 43 33 (« A1B2C3 »)).
Outre l'identifiant unique de l'appareil, chaque télécommande comporte un ou plusieurs boutons dont la pression transmet le code correspondant. Les codes des boutons correspondent à différentes actions :
Boutons d'appel |
Hex |
Boutons Annuler |
Hex |
| 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 - Annuler | 0xE4 | ||
| 16 - Annulation complète de tous les appels | 0xF4 |
Par exemple, la console SB9-3XBK possède 2 boutons d'appel (1 = 14 et 3 = 28) et un bouton d'annulation (15 = E4).
La console SB7-1PBK ne possède qu'un seul bouton d'appel 1 = 14
Description du protocole SOLT
La transmission des données dans le protocole SOLT s'effectue par paquets de longueur fixe de 19 octets . Le format du paquet est le suivant :
| Octet | But | Description |
| 1 | Symbole de départ (STX) | Définit le début du paquet |
| 2-3 | Code du bouton enfoncé | Deux caractères ASCII identifiant le bouton |
| 4 | Séparateur - | Séparateur fixe |
| 5-10 | ID distant | Identifiant unique du périphérique (ASCII) |
| 11 | Séparateur - | Séparateur fixe |
| 12-18 | Données utilisateur | Chaîne ASCII contenant l'identifiant utilisateur |
| 19 | Symbole de fin (ETX) | Complétion du colis |
Exemple de paquet en hexadécimal :
02 31 34 2D 41 31 42 32 43 33 2D 31 32 33 34 35 36 37 03
Où:
- 02 (STX) - le début du paquet,
- 31 34 (« 14 ») - code du bouton d'appel,
- 2D (« - ») - le premier séparateur,
- 41 31 42 32 43 33 (« A1B2C3 ») - ID du périphérique (ID distant),
- 2D ("-") - le deuxième séparateur,
- 31 32 33 34 35 36 37 (« 1234567 ») - données utilisateur (elles sont spécifiées lors de l'enregistrement du bouton dans le modem SR5-MPRT)
- 03 (ETX) - la fin du symbole du paquet.
Exemple de programme Python
Le code complet est fourni à la fin de l'article.
D'ailleurs, il vous suffit de copier ce code complet dans votre assistant IA et il le modifiera parfaitement pour votre environnement de développement.
Installation des bibliothèques requises
Avant d'exécuter le code, vous devez installer les bibliothèques nécessaires au fonctionnement du port COM. Utilisez la commande suivante pour les installer :
pip installer serial-asyncio
Ces bibliothèques permettent de travailler avec un port série en mode asynchrone.
Enregistrement des variables et des paramètres de port
Au début du programme, des constantes sont définies pour le fonctionnement avec le port série :
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
Ces paramètres définissent le port, le débit de données et la longueur attendue des paquets en octets.
Classe SerialReader
Pour traiter les données entrantes, la classe SerialReader est utilisée, qui implémente les méthodes suivantes :
- data_received(self, data) : reçoit les données entrantes, les accumule dans le tampon et les transmet pour analyse.
- process_packet(self, data) : analyse le paquet reçu.
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) # Transmettre les données à la méthode
def process_packet(self, data):
"""Analyser le paquet de données"""
si len(data) != PACKET_LENGTH :
print("Longueur de paquet invalide")
retour
# Extraire les différentes parties du paquet
symbole_début = chr(données[0]) # Premier octet - symbole de début
numéro_bouton = ''.join(chr(byte) for byte in data[1:3]).strip() # Octets 2-3 - numéro du bouton
separator1 = chr(data[3]) # 4e octet - séparateur
remote_id = ''.join(chr(byte) for byte in data[4:10]).strip() # Octets 5-10 - ID distant
separator2 = chr(data[10]) # 11e octet - deuxième séparateur
user_info = ''.join(chr(byte) for byte in data[11:18]).strip() # Octets 12-18 - données utilisateur
symbole_fin = chr(données[18]) # 19e octet - symbole de fin
données_analysées = (
f'Symbole de départ : {start_symbol}, Numéro du bouton : {button_number}, '
ID distant : {remote_id}, Données utilisateur : {user_info}, '
f'Séparateurs : {separator1}, {separator2}, Symbole de fin : {end_symbol}'
)
imprimer(données_analysées)
Maintenant que les paramètres de base et la classe de traitement des données ont été analysés, vous pouvez procéder à l'ouverture du port et à la réception des données.
Ouverture d'un port COM et réception de données
Cet exemple utilise la méthode asynchrone de gestion du port COM à l'aide de la bibliothèque serial_asyncio. Cette approche vous permet de :
- Traiter les données en temps réel sans bloquer le flux principal.
- Gérez plusieurs connexions ou effectuez d'autres tâches en parallèle.
- Évitez de bloquer le programme pendant l'attente des données.
Toutefois, si la tâche du programme requiert une approche plus simple ou plus cohérente, vous pouvez utiliser la méthode classique (synchrone) avec la bibliothèque pyserial. Par exemple, si vous devez traiter une petite quantité de données sans processus en arrière-plan.
Au démarrage du programme, un port série s'ouvre avec les paramètres spécifiés :
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() # Attente infinie
sauf asyncio.CancelledError :
passer
enfin:
transport.fermer()
print("Port fermé")
Analyse des paquets
À réception des données, le paquet est analysé :
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]) # 11e octet - deuxième séparateur
user_info = ''.join(chr(byte) for byte in data[11:18]).strip() # Octets 12-18 - données utilisateur
symbole_fin = chr(données[18]) # 19e octet - symbole de fin
données_analysées = (
f'Symbole de départ : {start_symbol}, Numéro du bouton : {button_number}, '
ID distant : {remote_id}, Données utilisateur : {user_info}, '
f'Séparateurs : {separator1}, {separator2}, Symbole de fin : {end_symbol}'
)
imprimer(données_analysées)
Exemple de code complet
# 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):
"""Traiter les données entrantes"""
self.buffer.extend(données)
tant que len(self.buffer) >= PACKET_LENGTH :
paquet = self.buffer[:PACKET_LENGTH]
self.buffer = self.buffer[PACKET_LENGTH:]
self.process_packet(packet) # Transmettre les données à la méthode
def process_packet(self, data):
"""Analyser le paquet de données"""
si len(data) != PACKET_LENGTH :
print("Longueur de paquet invalide")
retour
# Extraire les différentes parties du paquet
symbole_début = chr(données[0]) # Premier octet - symbole de début
numéro_bouton = ''.join(chr(byte) for byte in data[1:3]).strip() # Octets 2-3 - numéro du bouton
separator1 = chr(data[3]) # 4e octet - séparateur
remote_id = ''.join(chr(byte) for byte in data[4:10]).strip() # Octets 5-10 - ID distant
separator2 = chr(data[10]) # 11e octet - deuxième séparateur
user_info = ''.join(chr(byte) for byte in data[11:18]).strip() # Octets 12-18 - données utilisateur
symbole_fin = chr(données[18]) # 19e octet - symbole de fin
données_analysées = (
f'Symbole de départ : {start_symbol}, Numéro du bouton : {button_number}, '
ID distant : {remote_id}, Données utilisateur : {user_info}, '
f'Séparateurs : {separator1}, {separator2}, Symbole de fin : {end_symbol}'
)
imprimer(données_analysées)
async def main():
"""Démarrez la lecture asynchrone à partir du port COM"""
print(f'Ouverture du port {SERIAL_PORT} au débit en bauds {BAUDRATE}')
boucle = asyncio.get_running_loop()
transport, protocole = await serial_asyncio.create_serial_connection(
boucle, SerialReader, PORT_SÉRIE, DÉBIT_BAUD
)
essayer:
await asyncio.Event().wait() # Attente infinie
sauf asyncio.CancelledError :
passer
enfin:
transport.fermer()
print("Port fermé")
si __name__ == "__main__":
asyncio.run(main())