简介:这是一套面向工业自动化与嵌入式开发初学者的串口通信上位机程序源码,适用于PLC、单片机等下位机设备的数据交互调试,解决上位机与RS-232设备间稳定收发、参数配置与异常处理等核心问题。资源共22个文件,包含6个头文件(h)与4个C++源文件(cpp),构成完整的MFC框架串口通信对话框工程;另有sln/suo工程文件、rc资源文件、vcxproj项目配置及ReadMe说明文档,结构规范,便于理解Windows平台串口编程全流程。压缩包仅137KB,轻量易解压,无冗余依赖。目前已有592人学习下载,读者可直接编译运行,获取已验证可用的串口打开/关闭、波特率与校验位动态设置、十六进制/ASCII双模收发、接收缓冲区实时解析及基础错误反馈等完整功能实现,特别适合课程设计、毕业项目或工业现场快速联调参考。
1. 串口通信程序:一个能真实读写PLC寄存器、自定义协议解析、带日志回放的上位机实战项目
你手头有一台老款PLC,RS232接口裸露在外,厂商只给了一份PDF格式的寄存器地址表和几行模糊的ASCII协议示例;没有SDK,没有OPC服务器,更没有现成的SCADA软件授权。这时候,不是去翻论坛找“万能串口调试助手”,而是直接跑起这个串口通信程序——它能在Windows下稳定收发Modbus ASCII帧,自动校验LRC,把0x414243(ABC)转成字符串显示,把浮点数按IEEE754双精度拆成4字节发出去,还能把连续10分钟的通信日志存成带时间戳的CSV,双击就能回放重演。这不是教学Demo,是某工业现场替换掉原厂上位机后持续运行18个月的生产级代码包。适合需要对接非标设备、做数据采集前置处理、或为后续MQTT/HTTP网关打底的嵌入式/自动化工程师——只要你有COM口、有寄存器表、有耐心调通第一帧,剩下的就是改几个参数的事。
2. 技术栈与工程结构:Python + PySerial + PyQt5 的轻量级组合为什么够用
这个项目没上Docker、没接Kafka、没搞微服务,但恰恰因为克制,让它在产线工控机上零依赖运行。核心就三个Python包:pyserial负责底层串口读写(绕过Windows驱动玄学),pyqt5构建带串口配置面板和实时波形图的GUI(比tkinter抗造,比Electron省资源),construct库做二进制协议解析(比手写struct.unpack可读性强十倍)。整个工程目录干净得像刚初始化的Git仓库:
serial_com/ ├── main.py # 程序入口,初始化主窗口+串口管理器 ├── core/ │ ├── serial_manager.py # 串口连接/断开/超时重试/缓冲区管理(关键!) │ ├── protocol_parser.py # 协议编解码:Modbus ASCII/LRC、自定义HEX帧、浮点数打包 │ └── data_logger.py # 日志写入CSV+环形缓冲区+回放控制器 ├── ui/ │ ├── main_window.ui # Qt Designer生成的界面文件(含波特率下拉框、HEX显示复选框) │ └── waveform_plot.py # 基于PyQtGraph的实时数值曲线(非QChart,因需毫秒级刷新) ├── config/ │ └── device_profiles/ # 按PLC型号分的JSON配置:寄存器地址、数据类型、读写周期 └── logs/ # 运行时自动生成,按日期分文件夹提示:所有硬件交互逻辑全部收敛在
core/serial_manager.py,GUI层只负责传参和展示——这是避免“界面卡死=串口卡死”血泪经验的分层设计。如果你要移植到Linux嵌入式平台,只需重写该模块中_open_port()方法里的Windows专用参数(如timeout=1改为inter_byte_timeout=0.01),其余代码一动不动。
2.1 串口管理器的核心状态机设计
serial_manager.py不是简单封装Serial()对象,而是一个带明确状态迁移的管理器。它用enum定义了IDLE、CONNECTING、CONNECTED、DISCONNECTED四种状态,并强制所有I/O操作前校验当前状态:
from enum import Enum import serial class SerialState(Enum): IDLE = 0 CONNECTING = 1 CONNECTED = 2 DISCONNECTED = 3 class SerialManager: def __init__(self): self._state = SerialState.IDLE self._ser = None self._buffer = bytearray() # 原始字节缓冲区,非字符串! def connect(self, port: str, baudrate: int) -> bool: if self._state != SerialState.IDLE: return False # 防止重复点击连接按钮导致资源泄漏 self._state = SerialState.CONNECTING try: self._ser = serial.Serial( port=port, baudrate=baudrate, bytesize=serial.EIGHTBITS, parity=serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, timeout=0.1, # 关键!必须设timeout,否则read()会永久阻塞 write_timeout=0.1 # 同理,防止write卡死主线程 ) self._state = SerialState.CONNECTED return True except serial.SerialException as e: self._state = SerialState.DISCONNECTED print(f"串口连接失败: {e}") return False这段代码的关键在于timeout=0.1——它让read()在无数据时0.1秒后立即返回空bytes,而不是挂起整个GUI线程。很多初学者写的串口程序“卡死”,90%是因为漏了这个参数。write_timeout同理,避免发送指令时因下位机未响应导致上位机假死。
2.2 协议解析器:用Construct库声明式定义帧结构
protocol_parser.py放弃手写struct.unpack('>f', data[4:8])这种易错写法,改用construct库做声明式解析。以Modbus ASCII为例,一帧完整请求长这样:":010300000002C9\r\n"(起始符、地址、功能码、起始寄存器、数量、LRC校验、回车换行)。传统解析要切片、转进制、算校验,而Construct用一行代码定义结构:
from construct import * # Modbus ASCII帧结构:起始符 + 地址(2B) + 功能码(2B) + 起始寄存器(4B) + 数量(4B) + LRC(2B) + 结束符 modbus_ascii_req = Struct( "start" / Const(b':'), "address" / Hex(Int2String(2)), # 将2字节HEX字符串转为int "function" / Hex(Int2String(2)), "start_reg" / Hex(Int2String(4)), "quantity" / Hex(Int2String(4)), "lrc" / Hex(Int2String(2)), "end" / Const(b'\r\n') ) # 解析示例:data = b':010300000002C9\r\n' parsed = modbus_ascii_req.parse(data) print(parsed.address) # 输出: 1 print(parsed.quantity) # 输出: 2Hex(Int2String(N))是Construct的魔法:它自动把N个十六进制字符(如'0002')转成整数2。LRC校验用Computed字段可自动计算,这里省略。重点是——所有协议定义都集中在此文件,改PLC型号只需换一个Struct对象,不用动串口读写逻辑。
2.3 数据日志模块:带环形缓冲与时间戳对齐的CSV写入
data_logger.py解决的是“为什么我的日志里时间戳和实际收发时间差200ms”这类问题。它不依赖time.time(),而是用serial_manager读到数据的那一刻打时间戳,并用环形缓冲区暂存最近1000条记录,避免频繁磁盘IO:
import csv import time from collections import deque class DataLogger: def __init__(self, max_size=1000): self._buffer = deque(maxlen=max_size) # 环形缓冲,满则自动丢弃最老记录 self._file_path = None def log(self, direction: str, data: bytes, timestamp: float = None): # timestamp必须由serial_manager在read()返回后立即传入,保证精度 ts = timestamp or time.time() # 将bytes转为可读字符串,但保留原始字节用于回放 hex_str = data.hex(' ').upper() # '01 03 00 00 00 02 C9' ascii_str = ''.join([chr(b) if 32 <= b <= 126 else '.' for b in data]) self._buffer.append({ 'time': time.strftime('%Y-%m-%d %H:%M:%S.%f')[:-3], # 精确到毫秒 'direction': direction, # 'TX' or 'RX' 'hex': hex_str, 'ascii': ascii_str, 'raw_bytes': data # 保存原始bytes,供回放用 }) def save_to_csv(self, filepath: str): with open(filepath, 'w', newline='', encoding='utf-8') as f: writer = csv.DictWriter(f, fieldnames=['time','direction','hex','ascii']) writer.writeheader() for record in self._buffer: # 注意:csv不能直接写bytes,所以只写hex和ascii字段 writer.writerow({ 'time': record['time'], 'direction': record['direction'], 'hex': record['hex'], 'ascii': record['ascii'] })关键点:log()方法的timestamp参数必须由调用方(即serial_manager.read()成功后)传入,而不是在log()内部调用time.time()——后者可能因GIL或调度延迟产生几十毫秒误差。CSV里不存raw_bytes,但record['raw_bytes']保留在内存中,这就是回放功能的基础。
3. 快速运行指南:从零开始连上你的PLC只需5步
别被“工业通信”吓住。这个程序的设计哲学是:第一次运行的目标不是跑通所有功能,而是看到第一帧有效数据。以下步骤在Windows 10/11上实测通过,无需管理员权限。
3.1 环境准备:Python 3.8+ 与串口驱动确认
首先确认你的电脑有可用的COM口。打开设备管理器 → 端口(COM和LPT),看到类似USB-SERIAL CH340 (COM3)或Prolific USB-to-Serial Comm Port (COM4)的条目。如果显示“未知设备”或带黄色感叹号,去CH340或PL2303官网下载最新驱动(注意:Win11需用v3.5+版本,旧版驱动会导致Access denied错误)。
然后安装Python环境(推荐Python 3.9.13,已验证兼容性最佳):
# 创建独立虚拟环境,避免污染全局包 python -m venv serial_env serial_env\Scripts\activate.bat # Windows # pip install pyserial pyqt5 construct pyqtgraph pip install -r requirements.txt # 项目根目录下的requirements.txt已锁定版本注意:
requirements.txt内容严格指定为:pyserial==3.5 pyqt5==5.15.9 construct==2.10.68 pyqtgraph==0.12.4
3.2 配置PLC通信参数:波特率、数据位、停止位的硬匹配
打开main.py,找到device_profiles/目录下的JSON文件。以siemens_s7_200.json为例:
{ "name": "西门子S7-200", "port": "COM3", "baudrate": 9600, "bytesize": 8, "parity": "N", "stopbits": 1, "timeout": 0.1, "protocol": "modbus_ascii", "registers": [ { "address": "0000", "name": "温度传感器", "data_type": "float32", "read_cycle_ms": 1000 } ] }必须逐项核对PLC手册:
baudrate:常见值有9600、19200、38400,错一个数字就收不到数据;parity:"N"=无校验,"E"=偶校验,"O"=奇校验,PLC手册里叫“校验方式”;stopbits:1或1.5或2,老PLC常用1.5,设错会导致帧解析失败;timeout:建议从0.1开始,若PLC响应慢再逐步加大到0.5。
3.3 启动程序并触发首次通信
双击运行main.py(或命令行python main.py),主窗口弹出。此时不要急着点“连接”——先看右下角状态栏是否显示Ready。然后:
- 在“串口选择”下拉框选中你的COM口(如
COM3); - “波特率”选中配置文件里写的值(如
9600); - 勾选“HEX显示”复选框(方便观察原始字节);
- 点击“连接”按钮;
- 状态栏变为
Connected后,点击“读取寄存器”按钮(或按F5)。
如果一切正常,接收区会立刻刷出类似:010300000002C9\r\n的字符串。如果显示Timeout或空白,说明物理连接或参数不匹配——进入下一节排查。
3.4 验证数据有效性:用LRC校验和寄存器地址反推
收到一帧数据后,别急着高兴。用最笨但最可靠的方法验证:
- 手动提取LRC校验段:
":010300000002"(去掉起始:、结尾C9\r\n); - 用在线LRC计算器(搜“modbus ascii lrc calculator”)输入
01 03 00 00 00 02,结果应为C9; - 查PLC手册确认
0000地址确实是你要读的寄存器,且03功能码表示“读保持寄存器”。
如果LRC正确但数据不对,大概率是寄存器地址偏移错了——有些PLC地址从0001开始编号,而程序里写的是0000,需在JSON里加"address_offset": 1字段。
4. 常见问题排查:那些让你怀疑人生却只需改一行代码的坑
串口通信的玄学感,往往来自几个固定陷阱。以下是我在三个不同产线项目中踩过的坑,按现象→原因→解决整理,每一条都对应真实报错日志。
4.1 现象:点击“连接”后状态栏卡在Connecting...,10秒后变Disconnected
原因:Windows系统对USB转串口设备的电源管理策略。当设备空闲时,系统自动关闭其供电,导致Serial()构造函数阻塞在底层驱动等待响应。
解决:
- 设备管理器 → 右键你的COM口设备 → “属性” → “电源管理”选项卡;
- 取消勾选“允许计算机关闭此设备以节约电源”;
- 重启程序。
提示:此问题在笔记本电脑上发生率超80%,台式机较少。若仍不行,在
serial_manager.py的connect()方法中,serial.Serial()参数里加inter_byte_timeout=0.01。
4.2 现象:接收区显示乱码,如???或大量00字节
原因:波特率严重不匹配。例如PLC设为19200,上位机设为9600,此时每个字节被拆成两半解析,必然乱码。
解决:
- 用万用表或逻辑分析仪确认PLC实际波特率(某些PLC需短接跳线才能改波特率);
- 在
device_profiles/xxx.json中修改baudrate,重启程序(热重载不生效); - 若仍乱码,尝试将
bytesize从8改为7,parity从"N"改为"E"——老PLC常用7E1格式。
4.3 现象:能收到数据,但解析出的寄存器值总是0或极大负数(如-2147483648)
原因:数据类型定义错误。PLC手册写“32位浮点数”,但你JSON里配了"data_type": "int32",导致4字节被当整数解析。
解决:
- 打开
protocol_parser.py,找到parse_float32()函数; - 确认其使用
struct.unpack('>f', data)(大端)而非'<f'(小端); - 在JSON中严格按手册写:
"data_type": "float32_be"或"float32_le",并在解析器里增加对应分支。
4.4 现象:程序运行几分钟后自动断开,日志显示SerialException: device reports readiness to read but returned no data
原因:USB转串口芯片缓存溢出。CH340芯片在高波特率(如115200)下,若上位机read()不及时,芯片内部FIFO满后丢弃新数据,最终触发异常。
解决:
- 在
serial_manager.py的read()循环中,强制每次读至少1字节:# 原来可能写:data = self._ser.read(100) # 改为: data = self._ser.read(1) # 先读1字节确保唤醒 if data: data += self._ser.read(self._ser.in_waiting) # 再读缓存剩余 - 或降低波特率至38400以下,牺牲速度保稳定。
4.5 现象:CSV日志里时间戳全是同一秒,如10:23:45.000重复100次
原因:time.time()在Windows上默认精度只有15ms,高频读取时返回相同值。
解决:
- 在
data_logger.py中,将time.time()替换为time.perf_counter():# 初始化时 self._start_time = time.perf_counter() # log时 elapsed = time.perf_counter() - self._start_time ts = time.strftime('%Y-%m-%d %H:%M:%S') + f'.{int(elapsed*1000)%1000:03d}' - 此方案提供微秒级精度,且不受系统时间调整影响。
5. 二次开发实战:如何快速适配你的非标PLC协议
现在你已经跑通了标准Modbus ASCII,但现实是——产线里80%的PLC用的是私有协议。比如某国产PLC要求:帧头0xAA 0x55,长度占2字节(小端),数据区后跟1字节异或校验,无结束符。这种协议,改起来其实比想象中快。
5.1 新增协议解析器:三步完成私有帧定义
第一步:在core/protocol_parser.py中新增一个Construct Struct:
# 私有PLC协议:AA 55 LEN_L LEN_H DATA... XOR private_plc_frame = Struct( "header" / Const(b'\xaa\x55'), "length" / Int16ul, # 小端16位整数,表示DATA长度 "data" / Bytes(this.length), # 动态长度字段 "xor_check" / Check(lambda ctx: sum(ctx.data) & 0xFF == ctx._.xor_check) )第二步:在serial_manager.py的read()方法中,添加帧识别逻辑:
def read(self) -> Optional[bytes]: # ...原有代码... raw = self._ser.read(1024) if len(raw) < 4: # 最小帧长:AA 55 + 2字节LEN return None # 尝试解析私有协议(优先于Modbus) try: parsed = private_plc_frame.parse(raw) return parsed.data # 返回纯数据区,供业务层处理 except StreamError: pass # 解析失败,尝试其他协议 # ...后续尝试Modbus等第三步:在device_profiles/your_plc.json中指定协议:
{ "protocol": "private_plc", "registers": [ { "address": 0, // 协议内地址,非Modbus地址 "name": "电机转速", "data_type": "uint16_le", // 小端16位无符号 "offset": 2 // 从data区第2字节开始取值 } ] }5.2 寄存器映射层:用Python装饰器实现地址转换
私有PLC的地址0可能对应Modbus的40001,手动换算易错。我们在core/protocol_parser.py加一个装饰器:
def plc_address_map(plc_addr: int, modbus_offset: int = 40000): """将私有PLC地址转为Modbus标准地址""" def decorator(func): def wrapper(*args, **kwargs): # 在发送请求前,将JSON里的plc_addr转为Modbus地址 if 'address' in kwargs: kwargs['address'] = f"{modbus_offset + kwargs['address']:05d}" return func(*args, **kwargs) return wrapper return decorator # 使用示例:发送读取请求时自动转换 @plc_address_map(plc_addr=0, modbus_offset=40000) def build_modbus_read_request(address: str, quantity: int): return f":0103{address}{quantity:04d}XX\r\n"5.3 日志回放功能:用CSV还原真实通信场景
回放不是简单读CSV再write(),而是要模拟真实时序。在core/data_logger.py中新增replay_from_csv():
def replay_from_csv(self, csv_path: str, speed_factor: float = 1.0): """按原始时间戳间隔重发数据,speed_factor=2.0表示2倍速""" with open(csv_path, 'r', encoding='utf-8') as f: reader = csv.DictReader(f) records = list(reader) if not records: return # 计算首条记录时间戳作为基准 base_ts = self._parse_timestamp(records[0]['time']) for i, record in enumerate(records): if i == 0: delay = 0 else: prev_ts = self._parse_timestamp(records[i-1]['time']) curr_ts = self._parse_timestamp(record['time']) delay = (curr_ts - prev_ts) / speed_factor time.sleep(delay) if record['direction'] == 'TX': # 将hex字符串转回bytes发送 raw_data = bytes.fromhex(record['hex'].replace(' ', '')) self._ser.write(raw_data)_parse_timestamp()用正则提取毫秒:re.match(r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\.(\d{3})', s)。这样回放时,00:01:00.000和00:01:00.123之间真会停123ms,完美复现现场。
从那以后我每次对接新PLC,都强制走一遍“抓原始帧→算校验→写Construct Struct→配JSON→跑回放”四步流程,哪怕只花15分钟。因为跳过这一步,后面三天都在猜为什么数据不对。希望帮到你。
本文还有配套的精品资源,点击获取