python实现读取modbus485协议并打包exe客户端（wxpython+pyinstaller模块）

博主工作过程中接触到物联网，涉及modbus，mqtt等协议，想着python可以用来读取解析消息内容，实施过程中现场环境存在配置问题，那就开发一个客户端来帮助定位问题

客户端模块是 wxpython，协议对接用了 pymodbus 和 paho-mqtt，打包则使用pyinstaller

代码如下，实现了modbus的读取寄存器功能，后续待补充

import wx
import json
import time
import paho.mqtt.client as mqtt
from datetime import datetime
from pymodbus.client.sync import ModbusTcpClient
from pymodbus.exceptions import ConnectionException


class RunClient(wx.Frame):
    def __init__(self, *args, **kw):
        super(RunClient, self).__init__(*args, **kw)
        self.app = wx.App()
        self.title = "检测工具"

        self.mqtt_data_format = "({time})-({topic})-({data})"
        self.mqtt_data = []

        self.message_caption = "Info"

        self.ID_MENU_CLOSE = 99
        self.ID_MENU_RUN = 1

        self.ID_PANEL_MESSAGE_TYPE = 101
        self.ID_PANEL_MODBUS_TYPE = 102
        self.ID_PANEL_MODBUS_SLAVE = 103
        self.ID_PANEL_MODBUS_FUNCTION = 104
        self.ID_PANEL_MODBUS_HOST = 105
        self.ID_PANEL_MODBUS_ADDRESS = 106
        self.ID_PANEL_MODBUS_PORT = 107
        self.ID_PANEL_MODBUS_QUANTITY = 108
        self.ID_PANEL_MODBUS_INFO = 109

        self.check_msg_format = "{} 必须数字"

        self.init_ui()

    def init_ui(self):
        """初始化ui"""
        self.init_menu_bar()
        self.init_panel()

        self.CreateStatusBar()

        self.SetSize((500, 600))
        self.SetTitle(self.title)
        self.Centre()

    @property
    def message_type(self):
        """定义消息类型"""
        return [
            "Modbus Client",
            "MQTT Client"
        ]

    @property
    def modbus_functions(self):
        """定义modbus方法"""
        return [
            "03 Read Holding Registers"
        ]

    @property
    def modbus_type(self):
        """定义modbus类型"""
        return [
            "Modbus Tcp"
        ]

    def init_menu_bar(self):
        """初始化菜单按钮"""
        menu_bar = wx.MenuBar()

        # 定义菜单
        menu = wx.Menu()
        menu.Append(self.ID_MENU_RUN, "运行")
        menu.Append(self.ID_MENU_CLOSE, "关闭")

        menu_bar.Append(menu, "&开始")
        self.SetMenuBar(menu_bar)

        # 函数绑定菜单ID
        self.Bind(wx.EVT_MENU, self.run_info, id=self.ID_MENU_RUN)
        self.Bind(wx.EVT_MENU, self.button_close, id=self.ID_MENU_CLOSE)

    def init_panel(self):
        """初始panel"""
        panel = wx.Panel(self)

        # 10+60=70
        wx.StaticBox(panel, label="协议类型", pos=(10, 10), size=(450, 60))
        wx.ComboBox(panel, pos=(20, 30), choices=self.message_type, size=(200, -1), id=self.ID_PANEL_MESSAGE_TYPE, value=self.message_type[0])

        # 75+60=135
        wx.StaticBox(panel, label="Modbus来源类型", pos=(10, 75), size=(450, 60))
        wx.ComboBox(panel, pos=(20, 95), choices=self.modbus_type, size=(200, -1), id=self.ID_PANEL_MODBUS_TYPE, value=self.modbus_type[0])

        # 140+240=380
        wx.StaticBox(panel, label="Modbus读取参数", pos=(10, 140), size=(450, 240))
        wx.StaticText(panel, label="slave：", pos=(20, 160))
        wx.SpinCtrl(panel, pos=(242, 160), size=(200, -1), min=1, max=255, id=self.ID_PANEL_MODBUS_SLAVE, value="1")

        wx.StaticText(panel, label="function：", pos=(20, 190))
        wx.ComboBox(panel, pos=(240, 190), choices=self.modbus_functions, size=(200, -1), id=self.ID_PANEL_MODBUS_FUNCTION, value=self.modbus_functions[0])

        wx.StaticText(panel, label="address：", pos=(20, 220))
        wx.TextCtrl(panel, pos=(240, 220), size=(200, -1), id=self.ID_PANEL_MODBUS_ADDRESS, value="0")

        wx.StaticText(panel, label="quantity：", pos=(20, 250))
        wx.TextCtrl(panel, pos=(240, 250), size=(200, -1), id=self.ID_PANEL_MODBUS_QUANTITY, value="10")

        wx.StaticText(panel, label="host：", pos=(20, 280))
        wx.TextCtrl(panel, pos=(240, 280), size=(200, -1), id=self.ID_PANEL_MODBUS_HOST, value="127.0.0.1")

        wx.StaticText(panel, label="port：", pos=(20, 310))
        wx.TextCtrl(panel, pos=(240, 310), size=(200, -1), id=self.ID_PANEL_MODBUS_PORT, value="512")

        # 385+120=505
        wx.StaticBox(panel, label="Modbus返回信息", pos=(10, 385), size=(450, 120))
        wx.TextCtrl(panel, pos=(20, 405), size=(430, 90), id=self.ID_PANEL_MODBUS_INFO, style=wx.TE_MULTILINE, value="")

    def check_field_int(self, name, value):
        """检查字段INT"""
        try:
            value = int(value)
            return {
                "status": True,
                "value": value
            }
        except ValueError:
            return {
                "status": False,
                "msg": self.check_msg_format.format(name)
            }

    def mqtt_on_message(self, client, userdata, msg):
        """mqtt消息订阅"""
        data = json.loads(msg.payload)
        self.mqtt_data.append(self.mqtt_data_format.format(
            time=datetime.strftime(datetime.now(), "%d-%m-%Y %H:%M:%S"),
            topic=msg.topic,
            data=data
        ))

    def run_modbus_tcp(self, slave, address, quantity, host, port, function, timeout=3):
        """运行modbus tcp"""
        master = ModbusTcpClient(host=host, port=port, timeout=timeout)

        # 校验参数是否int
        resp = self.check_field_int(name="slave", value=slave)
        if not resp["status"]:
            return resp["msg"]
        resp = self.check_field_int(name="address", value=address)
        if not resp["status"]:
            return resp["msg"]
        resp = self.check_field_int(name="quantity", value=quantity)
        if not resp["status"]:
            return resp["msg"]

        results = []
        try:
            # 只写了一种情况
            if function == self.modbus_functions[0]:
                result = master.read_holding_registers(address=int(quantity), count=int(quantity), unit=int(slave))
                results = result.registers
                return results
            return results
        except AttributeError as e:
            return e
        except ConnectionException as e:
            return e

    def run_mqtt(self, topic, port):
        """运行mqtt"""
        client = mqtt.Client()
        client.connect(host=topic, port=port, keepalive=60)
        client.subscribe(topic)
        client.on_message = self.mqtt_on_message

        client.loop_start()
        time.sleep(20)
        client.loop_stop()

    def run_info(self, e):
        """运行"""
        modbus_type = self.FindWindowById(id=self.ID_PANEL_MODBUS_TYPE).GetValue()
        modbus_slave = self.FindWindowById(id=self.ID_PANEL_MODBUS_SLAVE).GetValue()
        modbus_function = self.FindWindowById(id=self.ID_PANEL_MODBUS_FUNCTION).GetValue()
        modbus_address = self.FindWindowById(id=self.ID_PANEL_MODBUS_ADDRESS).GetValue()
        modbus_quantity = self.FindWindowById(id=self.ID_PANEL_MODBUS_QUANTITY).GetValue()
        modbus_host = self.FindWindowById(id=self.ID_PANEL_MODBUS_HOST).GetValue()
        modbus_port = self.FindWindowById(id=self.ID_PANEL_MODBUS_PORT).GetValue()
        modbus_info = self.FindWindowById(id=self.ID_PANEL_MODBUS_INFO)

        results = self.run_modbus_tcp(
            slave=modbus_slave, address=modbus_address, quantity=modbus_quantity,
            host=modbus_host, port=modbus_port, function=modbus_function
        )

        modbus_info.SetValue(str(results))

        self.show_message()

    def show_message(self):
        """显示message"""
        wx.MessageBox(message="运行结束", caption=self.message_caption, style=wx.OK | wx.ICON_INFORMATION)

    def button_close(self, e):
        """按钮-关闭"""
        self.Close(True)


# 主函数
def main():

    app = wx.App()
    run = RunClient(None)
    run.Show()
    app.MainLoop()


if __name__ == '__main__':
    main()

运行效果：

打开的效果

运行后

 

如果需要打包则

pyinstall -F -w xx.py

具体参数可以自行查阅相应文档

打包后文件目录如下

 在dist下有相应的 exe文件