[PYTHON] Serielle Kommunikation mit Raspberry Pi + PySerial

Einführung

Ich wollte seriell mit Raspberry Pi kommunizieren, also habe ich versucht, über die serielle USB-Konvertierung zu kommunizieren. Ich bezog mich auf "Ich habe versucht, serielle Kommunikation mit Raspberry Pi". Der Empfangsverarbeitung wurde ein Empfangszeitlimit hinzugefügt, unter der Annahme, dass die Verarbeitung in der Reihenfolge Befehlsübertragung → Befehlsempfang ausgeführt wird.

Umgebung

• Raspberry Pi3
• Python 3.7.2
• pySerial 3.4

Installation

• Geben Sie den folgenden Befehl ein, um pySerial zu installieren.
• Die Installation von pipenv entfällt.

$ pipenv install pyserial

USB serielle Konvertierung

Ich habe [serielles FT232-USB-Konvertierungskabel] verwendet (http://akizukidenshi.com/catalog/g/gM-08343/). Wenn es mit dem FT232-Chip ausgestattet ist, scheint es ohne Hinzufügen eines Treibers erkannt zu werden.

Überprüfen Sie das Verbindungsziel mit dem folgenden Befehl.

$ ls -la /dev/ttyUSB*

Wenn es sich um die erste Einheit handelt, lautet das Verbindungsziel '/ dev / ttyUSB0'.

Hinweis: Verwenden Sie für die Kommunikation mit Rasberry Pi auf einem PC (Windows / Mac / Linux) ein Querkabel. Da Rasberry Pi auch ein PC ist, gibt es ein Querkabel zwischen PCs, und die Verbindung zu Kommunikationsgeräten ist (häufig) gerade.

Code

Serielle Kommunikationsklasse

• Es wird davon ausgegangen, dass die Verarbeitung in der Reihenfolge Befehlsübertragung → Befehlsempfang erfolgt. (Mit Timeout)
• Sie können die serielle Schnittstelle explizit öffnen / schließen. (Weil es für die persönliche Umsetzung notwendig war)

sample.py

# -*- coding: utf-8 -*-

import serial
import time
import threading

"""
Serielle Kommunikationsklasse
"""
class SampleComm:
    #Initialisieren
    def __init__(self):
        #Flagge öffnen
        self.isPortOpen = False
        #empfangene Daten
        self.recvData = bytearray()
        #Ereignisgenerierung
        self.event = threading.Event()

    #Warten auf Datenempfang(Mit Timeout[sec])
    def recv(self, timeout=3):
        #Holen Sie sich Zeit für eine Auszeit
        time_start = time.time()
        time_end = time_start
        #Ereignis "Thread warten" löschen
        self.event.clear()
        #Empfangene Daten löschen
        self.recvData.clear()
        #Erhaltenes Ergebnis True:Erfolg falsch:Fehler(Auszeit)
        result = False

        #Warten auf Datenempfang
        while not self.event.is_set():
            #Timeout-Prüfung
            time_end = time.time()
            if (time_end - time_start > timeout):
                #Datenübertragung / -empfang gestoppt und fehlgeschlagen(Auszeit)Zu
                result = False
                self.stop()
                print("timeout:{0}sec".format(timeout))
                break

            #Empfangene Daten lesen
            buff = self.comm.read()

            #Erhaltene Datenbeurteilung
            if len(buff) > 0:
                #Empfangene Daten hinzufügen
                self.recvData.extend(buff)
                # (Temporär)Erfolg, wenn \ n empfangen wurde
                if (self.recvData.find(b'\n')) >= 0:
                    #Stoppen Sie das Senden und Empfangen von Daten und machen Sie es erfolgreich
                    result = True
                    self.stop()
                    break

        #Gibt das Ergebnis zurück
        return (result, self.recvData)

    #Datenübertragung
    def send(self, data):
        self.comm.write(data)

    #Stoppen Sie das Senden und Empfangen von Daten
    def stop(self):
        self.event.set()

    #Cyril Port offen
    def open(self, tty, baud='115200'):
        try:
            self.comm = serial.Serial(tty, baud, timeout=0.1)
            self.isPortOpen = True
        except Exception as e:
            self.isPortOpen = False

        return self.isPortOpen

    #Serielle Schnittstelle geschlossen(Explizit schließen)
    def close(self):
        self.stop()
        if (self.isPortOpen):
            self.comm.close()
        self.isPortOpen = False

if __name__ == "__main__":
    #Seriennummer öffnen
    comm = SampleComm()
    comm.open('/dev/ttyUSB0', '115200')

    #Datenübertragung
    comm.send('test'.encode())
    #Datenempfang(Auszeit=10sec)
    result, data = comm.recv(10)
    print(result)
    print(data)

    #Seriennummer schließen
    comm.close();

wie benutzt man

In pySerial Version 2.5 und höher lautet das Argument von write () bytearray (). Wenn Sie eine Zeichenfolge senden möchten, führen Sie die entsprechende Codierung () aus. str.encode () ist standardmäßig 'utf-8'.

#Seriennummer öffnen
comm = SampleComm()
comm.open('/dev/ttyUSB0', '115200')

#Datenübertragung
comm.send('sample'.encode())
#Datenempfang(Auszeit=10sec)
result, data = comm.recv(10)
print(result)
print(data)

#Seriennummer schließen
comm.close();

Sie können den Vorgang auch überprüfen, indem Sie sample.py ausführen.

$ python sample.py

abschließend

Die Kommunikation mit dem Gerät wird häufig in der Reihenfolge Befehlsübertragung → Befehlsempfang verarbeitet, und ich habe es gemacht, weil ich den Empfangsprozess mit einer Zeitüberschreitung ausführen wollte. Es ist notwendig, das Urteil über den Abschluss des Empfangs entsprechend zu ändern. Ich hoffe, dass es für ähnliche Fälle in der seriellen Kommunikation hilfreich sein wird.