我有一个连接到我的计算机的 Arduino，它运行一个循环，每 100 毫秒通过串行端口将一个值发送回计算机。
我想制作一个 Python 脚本，每隔几秒钟从串口读取一次，所以我希望它只看到从 Arduino 发送的最后一个东西。
你如何在 Pyserial 中做到这一点？
这是我尝试过的代码，但它不起作用。它按顺序读取行。
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?
【问题讨论】：
标签： python serial-port arduino pyserial
也许我误解了您的问题，但由于它是串行线路，您必须按顺序读取从 Arduino 发送的所有内容 - 在您读取之前，它将在 Arduino 中缓冲。
如果您想要显示最新发送的状态显示 - 使用包含问题中代码的线程（减去睡眠），并将最后一个完整行读取为 Arduino 的最新行。
更新： mtasic 的示例代码相当不错，但如果调用 inWaiting() 时 Arduino 发送了部分行，则会得到截断的行。相反，您要做的是将最后 complete 行放入last_received，并将部分行保留在buffer 中，以便可以在下一次循环中附加它。像这样的：
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]
关于readline() 的使用：以下是 Pyserial 文档的内容（为清楚起见，稍作编辑并提及 readlines()）：
使用“readline”时要小心。做 打开时指定超时 串口，否则可能会阻塞 如果没有换行符，则永远 已收到。另请注意“readlines()” 仅适用于超时。它 取决于有一个超时和 将其解释为 EOF（文件结尾）。
这对我来说似乎很合理！
【讨论】：
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()
【讨论】：
mtasic 的代码看起来不错，除了我认为的一个小故障。
您可以使用ser.flushInput() 清除当前在缓冲区中的所有串行数据。
清除旧数据后，您可以使用 ser.readline() 从串口设备获取最新数据。
我认为它比这里提出的其他解决方案要简单一些。为我工作，希望它适合你。
【讨论】：
这些解决方案会在等待字符时占用 CPU。
您应该至少对 read(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
...像以前一样做拆分的事情。
【讨论】：
此方法允许您单独控制收集每行所有数据的超时时间，以及等待其他行的不同超时时间。
# 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
【讨论】：
您将需要一个循环来读取发送的所有内容，最后一次调用 readline() 会阻塞直到超时。所以：
def readLastLine(ser):
last_data=''
while True:
data=ser.readline()
if data!='':
last_data=data
else:
return last_data
【讨论】：
对 mtasic 和 Vinay Sajip 的代码稍作修改：
虽然我发现这段代码对于类似的应用程序对我很有帮助，但我需要所有从串行设备返回的行，该串行设备会定期发送信息。
我选择从顶部弹出第一个元素，记录它，然后将其余元素重新加入作为新缓冲区并从那里继续。
我知道这不是 Greg 所要求的不是，但我认为值得作为旁注分享。
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)
【讨论】：
在无限循环中使用.inWaiting() 可能会出现问题。它可能会占用整个CPU，具体取决于实现。相反，我建议使用特定大小的数据来读取。因此，在这种情况下，例如应该执行以下操作：
ser.read(1024)
【讨论】：
并发症太多
用换行符或其他数组操作分割字节对象的原因是什么？ 我写了最简单的方法，可以解决你的问题：
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 = ''
【讨论】：
这是一个使用包装器的示例，它允许您在没有 100% CPU 的情况下读取最新行
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())
【讨论】：