Python port série port d'écoute

J'ai commencé à écrire du code à l'aide de PySerial d'envoyer et de recevoir des données à un périphérique série.

Jusqu'à maintenant, j'ai seulement travaillé sur le lancement d'une transaction à partir d'un terminal et d'obtenir une réponse à partir de la série de l'appareil.

pseudo:

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

Maintenant, je voudrais mettre en œuvre un listener sur le port dans le cas où le numéro de série de l'appareil est celui de l'initiation de la communication. Également à supprimer les time_delay qui pourrait finir vers le haut de façon trop longtemps, ou pas assez longtemps pour que le périphérique série de répondre.

state = idle
register_interrupt(serial_device, read_response())

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

Jusqu'à présent, je n'ai pas trouvé de PySerial code qui utilise une interruption de communiquer.

J'ai juste aucune idée de comment mettre en place une interruption d'aucune sorte en python, comment

InformationsquelleAutor user3817250 | 2014-09-25
  1. 5

    Il est possible d'utiliser le select module pour sélectionner un serial connexion:

    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)

    Ci-dessous est un exemple complet qui crée un pseudo-tty dans le thread principal. Le thread principal écrit les données sur le pty dans des intervalles de 5 secondes. Un enfant thread est créé qui tente utilise select.select à attendre pour lire les données, ainsi que l'aide de la normale serial.Serial.read méthode pour lire les données de l'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 utilise en interne sélectionnez la méthode read (si vous êtes sur un système posix), en utilisant le délai prévu. Cependant, il va tenter de lire NUM octets jusqu'à ce NUM total d'octets ont été lus, ou jusqu'à ce que le délai d'attente a été atteint. C'est pourquoi l'exemple explicitement lit len("GOODBYE FROM MAIN THREAD") (24 octets) du pty, de sorte que le conn.read appel retourner immédiatement au lieu d'attendre pour l'ensemble de délai d'attente.

    TL;DR Il est possible d'utiliser serial.Serial connexions avec select.select, qui est ce qui, je crois, vous êtes à la recherche pour.

    • Selon mon expérience, il est bon de mettre un très petit délai d'attente sur les serial.Serial exemple, disons 1 ms. En fonction de l'appareil de transmission, les données peuvent arriver un peu plus lent et avec un délai d'expiration=0 vous pouvez terminer la lecture de presque octet par octet, donc en boucle / de la sélection d'un lot avec un notable de la consommation CPU de l'incrément.
    • Lorsque vous essayez cette solution sur mon environnement (Python 2.7.15, avec la version actuelle de PySerial de Déc. 2018), l'abréviation exemple lève l'exception io.UnsupportedOperation: fileno sur la ligne avec le select déclaration. Il fonctionne normalement, mais quand je suis tout simplement utiliser la read() méthode avec un délai d'attente de 1 seconde. Je ne sais pas quel est le problème, mais il me semble que vous pouvez tout simplement faire un enfant avec du fil d'une boucle infinie et un .read() avec un bref délai, ni de select est nécessaire.
    • PS - j'ai testé cette approche de l'enfant (thread, boucle infinie avec juste un read(), et un bref délai) dans mon application, et cela a fonctionné parfaitement.
    InformationsquelleAutor James Johnson