Python之串口收發(fā)的異步程序

更新時(shí)間：2023年09月06日 10:24:03 作者：土谷祠房客

這篇文章主要介紹了Python之串口收發(fā)的異步程序,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教

Python串口收發(fā)的異步程序

進(jìn)行Python串口程序開(kāi)發(fā)時(shí)，一般使用 pyserial 模塊。這是個(gè)同步方式運(yùn)行的庫(kù)。

串口運(yùn)行時(shí)，收發(fā)等待的時(shí)間較多，使用異步方式開(kāi)發(fā)可以提高開(kāi)發(fā)效率和運(yùn)行效率。

為了進(jìn)行異步編程，可以使用一個(gè)建立在 asyncio 基礎(chǔ)上的異步串口程序模塊，即pyserial_asyncio 模塊。

在使用這個(gè)模塊時(shí)，我們發(fā)現(xiàn)，網(wǎng)上資料中，能夠獨(dú)立直接運(yùn)行的簡(jiǎn)單實(shí)例較少。

為此，我們編寫了一個(gè)簡(jiǎn)單的可以獨(dú)立運(yùn)行的實(shí)例，以方便讀者使用這個(gè)模塊。

下面是這個(gè)程序的完整代碼

可以在 python 3.9 及以后的版本上運(yùn)行。

注意：

編程和運(yùn)行前要安裝 pyserial_asyncio 模塊。

import asyncio
import serial_asyncio
import time
# 接收異步函數(shù)，收到后打印時(shí)間、收到的內(nèi)容和接收的序號(hào)
async def read_from_serial(reader):
    count = 0
    while True:
        data = await reader.read(1000)
        p = time.strftime("%X", time.localtime())
        print(f'received at {p}, {data.decode()}, serial = {count}')
        count += 1
# 發(fā)送異步函數(shù)，每隔兩秒鐘發(fā)送一次
async def write_to_serial(writer):
    while True:
        writer.write(b'hello')
        await writer.drain()
        await asyncio.sleep(2)
async def main():
    # 連接串口，其中的'/dev/cu.usbserial-1140'是串口的名稱，
    # windows 的串口名一般是 'comx' 的格式，例如：'com3；
    # linux 的串口格式一般是 '/dev/ttyx' 的格式，例如：'dev/ttyUSB0'。
    reader, writer = await serial_asyncio.open_serial_connection(url='/dev/cu.usbserial-1140', baudrate=9600)
    # 生成串口讀寫異步任務(wù)各一個(gè)
    task_1 = asyncio.create_task(read_from_serial(reader))
    task_2 = asyncio.create_task(write_to_serial(writer))
    # 并行運(yùn)行兩個(gè)異步任務(wù)
    await task_1
    await task_2
if __name__ == '__main__':
    asyncio.run(main())

這個(gè)程序可以分為三個(gè)部分

1. 讀寫異步函數(shù)：read_from_serial 和 send_to_serial。其中后者每隔兩秒鐘向串口發(fā)送b“hello”字節(jié)串；前者則從串口接收數(shù)據(jù)，并將接收的時(shí)間的操作的序號(hào)打印出來(lái)。

2. 初始化異步函數(shù)main()，這個(gè)函數(shù)的功能是打開(kāi)串口、生成接收和發(fā)送任務(wù)，并行執(zhí)行這兩個(gè)任務(wù)，以進(jìn)行串口的收發(fā)。

3. 最后的__name__ == ‘__main__’: 的主程序部分啟動(dòng)程序的運(yùn)行。

這個(gè)程序的測(cè)試運(yùn)行過(guò)程

如下：

1. 準(zhǔn)備一個(gè) USB 轉(zhuǎn)串口的轉(zhuǎn)接板；

2. 將轉(zhuǎn)接板的 TxD 和 RxD 短接，使發(fā)送的數(shù)據(jù)直接送到接收端；

3. 在操作系統(tǒng)上查找插入的串口設(shè)備，在 windows 操作系統(tǒng)下，可以用設(shè)備管理器查找；MacOS 下可以在終端上用 ls /dev/cu.* 命令查找；Linux 下則可以用 ls /dev/ttyUSB* 查找。當(dāng)然，也可以用pyserial 模塊中的相關(guān)功能編程查找。

4. 根據(jù)查找的結(jié)果改寫異步 main 程序中的 uri 參數(shù)，上面程序是MacOS下的例子。

5. 啟動(dòng)程序，查看運(yùn)行結(jié)果（見(jiàn)下）。

/usr/local/bin/python3 /Users/xxxxxx/PycharmProjects/try_and_test/async_serial_3.py
received at 11:23:25, hello, serial = 0
received at 11:23:27, hello, serial = 1
received at 11:23:29, hello, serial = 2
received at 11:23:31, hello, serial = 3
received at 11:23:33, hello, serial = 4
received at 11:23:35, hello, serial = 5
received at 11:23:37, hello, serial = 6

在上述程序的基礎(chǔ)上，加上異常處理、內(nèi)部隊(duì)列接口等就可以實(shí)現(xiàn)自己所需功能的實(shí)用串口接口程序了。