прослушиватель последовательного порта 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 в основном потоке. Основной поток записывает данные в pty через 5 секунд. Ребенок поток создается, что попытки использует 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 внутренне использует отборный на методе чтения (если вы в системе posix), используя предоставленный тайм-аут. Тем не менее, он попытается прочитать NUM-байты, пока не будет прочитано общее количество NUM-байтов, или пока не будет достигнут тайм-аут. Вот почему пример явно читает len("GOODBYE FROM MAIN THREAD") (24 байта) из pty, так что вызов conn.read будет немедленно возвращаться, а не ждать полного таймаута.

TL; DR можно использовать serial.Serial соединения с select.select, что и, я полагаю, вы ищете.