Python模拟串口通信的示例详解_Python

本次分享利用python模拟串口通信案例。当前编程环境，并没有办法接好下位机平台，需要模拟出一个串口，不断发送、接收信息的过程。

串行通信端口(cluster communication port )，简称串口，即com口。

pyserial 是 python 中用于操作串口的第三方模块，它支持 windows、linux、osx、bsd等多个平台。

1.win 下载虚拟串口

下载： sourceforge.net/projects/com0com/files/latest/download

下载安装完成后，会在设备管理器中新增两个com。

这两个串口是互相接通的，一个com 可以当作上位机，一个com可以当作下位机。创建的一对虚拟串口。

或者利用虚拟串口的软件创建，vspd（virtual serial port driver）

2、确定串口号

pyserial模块封装了python对串口的访问，为多平台的使用提供了统一的接口。

import serial
import serial.tools.list_ports
 
# 获取所有串口设备实例。

port_list = list(serial.tools.list_ports.comports())
def show_all_com():
    # 如果找到串口设备，则依次输出每个设备对应的串口号和描述信息。
    port_list_name = []
    if len(port_list) <= 0:
        print("the serial port can't find!")
    else:
        for itms in port_list:
            print(list(itms)[0], list(itms)[1])
            port_list_name.append(itms.device)

    return port_list_name

运行结果

com1 通信端口 (com1)
com3 com0com - serial port emulator (com3)
com4 com0com - serial port emulator (com4)
['com1', 'com3', 'com4']

3、配置串口

import serial
 
# 打开 com3，将波特率配置为115200，数据位为7，停止位为2，无校验位，读超时时间为0.5秒。
ser = serial.serial(port="com3",
                    baudrate=115200,
                    bytesize=serial.sevenbits,
                    parity=serial.parity_none,
                    stopbits=serial.stopbits_two,
                    timeout=0.5)

常用的参数如下：

port - 串口设备名或 none。

baudrate - 波特率，可以是50, 75, 110, 134, 150, 200, 300, 600, 1200, 1800, 2400, 4800, 9600, 19200, 38400, 57600, 115200, 230400, 460800, 500000, 576000, 921600, 1000000, 1152000, 1500000, 2000000, 2500000, 3000000, 3500000, 4000000。

bytesize - 数据位，可取值为：fivebits, sixbits, sevenbits, eightbits。

parity - 校验位，可取值为：parity_none, parity_even, parity_odd, parity_mark, parity_space。

stopbits - 停止位，可取值为：stopbits_one, stopbits_one_point_five, stopbits_tow。

xonxoff - 软件流控，可取值为 true, false。

rtscts - 硬件（rts/cts）流控，可取值为 true, false。

dsr/dtr - 硬件（dsr/dtr）流控，可取值为 true, false。

timeout - 读超时时间，可取值为 none, 0 或者其他具体数值（支持小数）。当设置为 none 时，表示阻塞式读取，一直读到期望的所有数据才返回；当设置为 0 时，表示非阻塞式读取，无论读取到多少数据都立即返回；当设置为其他数值时，表示设置具体的超时时间（以秒为单位），如果在该时间内没有读取到所有数据，则直接返回。

write_timeout: 写超时时间，可取值为 none, 0 或者其他具体数值（支持小数）。参数值起到的效果参考 timeout 参数。

4、串口通信示例

python 实现向一个串口内写数据（发送端），另一个串口内接受数据（接收端）

1.串口类工具

串口类功能函数组成：打开、关闭、发送数据、接受数据、解析保存数据功能

import serial
from time import sleep
import queue
import time
from datetime import datetime

q = queue.queue()

class serialport:
    def __init__(self, port, buand):
        self.port = serial.serial(port, buand)
        self.port.close()
        if not self.port.isopen():
            self.port.open()

    def port_open(self):
        if not self.port.isopen():
            self.port.open()

    def port_close(self):
        self.port.close()

    #发送数据
    def send_data(self):
        """
        #write() 方法只能发送 bytes 类型的数据，所以需要对字符串进行 encode 编码。
        # write() 方法执行完成后，会将发送的字节数作为返回值。
        """
        while true:   # 循环发送数据
            date = datetime.now().strftime('%h:%m:%s.%f')[:-3]
            print("send:", date)
            self.port.write(date.encode())
            time.sleep(1)

    # 读取数据
    def read_data(self):
        """
        #  read()方法默认一次读取一个字节，可以通过传入参数指定每次读取的字节数；
        read() 方法会将读取的内容作为返回值，类型为 bytes。
        """
        while true:
            count = self.port.inwaiting()
            if count != 0:
                recv = self.port.read(count)
                self.port.flushinput()
                q.put(recv.decode())
                print("receive:",recv.decode())
            sleep(0.5)

    #解析保存数据
    def save_json(self):
        while true:
            if q.empty():
                pass
            else:
                print("队列长度",q.qsize())
                data = q.get()
                print(data)
            sleep(0.1)

2.发送端

开启一个线程定时发送数据

from tools.loraserial import serialport
import threading
from time import sleep

if __name__=='__main__':
    baunrate = 115200
    serialport_w = "com3"
    mserial_w = serialport(serialport_w, baunrate)
    t1 = threading.thread(target=mserial_w.send_data)
    t1.daemon = true
    t1.start()

    while true:
        sleep(10)

3.接收端

****接收端开启两个线程接收数据：读取数据和解析存储数据。

from tools.loraserial import serialport
import threading
from time import sleep

if __name__=='__main__':
    threads=[]
    baunrate = 115200

    serialport_r = "com4"
    mserial_r = serialport(serialport_r, baunrate)

    t2 = threading.thread(target=mserial_r.read_data)
    threads.append(t2)

    t3 = threading.thread(target=mserial_r.save_json)
    threads.append(t3)

    for t in threads:
        t.start()

    while true:
        sleep(10)

5.运行结果

从一个线程向另一个线程发送数据最安全的方式可能就是使用 queue 库中的队列了。创建一个被多个线程共享的 queue 对象，这些线程通过使用put() 和 get() 操作来向队列中添加或者删除元素。

**当终端发送数据量较大时，其数据解析也是比较耗时，可以用消息队列缓存接收到的发送端数据，当接收线程进行io监听时，负责解析线程去消息队列取消息并解析。如下图，将解析过程时间延迟由0.1至2s。
**

为了方便模拟传递数据变化效果，这里用时间函数代替的，实际上设备传递具体消息可解析为具体的json数据格式****

消息类别	ori_id****发送端设备id	des_id****接收端设备id	cmd****消息类别	value****数组	消息方向
上线消息	1	123	online	[]	终端—网关
心跳消息	1	123	heart	[{“battery”:”60”}]	终端—网关
环境信息	1	123	env	[{“temp”:”35”},{“hum”:”66”}]	终端—网关
控制指令	1	123	alarm	[{“light”:”on”},{“pump”:”stop”}]	网关—终端
报警消息	123	1	control	[]	终端—网关