pyserial - Cómo leer la última línea enviada desde un dispositivo serial

Question

Tengo un Arduino conectado a mi computadora ejecutando un bucle, enviando un valor sobre el puerto serie a la computadora cada 100   ms.

Quiero hacer una secuencia de comandos de Python que se leerá desde el puerto serie cada pocos segundos, por lo que quiero que vea lo último enviado desde el Arduino.

¿Cómo se hace esto en Pyserial?

Aquí está el código que probé que no funciona. Lee las líneas secuencialmente.

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

Tal vez estoy malentendido su pregunta, pero ya que es una línea serie, que tendrá que leer todo lo enviado desde la secuencialmente Arduino - que va a ser amortiguada en el Arduino hasta que haya leído eso.

Si desea tener una pantalla de estado que muestre lo último enviado, use un hilo que incorpore el código en su pregunta (menos la suspensión) y mantenga la última línea completa leída como la última línea del Arduino.

Actualización:mtasic 's código de ejemplo es bastante bueno, pero si el Arduino ha enviado una línea parcial cuando inWaiting() se llama, obtendrá una línea truncada. En su lugar, lo que desea hacer es colocar la última línea en last_received, y mantener la línea parcial en buffer para que pueda adjuntarse a la próxima vez en el ciclo. Algo como esto:

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 cuanto al uso de readline(): Esto es lo que la documentación PySerial tiene que decir (ligeramente editado para mayor claridad y con una mención a readlines()):

Tenga cuidado al usar " readline ". Do especifique un tiempo de espera al abrir el puerto serie , de lo contrario podría bloquear para siempre si no se recibe ningún carácter de nueva línea . También tenga en cuenta que "readlines()" solo funciona con un tiempo de espera excedido. Es depende de tener un tiempo de espera y lo interpreta como EOF (fin de archivo).

que me parece bastante razonable!

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

Necesitará un ciclo para leer todo lo enviado, con la última llamada al bloqueo readline() hasta el tiempo de espera. Por lo tanto:

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

Answer 4

modificación leve a mtasic código de & Vinay Sajip:

Mientras que encontré este código muy útil para mí para una aplicación similar, que necesitaba todos las líneas que regresan de un dispositivo serie que enviaría información periódicamente.

Opté por hacer estallar el primer elemento de la parte superior, registrarlo, y luego volver a unir los elementos restantes como el nuevo búfer y continuar desde allí.

Me doy cuenta de que esto es no lo que Greg estaba pidiendo, pero pensé que valía la pena compartirlo como una nota al margen.

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 5

Estas soluciones engancharán la CPU mientras esperan caracteres.

Usted debe hacer al menos una llamada de bloqueo para leer (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 

... y hacer lo dividida que antes.

Answer 6

El uso de .inWaiting() dentro de un bucle infinito puede ser problemático. Puede acaparar todo CPU dependiendo de la implementación. En cambio, recomendaría usar un tamaño específico de datos para leer. Así que en este caso el siguiente debe hacerse, por ejemplo:

ser.read(1024) 

Answer 7

Este método le permite controlar por separado el tiempo de espera para la recopilación de todos los datos para cada línea, y un tiempo de espera diferente para esperar en líneas adicionales.

# 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 8

Puede utilizar ser.flushInput() para eliminar todos los datos en serie que se encuentra actualmente en el búfer.

Después de borrar los datos anteriores, puede utilizar ser.readline() para obtener los datos más recientes del dispositivo serie.

Creo que es un poco más simple que las otras soluciones propuestas aquí. Trabajado para mí, espero que sea adecuado para ti.

complicaciones

Answer 9

Demasiado

¿Cuál es la razón para dividir el objeto bytes por salto de línea o por otras manipulaciones matriz? escribo el método más simple, que va a resolver su problema:

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 = ''