Python-serial-port-listener

Habe ich begonnen, einige code schreiben mit PySerial zum senden und empfangen von Daten an ein serielles Gerät.

Bis jetzt habe ich nur gearbeitet, auf die Initiierung einer Transaktion an einem terminal und dem empfangen einer Antwort von dem seriellen Gerät.

pseudo:

main:
    loop:
        message = get_message()
        send_to_serial(message)
        time_delay(1 second)
        read_response()

Nun möchte ich zum implementieren eines listener an diesem port in dem Fall, dass das serielle Gerät ist die eine Initiierung der Kommunikation. Auch zum entfernen der time_delay das könnte am Ende viel zu lange, oder nicht lange genug für das serielle Gerät, um zu Antworten.

state = idle
register_interrupt(serial_device, read_response())

main:
    loop:
        message = get_message()
        state = sending
        send_to_serial(message)

Bisher habe ich noch keine gefunden PySerial-code, verwendet einen interrupt zu kommunizieren.

Ich habe nur keine Ahnung, wie man ein interrupt jeder Art in python, wie würde

InformationsquelleAutor user3817250 | 2014-09-25

  1. 5

    Ist es möglich, die select - Modul wählen Sie auf einem serial Verbindung:

    import serial
import select

timeout = 10
conn = serial.Serial(serial_name, baud_rate, timeout=0)
read,_,_ = select.select([conn], [], [], timeout)
read_data = conn.read(0x100)

    Unten ist ein vollständiges Beispiel, das schafft eine pseudo-tty-Schnittstelle in den Haupt-thread. Der main-thread schreibt die Daten, um die pty-Funktion in 5-Sekunden-Intervallen. Ein Kind, thread ist erstellt, dass die versuche verwendet select.select zu warten, für Lesen von Daten, als auch mit den normalen serial.Serial.read Methode zum Lesen von Daten aus der pty:

    #!/bin/bash

import os
import pty
import select
import serial
import sys
import threading
import time

def out(msg):
    print(msg)
    sys.stdout.flush()

# child process
def serial_select():
    time.sleep(1)

    out("CHILD THREAD: connecting to serial {}".format(serial_name))
    conn = serial.Serial(serial_name, timeout=0)
    conn.nonblocking()

    out("CHILD THREAD: selecting on serial connection")
    avail_read, avail_write, avail_error = select.select([conn],[],[], 7)
    out("CHILD THREAD: selected!")

    output = conn.read(0x100)
    out("CHILD THREAD: output was {!r}".format(output))

    out("CHILD THREAD: normal read serial connection, set timeout to 7")
    conn.timeout = 7

    # len("GOODBYE FROM MAIN THREAD") == 24
    # serial.Serial.read will attempt to read NUM bytes for entire
    # duration of the timeout. It will only return once either NUM
    # bytes have been read, OR the timeout has been reached
    output = conn.read(len("GOODBYE FROM MAIN THREAD"))
    out("CHILD THREAD: read data! output was {!r}".format(output))

master, slave = pty.openpty()
serial_name = os.ttyname(slave)

child_thread = threading.Thread(target=serial_select)
child_thread.start()

out("MAIN THREAD: sleeping for 5")
time.sleep(5)

out("MAIN THREAD: writing to serial")
os.write(master, "HELLO FROM MAIN THREAD")

out("MAIN THREAD: sleeping for 5")
time.sleep(5)

out("MAIN THREAD: writing to serial")
os.write(master, "GOODBYE FROM MAIN THREAD")

child_thread.join()

    PySerial intern verwendet Sie die Methode Lesen (wenn man auf einem posix-system), mit dem timeout zur Verfügung gestellt. Es wird jedoch versucht zu Lesen NUM bytes, bis NUM Gesamtanzahl bytes, die gelesen wurden, oder bis der timeout erreicht wurde. Dies ist der Grund, warum das Beispiel explizit liest len("GOODBYE FROM MAIN THREAD") (24 bytes) von den pty, so dass die conn.read nennen würde sofort zurückkehren, anstatt zu warten, für die gesamte timeout.

    TL;DR Es ist möglich, verwenden Sie serial.Serial verbindungen mit select.select, das ist, was, glaube ich, die Sie suchen.

    Nach meiner Erfahrung ist es gut, einen sehr kleinen timeout auf der serial.Serial Beispiel, sagen wir einmal 1 ms. Abhängig von dem sendenden Gerät, die Daten gelangen kann, ein wenig langsamer und mit einem timeout=0 kann man am Ende der Lektüre fast byte für byte, daher Schleife / Auswahl von viele mit einer spürbaren CPU-Verbrauch zu erhöhen.
    Wenn Sie versuchen, diese Lösung auf meine Umgebung (Python 2.7.15, mit der aktuellen version von PySerial Stand Dez. 2018), die gekürzte Beispiel wirft die Ausnahme io.UnsupportedOperation: fileno auf der Linie mit der select - Anweisung. Es funktioniert normal, aber wenn ich einfach die read() - Methode mit einem timeout von 1 Sekunde. Ich bin mir nicht sicher, was das problem ist, aber es scheint mir, dass Sie nur ein Kind-thread mit einer Endlosschleife und ein .read() mit einem kurzen timeout -- und nicht select benötigt wird.
    PS - getestet habe ich diesen Ansatz (child-thread, infinite loop mit nur einem read(), und eine kurze timeout) in meiner eigenen Anwendung, und es funktionierte perfekt.

    InformationsquelleAutor James Johnson

Schreibe einen Kommentar