Python串口通信案例

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

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

pySerial 是 Python 中用于操作串口的第三方模块，它支持 Windows、Linux、OSX、BSD等多个平台。

1.win 下载虚拟串口

https://sourceforge.net/projects/com0com/files/latest/download 下载安装完成后，会在设备管理器中新增两个com。

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

或者利用虚拟串口的软件创建，VSPD（Virtual Serial Port Driver）

2、确定串口号

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

代码语言：javascript复制

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

代码语言：javascript复制

运行结果

COM1 通信端口 (COM1)
COM3 com0com - serial port emulator (COM3)
COM4 com0com - serial port emulator (COM4)
['COM1', 'COM3', 'COM4']

3、配置串口

代码语言：javascript复制

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) 

代码语言：javascript复制

常用的参数如下：

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 实现向一个串口内写数据（发送端），另一个串口内接受数据（接收端）

• 串口类工具

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

代码语言：javascript复制

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()

    # 读取数据
    def read_data(self):
        """
        #  read()方法默认一次读取一个字节，可以通过传入参数指定每次读取的字节数；
        read() 方法会将读取的内容作为返回值，类型为 bytes。
        """
        while True:
            count = self.port.inWaiting()
            if count != :
                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)

• 发送端

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

代码语言：javascript复制

from tools.LoRaSerial import SerialPort
import threading
from time import sleep

if __name__=='__main__':
    baunRate = 
    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()

• 接收端

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

代码语言：javascript复制

from tools.LoRaSerial import SerialPort
import threading
from time import sleep

if __name__=='__main__':
    threads=[]
    baunRate = 

    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()

运行结果

从一个线程向另一个线程发送数据最安全的方式可能就是使用 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	[]	终端—网关

备注：

1、 上线消息-------终端启动时给网关发送上线的消息，网关返回该终端的各种数据及运行状态

2、 心跳消息--------终端定期发送心跳消息给网关

3、 环境信息--------终端定期上报各种环境参数

4、 控制指令-------网关发送指令给终端，实现对终端的控制，如开灯、关灯等

5、 报警消息-------设备监测到报警信息，发送报警信息给网关

kafka 消息队列CMQ版 python

0 人点赞