python的工业过程控制场景模拟第三十一篇:Modbus TCP与PLC通讯，远程读写调节阀输出值，搭建低成本简易上位机监控界面。-北京尧图网络

Modbus TCP 与 PLC 通讯及简易上位机监控系统 —— 基于 OOP 的工业数据实战一台 S7-1200 PLC 控制着车间的所有调节阀但操作工要看阀门开度还得走到电柜前面看面板。花半天写一个 Python 上位机坐在办公室就能监控所有阀门——这才是工程师该做的事。—— 哈尔滨工程大学《工业过程控制》课程核心思想一、实际应用场景描述在中小型自动化项目中PLC 负责逻辑控制和 PID 运算但缺少一个低成本的人机界面HMI。大型项目会用 WinCC、iFIX、InTouch 等专业 SCADA 软件但一套授权动辄几万块——对小车间、试验台、教学实训来说性价比极低。典型的 Modbus TCP 通讯架构┌─────────────────────┐ Modbus TCP (502端口) ┌─────────────────────┐│ PC (Python上位机) │═════════════════════════════════════→│ PLC (S7-1200) ││ │ 以太网 100Mbps │ ││ · 读保持寄存器 │────→ 读取阀门开度 PV │ DB1.DBW0 → 阀门1 ││ · 写单个寄存器 │────→ 写入手动设定值 SP │ DB1.DBW2 → 阀门2 ││ · 读线圈状态 │────→ 读取泵运行状态 │ DB1.X0 → 泵1 ││ · 刷新周期 1s │ │ DB1.X1 → 泵2 │└─────────────────────┘ └─────────────────────┘哈尔滨工程大学《工业过程控制》课程彭秀艳教授主讲国家级一流本科课程在第九章计算机控制系统通信中系统讲解了 Modbus 协议栈和主从通讯机制在第五章PID 控制器实现中介绍了上位机设定值写入和下位机执行的控制分工。课程明确指出Modbus TCP 是工业界最简单的普通话——几乎所有 PLC 都听得懂。学会用它你就拥有了和任何设备对话的能力。上位机不需要复杂能读写寄存器、能画图、能报警就是一个合格的监控站。二、引入痛点2.1 现场的真实困境场景现场发生了什么根因教学实训想让学生练 PID 整定但没有 HMI 专业软件太贵实验室经费有限小型车间操作工要跑到现场看阀门开多大没有低成本监控方案试验台每次改 PID 参数都要重新下载程序缺少在线调试接口设备调试怎么知道 PLC 里的数据对不对没有一个轻量工具做通讯测试远程运维半夜阀门报警工程师必须到现场没有远程查看手段2.2 核心矛盾PLC 已经在稳定运行数据就在寄存器里但用最低成本把这些数据变成人能看懂的画面这件事很多人觉得很难——其实只需要几十行 Python。- 买一套 WinCC预算不够。- 用组态王还要学它的脚本语言。- 自己写 C开发周期太长。- Python pymodbus tkinter 半天搞定。2.3 我们要解决什么用一段 Python 程序实现1. Modbus TCP 通讯 —— 连接 PLC读写寄存器和线圈2. 面向对象封装 —— 通讯逻辑、数据解析、UI 渲染分层3. 简易 GUI 界面 —— tkinter 画阀门状态、实时曲线、报警列表4. 远程读写 —— 双击阀门就能改设定值5. 模拟模式 —— 没有 PLC 也能运行内置模拟数据6. 通讯诊断 —— 连接状态、读写计数、错误日志三、核心逻辑讲解3.1 理论依据Modbus TCP 协议基础本工具基于哈工程《工业过程控制》第九章工业通讯网络① Modbus 四种数据类型类型地址范围读写对应 PLC 区域线圈 (Coil) 00001~09999 读/写 Q区输出离散输入 (DI) 10001~19999 只读 I区输入输入寄存器 (IR) 30001~39999 只读 PIW模拟输入保持寄存器 (HR) 40001~49999 读/写 DB块/M区② 功能码速查功能码含义用途0x01 读线圈读泵/阀开关状态0x02 读离散输入读限位开关/急停0x03 读保持寄存器读阀门开度/设定值0x04 读输入寄存器读传感器测量值0x05 写单个线圈远程启停泵0x06 写单个寄存器写入 PID 设定值0x10 写多个寄存器批量写入参数③ 寄存器地址映射PLC 侧 (S7-1200):DB1.DBD0 → 阀门1开度 (REAL, 占2个寄存器)DB1.DBD4 → 阀门2开度 (REAL, 占2个寄存器)DB1.DBX8.0 → 泵1运行 (BOOL)DB1.DBX8.1 → 泵2运行 (BOOL)Modbus 侧 (保持寄存器, 40001偏移):40001~40002 → 阀门1开度 (浮点数, IEEE754)40003~40004 → 阀门2开度线圈地址 0 → 泵1线圈地址 1 → 泵23.2 系统架构图┌────────────────────────────────────────────────────────────┐│ Python 上位机 ││ ││ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ││ │ ModbusClient │ │ ValveMonitor │ │ AlarmEngine │ ││ │ (通讯层) │──→│ (数据层) │──→│ (逻辑层) │ ││ │ │ │ │ │ │ ││ │ · connect() │ │ · update() │ │ · check() │ ││ │ · read_hr() │ │ · get_all() │ │ · trigger() │ ││ │ · write_hr() │ │ · set_sp() │ │ · history[] │ ││ └──────────────┘ └──────────────┘ └──────────────┘ ││ │ │ │ ││ └────────────────────┼────────────────────┘ ││ ▼ ││ ┌──────────────┐ ││ │ MonitorGUI │ ││ │ (界面层) │ ││ │ │ ││ │ · 阀门状态卡 │ ││ │ · 实时曲线 │ ││ │ · 报警列表 │ ││ │ · 设定值输入 │ ││ └──────────────┘ │└────────────────────────────────────────────────────────────┘3.3 为什么用面向对象而不是脚本式写法# 脚本式 (新手常见写法)client ModbusTcpClient(192.168.1.10)result client.read_holding_registers(0, 2)val decode_float(result.registers)print(val)# 过200行后... 到处都是 client.read_xxx()改IP要改10处# OOP式 (本项目)plc PLCConnection(host192.168.1.10)monitor ValveMonitor(plc)gui MonitorGUI(monitor)gui.run()# IP只在一个地方配置通讯逻辑和界面完全解耦分层的好处换 PLC 品牌只改通讯层换界面框架只改 UI 层核心逻辑不动。四、代码模块化讲解面向对象设计4.1 类结构总览类名职责设计模式PLCConnection Modbus TCP 通讯封装适配器模式ValveMonitor 阀门数据管理与解析观察者模式AlarmEngine 报警逻辑判断策略模式MonitorGUI tkinter 界面渲染 MVC 中的 ViewMockPLC 模拟 PLC无硬件时测试模拟对象4.2 通讯层PLCConnection# plc_connection.pyfrom pymodbus.client import ModbusTcpClientfrom pymodbus.exceptions import ModbusExceptionimport structclass PLCConnection:PLC Modbus TCP 通讯封装对应课程 §9.x: Modbus 协议的主站实现def __init__(self, host: str 192.168.1.10, port: int 502,unit_id: int 1, timeout: int 5):self.host hostself.port portself.unit_id unit_idself.timeout timeoutself._client Noneself.connected Falseself.error_count 0self.read_count 0def connect(self) - bool:建立连接try:self._client ModbusTcpClient(hostself.host,portself.port,timeoutself.timeout)self.connected self._client.connect()return self.connectedexcept Exception as e:print(f[ERROR] 连接失败: {e})self.connected Falsereturn Falsedef disconnect(self):断开连接if self._client:self._client.close()self.connected Falsedef read_float(self, address: int) - float | None:读取浮点数 (占2个寄存器, IEEE754)Modbus 保持寄存器地址从0开始:40001 → address040003 → address2if not self.connected:return Nonetry:result self._client.read_holding_registers(addressaddress,count2,slaveself.unit_id)if result.isError():self.error_count 1return None# IEEE754 解码: 两个 16 位寄存器 → 32 位浮点数reg result.registers# 注意字节序: 西门子 PLC 是 big-endianval struct.unpack(f, struct.pack(HH, reg[0], reg[1]))[0]self.read_count 1return round(val, 2)except (ModbusException, struct.error) as e:self.error_count 1return Nonedef write_float(self, address: int, value: float) - bool:写入浮点数if not self.connected:return Falsetry:# 浮点数 → 两个 16 位寄存器packed struct.pack(f, value)regs struct.unpack(HH, packed)result self._client.write_registers(addressaddress,valueslist(regs),slaveself.unit_id)return not result.isError()except Exception:self.error_count 1return Falsedef read_coils(self, address: int, count: int 1) - list | None:读线圈状态if not self.connected:return Nonetry:result self._client.read_coils(address, count, slaveself.unit_id)if result.isError():return Nonereturn result.bits[:count]except Exception:return Nonedef write_coil(self, address: int, value: bool) - bool:写单个线圈if not self.connected:return Falsetry:result self._client.write_coil(address, value, slaveself.unit_id)return not result.isError()except Exception:return False亮点-read_float() 内部处理了 IEEE754 编解码——调用方只需传地址和值- 错误计数和读写计数自动统计——方便诊断通讯质量- 连接状态统一管理——避免重复连接4.3 数据层ValveMonitor# valve_monitor.pyfrom dataclasses import dataclass, fieldfrom datetime import datetimefrom typing import Callabledataclassclass ValveStatus:单个阀门的状态快照name: strpv: float 0.0 # 过程值 (实际开度%)sp: float 0.0 # 设定值%auto: bool True # 自动/手动模式running: bool False # 运行中alarm_high: bool Falsealarm_low: bool Falsetimestamp: datetime field(default_factorydatetime.now)# 类型别名ValveUpdateCallback Callable[[dict], None]class ValveMonitor:阀门数据管理器职责:1. 定时从 PLC 读取所有阀门数据2. 解析并缓存最新状态3. 通知 GUI 更新def __init__(self, plc, poll_interval_ms: int 1000):self.plc plcself.poll_interval poll_interval_ms / 1000.0# 阀门配置: {名称: {pv_addr: xx, sp_addr: xx, ...}}self.valves {FV-101: {pv_addr: 0, sp_addr: 2, coil_addr: 0},FV-102: {pv_addr: 4, sp_addr: 6, coil_addr: 1},FV-103: {pv_addr: 8, sp_addr: 10, coil_addr: 2},}# 最新状态缓存self.status: dict[str, ValveStatus] {}for name in self.valves:self.status[name] ValveStatus(namename)# 历史数据 (用于曲线绘制)self.history: dict[str, list[tuple[datetime, float]]] {name: [] for name in self.valves}# 回调函数列表 (观察者模式)self._callbacks: list[ValveUpdateCallback] []def add_callback(self, cb: ValveUpdateCallback):注册数据更新回调self._callbacks.append(cb)def _notify(self, data: dict):通知所有观察者for cb in self._callbacks:try:cb(data)except Exception:passdef poll_once(self) - dict:执行一次轮询返回: {阀门名: ValveStatus} 的字典for name, cfg in self.valves.items():# 读 PV (过程值)pv self.plc.read_float(cfg[pv_addr])if pv is not None:self.status[name].pv pvself.history[name].append((datetime.now(), pv))# 保留最近300个点 (5分钟1s)if len(self.history[name]) 300:self.history[name].pop(0)# 读 SP (设定值)sp self.plc.read_float(cfg[sp_addr])if sp is not None:self.status[name].sp sp# 读运行状态 (线圈)coils self.plc.read_coils(cfg[coil_addr], 1)if coils:self.status[name].running coils[0]self.status[name].timestamp datetime.now()# 转为可序列化字典snapshot {name: {pv: s.pv,sp: s.sp,auto: s.auto,running: s.running,ts: s.timestamp.strftime(%H:%M:%S),}for name, s in self.status.items()}self._notify(snapshot)return snapshotdef write_setpoint(self, valve_name: str, value: float) - bool:写入设定值if valve_name not in self.valves:return Falseaddr self.valves[valve_name][sp_addr]return self.plc.write_float(addr, max(0.0, min(100.0, value)))def toggle_valve(self, valve_name: str, on: bool) - bool:开关阀门if valve_name not in self.valves:return Falseaddr self.valves[valve_name][coil_addr]return self.plc.write_coil(addr, on)亮点- 观察者模式add_callback() 让 GUI 自动接收数据更新无需轮询- 历史数据自动裁剪保留最近 300 点防止内存无限增长- 设定值限幅max(0.0, min(100.0, value)) 防止写入非法值4.4 报警引擎AlarmEngine# alarm_engine.pyfrom dataclasses import dataclass, fieldfrom datetime import datetimefrom typing import Callabledataclassclass AlarmEvent:报警事件timestamp: datetimevalve: strlevel: str # INFO / WARN / CRITICALmessage: strack: bool FalseAlarmCallback Callable[[AlarmEvent], None]class AlarmEngine:报警逻辑引擎规则:PV 95% → 开度过高警告PV 5% 且 Running → 开度过低警告通讯中断 → 严重报警def __init__(self):self.alarms: list[AlarmEvent] []self.callbacks: list[AlarmCallback] []self._prev_states: dict[str, dict] {} # 防抖: 状态变化时才报警def add_callback(self, cb: AlarmCallback):self.callbacks.append(cb)def _fire(self, valve: str, level: str, msg: str):evt AlarmEvent(datetime.now(), valve, level, msg)self.alarms.append(evt)for cb in self.callbacks:cb(evt)def check(self, data: dict):检查所有阀门的报警条件for name, info in data.items():pv info.get(pv, 50)running info.get(running, False)prev self._prev_states.get(name, {})# 变化检测: 只在状态变化时触发if pv 95 and not prev.get(high_alarm):self._fire(name, WARN, f阀门开度过高 ({pv}%))if pv 5 and running and not prev.get(low_alarm_running):self._fire(name, WARN, f运行中开度过低 ({pv}%))self._prev_states[name] {high_alarm: pv 95,low_alarm_running: pv 5 and running,}def communication_lost(self):通讯中断报警self._fire(SYSTEM, CRITICAL, PLC 通讯中断!)def acknowledge_all(self):确认所有报警for a in self.alarms:a.ack True4.5 界面层MonitorGUI核心片段# monitor_gui.pyimport tkinter as tkfrom tkinter import ttk, messageboximport matplotlib.pyplot as pltfrom matplotlib.backends.backend_tkagg import FigureCanvasTkAggfrom matplotlib.animation import FuncAnimationimport threadingplt.rcParams[font.sans-serif] [SimHei] # Windows 中文plt.rcParams[axes.unicode_minus] Falseclass MonitorGUI:简易上位机监控界面布局:┌─────────────────────────────────────┐│ 连接栏: IP [_______] [连接] [断开] │├──────────────────┬──────────────────┤│ 阀门状态面板 │ 实时曲线 ││ ┌────┐ ┌────┐ │ (matplotlib) ││ │FV101│ │FV102│ │ ││ │PV:50│ │PV:75│ │ ━━━━━━━━━━ ││ │SP:[__]│ │SP:[__]│ │ ││ └────┘ └────┘ │ ││ ┌────┐ │ ││ │FV103│ │ ││ └────┘ │ │├──────────────────┴──────────────────┤│ 报警列表: [确认全部] ││ [12:01] FV-101 开度过高 (98%) │└─────────────────────────────────────┘def __init__(self, monitor, alarms):self.monitor monitorself.alarms alarmsself.root tk.Tk()self.root.title(简易上位机监控系统 v1.0)self.root.geometry(900x650)self._build_ui()self._bind_callbacks()# 启动后台轮询线程self._polling Falseself._poll_thread Nonedef _build_ui(self):构建界面# 顶部连接栏 top ttk.Frame(self.root, padding5)top.pack(filltk.X)ttk.Label(top, textPLC IP:).pack(sidetk.LEFT)self.ip_var tk.StringVar(value192.168.1.10)ttk.Entry(top, textvariableself.ip_var, width15).pack(sidetk.LEFT, padx5)self.btn_connect ttk.Button(top, text连接, commandself._on_connect)self.btn_connect.pack(sidetk.LEFT, padx5)self.btn_disconnect ttk.Button(top, text断开, commandself._on_disconnect, statetk.DISABLED)self.btn_disconnect.pack(sidetk.LEFT)self.conn_status ttk.Label(top, text● 未连接, foregroundred)self.conn_status.pack(sidetk.RIGHT)# 中间主区域 main ttk.PanedWindow(self.root, orienttk.HORIZONTAL)main.pack(filltk.BOTH, expandTrue, padx5, pady5)# 左: 阀门面板left ttk.Frame(main, padding5)main.add(left, weight1)self._build_valve_panel(left)# 右: 曲线right ttk.Frame(main, padding5)main.add(right, weight2)self._build_chart(right)# 底部报警栏 bot ttk.LabelFrame(self.root, text报警列表, padding5)bot.pack(filltk.X, padx5, pady5)self.alarm_listbox tk.Listbox(bot, height5, fgred)self.alarm_listbox.pack(filltk.X)ttk.Button(bot, text确认全部, commandself._ack_all).pack(anchortk.E, pady2)def _build_valve_panel(self, parent):构建阀门卡片self.valve_cards {}for i, name in enumerate(self.monitor.valves):card ttk.LabelFrame(parent, textname, padding8)card.grid(rowi//2, columni%2, padx5, pady5, stickyew)ttk.Label(card, textPV:).grid(row0, column0, stickyw)pv_label ttk.Label(card, text--%, font(Consolas, 14, bold))pv_label.grid(row0, column1, padx10)ttk.Label(card, textSP:).grid(row1, column0, stickyw)sp_var tk.DoubleVar(value50.0)sp_entry ttk.Entry(card, textvariablesp_var, width8)sp_entry.grid(row1, column1, padx10)sp_entry.bind(Return, lambda e, nname, vsp_var: self._write_sp(n, v.get()))running_label ttk.Label(card, text● 停止, foregroundgray)running_label.grid(row2, column0, columnspan2, pady3)self.valve_cards[name] {pv: pv_label, sp: sp_var, running: running_label}def _build_chart(self, parent):构建 matplotlib 曲线self.fig, self.ax plt.subplots(figsize(5, 3), dpi80)self.ax.set_title(阀门开度实时趋势)self.ax.set_xlabel(时间)self.ax.set_ylabel(开度 (%))self.ax.set_ylim(0, 105)self.ax.grid(True, alpha0.3)self.lines {}colors [blue, green, orange]for i, name in enumerate(self.monitor.valves):line, self.ax.plot([], [], colorcolors[i], labelname)self.lines[name] lineself.ax.legend(fontsize8)self.canvas FigureCanvasTkAgg(self.fig, parent)self.canvas.get_tk_widget().pack(filltk.BOTH, expandTrue)def _bind_callbacks(self):绑定数据回调self.monitor.add_callback(self._on_data_update)self.alarms.add_callback(self._on_alarm)def _on_data_update(self, data: dict):数据更新回调 (在后台线程触发, 需用 after 切回主线程)self.root.after(0, self._refresh_ui, data)def _refresh_ui(self, data: dict):刷新界面 (主线程)for name, info in data.items():card self.valve_cards[name]card[pv].config(textf{info[pv]:.1f}%)card[running].config(text● 运行中 if info[running] else ● 停止,foregroundgreen if info[running] else gray)# 刷新曲线for name, line in self.lines.items():hist self.monitor.history[name][-50:] # 最近50点if len(hist) 1:times [h[0].strftime(%H:%M:%S) for h in hist]vals [h[1] for h in hist]line.set_data(range(len(vals)), vals)if self.monitor.history[FV-101]:self.ax.set_xlim(0, 49)self.canvas.draw_idle()def _on_alarm(self, evt):报警回调text f[{evt.timestamp.strftime(%H:%M:%S)}] {evt.valve利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛

python的工业过程控制场景模拟第三十一篇:Modbus TCP与PLC通讯，远程读写调节阀输出值，搭建低成本简易上位机监控界面。

相关推荐

河南AI人事招聘排名

一个靠谱的C端网关服务应该包含什么内容

NVIC：Cortex-M内核的“中断CPU”

更多资讯

汇川InoProShop Modbus从站配置避坑指南：从站地址、寄存器映射与通讯排查

I2C总线深度解析：从开漏输出到多主仲裁的工程实践

EMC整改流程及常见问题：从辐射超标到顺利通过

8款高效AI工具实测：写作、设计与编程提效神器

两轮车充电器为何转向LLC谐振+全数字CC/CV

解析编程中看似矛盾的比较表达式

今日更新

2k显示屏性能优化踩坑：版本升级后API全变了，这份源码解析救了我

3步搞定美眉图实战项目，告别官方文档抓不住重点

3步搞定黄金大劫案项目搭建从入门到精通

本周精选

深入解析Transformer多头注意力机制与工程优化

OpenClaw 的 Skills 跑学习任务，模型通道改到 TaoToken 通道行不行？

ChatGPT报错Oops, an error occurred! 全链路排查指南

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

读完文章，想聊聊您的网站？