Python CustomTkinter Modbus TCP客户端开发实战

bash1pip install customtkinter pymodbus pillow

bash1pip show pymodbus

1modbus_client/2├── main.py              # 程序入口3├── modbus_core.py       # 通信层：只管连接和读写4├── data_model.py        # 数据层：寄存器映射和数据处理5└── ui/6    ├── main_window.py   # 主窗口7    ├── connect_panel.py # 连接配置面板8    └── data_panel.py    # 数据显示面板

python1from pymodbus.client import ModbusTcpClient2from pymodbus.exceptions import ModbusException3import threading4import logging56logger = logging.getLogger(__name__)789class ModbusCore:10"""11    Modbus TCP通信核心12    职责单一：只负责连接管理和寄存器读写13    """14def __init__(self):15        self._client = None16        self._lock = threading.Lock()  # 多线程环境必须加锁17        self.is_connected = False1819def connect(self, host: str, port: int = 502, timeout: int = 3) -> tuple[bool, str]:20"""21        建立连接，返回(成功标志, 错误信息)22        timeout建议3秒，太长了界面会卡住23        """        try:24            self._client = ModbusTcpClient(25                host=host,26                port=port,27                timeout=timeout28            )29            result = self._client.connect()30            self.is_connected = result31if result:32return True, ""33return False, f"无法连接到 {host}:{port}"34except Exception as e:35            self.is_connected = False36return False, str(e)3738def disconnect(self):39if self._client:40            self._client.close()41        self.is_connected = False4243def read_holding_registers(44            self,45            address: int,46            count: int,47            slave_id: int = 148    ) -> tuple[list | None, str]:49"""50        读保持寄存器（FC03）51        注意：address是0-based，对应PLC文档中的4xxxx地址要减152        """        if not self.is_connected:53return None, "未连接"5455with self._lock:56try:57                response = self._client.read_holding_registers(58                    address=address,59                    count=count,60                    device_id=slave_id61                )62if response.isError():63return None, f"读取错误: {response}"64return response.registers, ""65except ModbusException as e:66                self.is_connected = False67return None, f"通信异常: {e}"6869def write_register(70            self,71            address: int,72            value: int,73            slave_id: int = 174    ) -> tuple[bool, str]:75"""写单个寄存器（FC06）"""76if not self.is_connected:77return False, "未连接"7879with self._lock:80try:81                response = self._client.write_register(82                    address=address,83                    value=value,84                    device_id=slave_id85                )86if response.isError():87return False, f"写入错误: {response}"88return True, ""89except ModbusException as e:90                self.is_connected = False91return False, f"通信异常: {e}"

python1from dataclasses import dataclass, field2from typing import Callable345@dataclass6class RegisterMap:7"""寄存器映射表——把地址和业务含义绑定在一起"""8    name: str  # 显示名称9    address: int  # 寄存器地址（0-based）10    scale: float = 1.0  # 缩放系数（如温度值/10）11    unit: str = ""  # 单位12    writable: bool = False  # 是否可写131415# 项目寄存器配置——集中管理，修改方便16REGISTER_CONFIG = [17RegisterMap("设备温度", address=0, scale=0.1, unit="℃"),18RegisterMap("运行速度", address=1, scale=1.0, unit="rpm"),19RegisterMap("压力值", address=2, scale=0.01, unit="MPa"),20RegisterMap("目标速度", address=10, scale=1.0, unit="rpm", writable=True),21RegisterMap("启停控制", address=20, scale=1.0, unit="", writable=True),22]232425class DataModel:26"""数据处理层：原始寄存器值 → 工程量转换"""2728def process_raw_value(self, reg: RegisterMap, raw: int) -> float:29"""16位无符号整数转工程量"""30# 处理有符号数（某些传感器会返回负值）31if raw > 32767:32            raw -= 6553633return round(raw * reg.scale, 3)3435def engineering_to_raw(self, reg: RegisterMap, value: float) -> int:36"""工程量转寄存器原始值（写入时使用）"""37        raw = int(value / reg.scale)38# 限制在16位范围内39return max(0, min(65535, raw))

python1import customtkinter as ctk2import threading3import time4from modbustcpDemo.modbus_core import ModbusCore5from modbustcpDemo.data_model import DataModel, REGISTER_CONFIG67# 全局主题设置8ctk.set_appearance_mode("dark")9ctk.set_default_color_theme("blue")101112class MainWindow(ctk.CTk):1314def __init__(self, settings):15super().__init__()16        self._settings = settings17        self.title("Modbus TCP 客户端  v1.0")18        self.geometry("900x650")19        self.resizable(True, True)2021# 核心对象22        self._modbus = ModbusCore()23        self._data_model = DataModel()2425# 轮询控制26        self._polling = False27        self._poll_thread = None28        self._slave_id = int(self._get_setting("slave_id", 1))2930        self._build_ui()3132def _get_setting(self, name: str, default):33if isinstance(self._settings, dict):34return self._settings.get(name, default)35return getattr(self._settings, name, default)3637def _build_ui(self):38"""构建界面布局"""39# 顶部连接区域40        self._build_connection_bar()4142# 主内容区：左侧数据显示，右侧控制面板43        content_frame = ctk.CTkFrame(self, fg_color="transparent")44        content_frame.pack(fill="both", expand=True, padx=15, pady=(0, 15))45        content_frame.columnconfigure(0, weight=3)46        content_frame.columnconfigure(1, weight=2)47        content_frame.rowconfigure(0, weight=1)4849        self._build_data_panel(content_frame)50        self._build_control_panel(content_frame)5152def _build_connection_bar(self):53"""连接配置栏"""54        bar = ctk.CTkFrame(self, height=70)55        bar.pack(fill="x", padx=15, pady=(15, 10))56        bar.pack_propagate(False)5758# IP地址输入59        ctk.CTkLabel(bar, text="设备IP：", width=60).pack(side="left", padx=(15, 5), pady=15)60        self._ip_entry = ctk.CTkEntry(bar, width=140, placeholder_text="192.168.1.100")61        self._ip_entry.pack(side="left", pady=15)62        self._ip_entry.insert(0, str(self._get_setting("default_ip", "127.0.0.1")))6364# 端口65        ctk.CTkLabel(bar, text="端口：", width=45).pack(side="left", padx=(15, 5))66        self._port_entry = ctk.CTkEntry(bar, width=70)67        self._port_entry.pack(side="left", pady=15)68        self._port_entry.insert(0, str(self._get_setting("default_port", 502)))6970# 从站ID71        ctk.CTkLabel(bar, text="从站ID：", width=60).pack(side="left", padx=(15, 5))72        self._slave_entry = ctk.CTkEntry(bar, width=50)73        self._slave_entry.pack(side="left", pady=15)74        self._slave_entry.insert(0, str(self._get_setting("slave_id", 1)))7576# 连接按钮77        self._conn_btn = ctk.CTkButton(78            bar, text="连接", width=90,79            command=self._toggle_connection,80            fg_color="#2d7a2d", hover_color="#1f5c1f"81        )82        self._conn_btn.pack(side="left", padx=20, pady=15)8384# 状态指示85        self._status_label = ctk.CTkLabel(86            bar, text="● 未连接",87            text_color="#888888", font=("微软雅黑", 13)88        )89        self._status_label.pack(side="left", padx=10)9091# 轮询间隔92        ctk.CTkLabel(bar, text="轮询(ms)：").pack(side="right", padx=(0, 5))93        self._interval_var = ctk.StringVar(value="500")94        self._interval_entry = ctk.CTkEntry(95            bar, width=60, textvariable=self._interval_var96        )97        self._interval_entry.pack(side="right", padx=(0, 15))9899def _build_data_panel(self, parent):100"""数据显示面板——根据寄存器配置自动生成"""101        panel = ctk.CTkScrollableFrame(parent, label_text="实时数据监控")102        panel.grid(row=0, column=0, sticky="nsew", padx=(0, 8))103104        self._value_labels = {}  # 存储各寄存器的显示Label105106for i, reg in enumerate(REGISTER_CONFIG):107            row_frame = ctk.CTkFrame(panel, height=45)108            row_frame.pack(fill="x", pady=3, padx=5)109            row_frame.pack_propagate(False)110111# 寄存器名称112            ctk.CTkLabel(113                row_frame,114                text=f"{reg.name}  [addr: {reg.address}]",115                anchor="w", width=200,116                font=("微软雅黑", 12)117            ).pack(side="left", padx=15)118119# 数值显示120            val_label = ctk.CTkLabel(121                row_frame,122                text="---",123                font=("Consolas", 16, "bold"),124                text_color="#4fc3f7",125                width=100126            )127            val_label.pack(side="left", padx=10)128            self._value_labels[reg.address] = val_label129130# 单位131            ctk.CTkLabel(132                row_frame,133                text=reg.unit,134                text_color="#aaaaaa",135                width=40136            ).pack(side="left")137138def _build_control_panel(self, parent):139"""控制写入面板"""140        panel = ctk.CTkFrame(parent)141        panel.grid(row=0, column=1, sticky="nsew")142143        ctk.CTkLabel(144            panel, text="参数控制",145            font=("微软雅黑", 14, "bold")146        ).pack(pady=(15, 10))147148# 只显示可写寄存器149        writable_regs = [r for r in REGISTER_CONFIG if r.writable]150151        self._write_entries = {}152153for reg in writable_regs:154            frame = ctk.CTkFrame(panel, fg_color="transparent")155            frame.pack(fill="x", padx=15, pady=5)156157            ctk.CTkLabel(frame, text=reg.name, anchor="w").pack(fill="x")158159            entry_row = ctk.CTkFrame(frame, fg_color="transparent")160            entry_row.pack(fill="x")161162            entry = ctk.CTkEntry(entry_row, placeholder_text=f"输入{reg.unit}")163            entry.pack(side="left", fill="x", expand=True, padx=(0, 8))164            self._write_entries[reg.address] = entry165166# 用lambda捕获当前reg，避免闭包陷阱167            ctk.CTkButton(168                entry_row, text="写入", width=60,169                command=lambda r=reg: self._write_value(r)170            ).pack(side="right")171172# 日志区域173        ctk.CTkLabel(panel, text="操作日志", anchor="w").pack(174            fill="x", padx=15, pady=(20, 5)175        )176        self._log_box = ctk.CTkTextbox(panel, height=180, font=("Consolas", 11))177        self._log_box.pack(fill="both", expand=True, padx=15, pady=(0, 15))178179180def _toggle_connection(self):181if not self._modbus.is_connected:182            self._do_connect()183else:184            self._do_disconnect()185186def _do_connect(self):187        host = self._ip_entry.get().strip()188try:189            port = int(self._port_entry.get())190            self._slave_id = int(self._slave_entry.get())191except ValueError:192            self._log("端口或从站ID格式错误")193return194195        self._log(f"正在连接 {host}:{port}...")196        self._conn_btn.configure(state="disabled", text="连接中...")197198# 连接操作放到线程里，避免界面卡死199def do_connect_thread():200            ok, err = self._modbus.connect(host, port)201            self.after(0, self._on_connect_result, ok, err)202203        threading.Thread(target=do_connect_thread, daemon=True).start()204205def _on_connect_result(self, ok: bool, err: str):206if ok:207            self._status_label.configure(text="● 已连接", text_color="#4caf50")208            self._conn_btn.configure(209                state="normal", text="断开",210                fg_color="#8b2020", hover_color="#6b1515"211            )212            self._log("连接成功，开始轮询数据")213            self._start_polling()214else:215            self._status_label.configure(text="● 连接失败", text_color="#f44336")216            self._conn_btn.configure(state="normal", text="连接",217                                     fg_color="#2d7a2d", hover_color="#1f5c1f")218            self._log(f"连接失败：{err}")219220def _do_disconnect(self):221        self._stop_polling()222        self._modbus.disconnect()223        self._status_label.configure(text="● 未连接", text_color="#888888")224        self._conn_btn.configure(state="normal", text="连接",225                                 fg_color="#2d7a2d", hover_color="#1f5c1f")226# 清空显示227for label in self._value_labels.values():228            label.configure(text="---")229        self._log("已断开连接")230231def _start_polling(self):232        self._polling = True233        self._poll_thread = threading.Thread(234            target=self._poll_loop, daemon=True235        )236        self._poll_thread.start()237238def _stop_polling(self):239        self._polling = False240241def _poll_loop(self):242"""后台轮询线程——持续读取寄存器数据"""243while self._polling and self._modbus.is_connected:244try:245                interval_ms = int(self._interval_var.get())246except ValueError:247                interval_ms = 500248249for reg in REGISTER_CONFIG:250if not self._polling:251break252                values, err = self._modbus.read_holding_registers(253                    address=reg.address, count=1, slave_id=self._slave_id254                )255if values:256                    eng_val = self._data_model.process_raw_value(reg, values[0])257                    display = f"{eng_val} {reg.unit}"258                    label = self._value_labels[reg.address]259                    self.after(0, lambda lbl=label, d=display: lbl.configure(text=d))260elif err:261                    self.after(0, self._log, f"读取[{reg.name}]失败: {err}")262263            time.sleep(interval_ms / 1000.0)264265if not self._modbus.is_connected and self._polling:266            self.after(0, self._on_connection_lost)267268def _on_connection_lost(self):269"""连接意外断开的处理"""270        self._polling = False271        self._status_label.configure(text="● 连接断开", text_color="#ff9800")272        self._conn_btn.configure(state="normal", text="连接",273                                 fg_color="#2d7a2d", hover_color="#1f5c1f")274        self._log("连接意外断开，请检查设备")275276def _write_value(self, reg):277        entry = self._write_entries.get(reg.address)278if not entry:279return280281        raw_text = entry.get().strip()282if not raw_text:283            self._log(f"请输入{reg.name}的值")284return285286try:287            eng_val = float(raw_text)288except ValueError:289            self._log("输入值格式错误，请输入数字")290return291292        raw_val = self._data_model.engineering_to_raw(reg, eng_val)293294def do_write():295            ok, err = self._modbus.write_register(296                address=reg.address, value=raw_val, slave_id=self._slave_id297            )298            msg = f"写入[{reg.name}]={eng_val}{reg.unit} {'成功' if ok else '失败: ' + err}"299            self.after(0, self._log, msg)300301        threading.Thread(target=do_write, daemon=True).start()302303def _log(self, msg: str):304"""向日志框追加消息"""305import datetime306        ts = datetime.datetime.now().strftime("%H:%M:%S")307        self._log_box.insert("end", f"[{ts}] {msg}\n")308        self._log_box.see("end")  # 自动滚动到底部

第四步：程序入口

python1import logging2import sys3from pathlib import Path45PROJECT_ROOT = Path(__file__).resolve().parents[1]6if str(PROJECT_ROOT) not in sys.path:7    sys.path.insert(0, str(PROJECT_ROOT))89from modbustcpDemo.ui.main_window import MainWindow1011# 配置日志，方便调试时查问题12logging.basicConfig(13    level=logging.WARNING,14    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"15)16settings = {17"default_ip": "127.0.0.1",18"default_port": 502,19"slave_id": 1,20"window_width":800,21"window_height":60022}23if __name__ == "__main__":24    app = MainWindow(settings)25    app.mainloop()

⚠️ 几个必须知道的坑

线程安全是重中之重。 CustomTkinter和tkinter一样，所有UI更新必须在主线程执行。轮询线程里用self.after(0, callback)把更新任务丢回主线程，这个模式要形成肌肉记忆。直接在子线程里改Label的text，有时候能跑，有时候会莫名崩溃，而且复现困难。

批量读比逐个读快得多。 上面的代码为了清晰，是逐个寄存器读的。实际项目里，如果监控点多，建议把地址连续的寄存器合并成一次批量读——一次读20个寄存器，比发20次请求快一个数量级，也减少了设备侧的负载。

连接超时要设合理值。 timeout设太大，连接失败时界面会卡住很久，用户体验很差。我通常设3秒，配合把连接操作放到线程里，界面始终保持响应。

寄存器地址的0-based和1-based问题。 pymodbus使用0-based地址，但很多PLC文档和组态软件使用1-based（40001对应地址0）。对接新设备前，先用Modbus Poll这类工具验证地址，别上来就在代码里试。

🚀 进阶：让它更接近生产级

目前这个版本已经能用了，但距离真正的生产环境还有几步路：

断线重连机制——工控现场网络不稳定是常态。可以在_on_connection_lost里加一个自动重连的逻辑，比如每5秒尝试一次，重连成功后自动恢复轮询。

数据持久化——把采集到的数据写入SQLite或CSV，方便后续分析和追溯。pymodbus读到的数据加一个时间戳，存起来，现场排查问题时价值巨大。

报警阈值设置——当某个寄存器值超出范围时，在界面上变色提示，甚至触发声音报警。CustomTkinter的Label可以动态改text_color，实现起来很直接。

配置文件化——把IP、端口、寄存器映射表存到JSON或YAML文件里，不同设备切换时不用改代码，直接换配置文件。

完整项目已开源，包含模拟器测试脚本（不需要真实PLC也能跑起来验证界面）。欢迎在评论区分享你在工控项目里踩过的Modbus坑，或者你对这套架构的改进思路——这类经验，往往比文档更值钱。

#Python #工控开发 #Modbus #CustomTkinter #上位机开发