Python串口侦听器

Question

我已经开始使用PySerial编写一些代码来向串行设备发送和接收数据.

到目前为止,我一直致力于从终端启动事务并从串行设备接收响应.

伪:

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

现在我想在串口设备启动通信的情况下在该端口上实现一个监听器.另外要移除time_delay哪个可能会结束太长时间,或者不足以让串行设备响应.

state = idle
register_interrupt(serial_device, read_response())

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

到目前为止,我还没有找到任何使用中断进行通信的PySerial代码.

我只是不知道如何在python中设置任何类型的中断,怎么会

Answer 1

可以使用select模块在serial连接上进行选择:

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)

下面是一个在主线程中创建伪tty的完整示例.主线程以5秒的间隔将数据写入pty.创建子线程,尝试用于select.select等待读取数据,以及使用常规serial.Serial.read方法从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在内部使用read方法上的select(如果你在posix系统上),使用提供的超时.但是,它将尝试读取NUM个字节,直到读取总共NUM个字节,或者直到达到超时.这就是为什么示例显式len("GOODBYE FROM MAIN THREAD")从pty 读取(24字节),以便conn.read调用立即返回而不是等待整个超时.

TL; DR可以使用serial.Serial连接select.select,我相信你正在寻找.