pyserial - Comment lire la dernière ligne envoyé à partir d'un périphérique série

Question

J'ai un Arduino connecté à mon ordinateur exécutant une boucle, l'envoi d'une valeur sur le port série à l'ordinateur toutes les 100 ms.

Je veux faire un script Python qui va lire à partir du port série que tous les quelques secondes, donc je veux voir que la dernière chose envoyée par l'Arduino.

Comment faites-vous cela dans pyserial?

Voici le code que j'ai essayé qui does't travail. Il lit les lignes séquentiellement.

import serial
import time

ser = serial.Serial('com4',9600,timeout=1)
while 1:
    time.sleep(10)
    print ser.readline() #How do I get the most recent line sent from the device?

Answer 1

Peut-être que je suis malentendu à votre question, mais comme il est une ligne série, vous devrez lire tout envoyé de la séquence Arduino -. Il va tamponner dans l'Arduino jusqu'à ce que vous lisez

Si vous voulez avoir un indicateur d'état qui montre la dernière chose envoyée -. Utiliser un fil qui intègre le code dans votre question (moins le sommeil), et de garder la dernière ligne complète lire comme la dernière ligne de l'Arduino

Mise à jour: mtasic code exemple de est assez bonne, mais si l'Arduino a envoyé une ligne partielle lorsque inWaiting() est appelée, vous obtiendrez une ligne tronquée. Au lieu de cela, ce que vous voulez faire est de mettre la dernière Ligne dans last_received et garder la ligne partielle buffer de sorte qu'il peut être ajouté à la prochaine fois autour de la boucle. Quelque chose comme ceci:

def receiving(ser):
    global last_received

    buffer_string = ''
    while True:
        buffer_string = buffer_string + ser.read(ser.inWaiting())
        if '\n' in buffer_string:
            lines = buffer_string.split('\n') # Guaranteed to have at least 2 entries
            last_received = lines[-2]
            #If the Arduino sends lots of empty lines, you'll lose the
            #last filled line, so you could make the above statement conditional
            #like so: if lines[-2]: last_received = lines[-2]
            buffer_string = lines[-1]

En ce qui concerne l'utilisation de readline(): Voici ce que la documentation pyserial a à dire (légèrement modifié pour plus de clarté et une mention à readlines ()):

  

Soyez prudent lorsque vous utilisez "readline". Faire   spécifier un temps d'arrêt lors de l'ouverture du   port série, sinon il pourrait bloquer   jamais si aucun caractère de nouvelle ligne est   reçu. Notez également que « readlines () »   ne fonctionne qu'avec un délai d'attente. Il   dépend d'avoir un délai d'attente et   interprète cela comme EOF (fin de fichier).

qui me semble tout à fait raisonnable!

Answer 2

from serial import *
from threading import Thread

last_received = ''

def receiving(ser):
    global last_received
    buffer = ''

    while True:
        # last_received = ser.readline()
        buffer += ser.read(ser.inWaiting())
        if '\n' in buffer:
            last_received, buffer = buffer.split('\n')[-2:]

if __name__ ==  '__main__':
    ser = Serial(
        port=None,
        baudrate=9600,
        bytesize=EIGHTBITS,
        parity=PARITY_NONE,
        stopbits=STOPBITS_ONE,
        timeout=0.1,
        xonxoff=0,
        rtscts=0,
        interCharTimeout=None
    )

    Thread(target=receiving, args=(ser,)).start()

Answer 3

Ces solutions hog la CPU en attendant caractères.

Vous devez faire au moins un appel de blocage à lire (1)

while True:
    if '\n' in buffer: 
        pass # skip if a line already in buffer
    else:
        buffer += ser.read(1)  # this will block until one more char or timeout
    buffer += ser.read(ser.inWaiting()) # get remaining buffered chars

... et faire la chose split comme avant.

Answer 4

Vous pouvez utiliser ser.flushInput() pour débusquer toutes les données série qui est actuellement dans le tampon.

Après avoir effacé les anciennes données, vous pouvez ser.readline utilisateur () pour obtenir les données les plus récentes du périphérique série.

Je pense que son un peu plus simple que les autres solutions proposées ici. A travaillé pour moi, espérons qu'il est approprié pour vous.

Answer 5

Cette méthode vous permet de contrôler séparément le délai d'attente pour recueillir toutes les données pour chaque ligne, et un délai d'attente différent d'attente sur les lignes supplémentaires.

# get the last line from serial port
lines = serial_com()
lines[-1]              

def serial_com():
    '''Serial communications: get a response'''

    # open serial port
    try:
        serial_port = serial.Serial(com_port, baudrate=115200, timeout=1)
    except serial.SerialException as e:
        print("could not open serial port '{}': {}".format(com_port, e))

    # read response from serial port
    lines = []
    while True:
        line = serial_port.readline()
        lines.append(line.decode('utf-8').rstrip())

        # wait for new data after each line
        timeout = time.time() + 0.1
        while not serial_port.inWaiting() and timeout > time.time():
            pass
        if not serial_port.inWaiting():
            break 

    #close the serial port
    serial_port.close()   
    return lines

Answer 6

Vous aurez besoin d'une boucle pour lire tout envoyé, avec le dernier appel à readline () le blocage jusqu'à ce que le délai d'attente. Donc:

def readLastLine(ser):
    last_data=''
    while True:
        data=ser.readline()
        if data!='':
            last_data=data
        else:
            return last_data

Answer 7

Une légère modification au code de mtasic & Vinay Sajip:

Alors que j'ai trouvé ce code très utile pour moi une application similaire, je avais besoin tous les lignes qui reviennent d'un périphérique série qui enverrait des informations périodiquement.

J'ai choisi de pop le premier élément du haut, l'enregistrer, puis rejoindre les autres éléments que le nouveau tampon et continuer à partir de là.

Je me rends compte que cela est pas ce que Greg demandait, mais je pensais qu'il était utile de partager comme une note de côté.

def receiving(ser):
    global last_received

    buffer = ''
    while True:
        buffer = buffer + ser.read(ser.inWaiting())
        if '\n' in buffer:
            lines = buffer.split('\n')
            last_received = lines.pop(0)

            buffer = '\n'.join(lines)

Answer 8

Utilisation .inWaiting() dans une boucle infinie peut être problématique. Il peut monopoliser l'ensemble CPU en fonction de la mise en œuvre. Au lieu de cela, je vous conseille d'utiliser une taille spécifique de données à lire. Donc, dans ce cas, ce qui suit devrait se faire par exemple:

ser.read(1024)

Answer 9

Trop de complications

Quelle est la raison de diviser l'objet octets par saut de ligne ou par d'autres manipulations de tableau? Je vous écris la méthode la plus simple, qui va résoudre votre problème:

import serial
s = serial.Serial(31)
s.write(bytes("ATI\r\n", "utf-8"));
while True:
    last = ''
    for byte in s.read(s.inWaiting()): last += chr(byte)
    if len(last) > 0:
        # Do whatever you want with last
        print (bytes(last, "utf-8"))
        last = ''

Answer 10

Voici un exemple en utilisant une enveloppe qui vous permet de lire la ligne la plus récente sans CPU 100%

class ReadLine:
    """
    pyserial object wrapper for reading line
    source: https://github.com/pyserial/pyserial/issues/216
    """
    def __init__(self, s):
        self.buf = bytearray()
        self.s = s

    def readline(self):
        i = self.buf.find(b"\n")
        if i >= 0:
            r = self.buf[:i + 1]
            self.buf = self.buf[i + 1:]
            return r
        while True:
            i = max(1, min(2048, self.s.in_waiting))
            data = self.s.read(i)
            i = data.find(b"\n")
            if i >= 0:
                r = self.buf + data[:i + 1]
                self.buf[0:] = data[i + 1:]
                return r
            else:
                self.buf.extend(data)

s = serial.Serial('/dev/ttyS0')
device = ReadLine(s)
while True:
    print(device.readline())