# mqtt_gui.py # pip install paho-mqtt import json import re import tkinter as tk from tkinter import ttk, messagebox from paho.mqtt import client as mqtt DEFAULT_HOST = "169.254.11.130" DEFAULT_PORT = 1886 DEFAULT_TOPIC = "topic_plc_and_py_for_AXIS" DEFAULT_QOS = 1 def robust_json_parse(raw: str): """ 將 PLC 常見不標準 JSON 修正後解析成 dict。 例:'{Forward_RPM:1500,UPDown_RPM:800}' -> {'Forward_RPM':1500,'UPDown_RPM':800} 也會把 TRUE/FALSE -> true/false;補 key 雙引號;去外層引號。 """ s = raw.strip() # 去掉最外層引號(若有) if (s.startswith("'") and s.endswith("'")) or (s.startswith('"') and s.endswith('"')): s = s[1:-1].strip() # PLC 常見布林大寫 -> JSON 規範小寫 s = s.replace("TRUE", "true").replace("FALSE", "false") # 自動補 key 的雙引號:{key: -> {"key": s = re.sub(r'([{,]\s*)([A-Za-z_]\w*)\s*:', r'\1"\2":', s) # 將單引號字串轉雙引號(在補 key 之後) s = s.replace("'", '"') return json.loads(s) class MQTTGui: def __init__(self, root: tk.Tk): self.root = root self.root.title("MQTT PLC Subscriber") # 狀態 self.client: mqtt.Client | None = None self.connected = False # UI 變數 self.var_host = tk.StringVar(value=DEFAULT_HOST) self.var_port = tk.IntVar(value=DEFAULT_PORT) self.var_topic = tk.StringVar(value=DEFAULT_TOPIC) self.var_qos = tk.IntVar(value=DEFAULT_QOS) self.var_button_y = tk.StringVar(value="None") self.var_forward_rpm = tk.StringVar(value="None") self.var_updown_rpm = tk.StringVar(value="None") self.var_status = tk.StringVar(value="Disconnected") self._build_ui() # 關閉事件 self.root.protocol("WM_DELETE_WINDOW", self.on_close) def _build_ui(self): pad = {"padx": 8, "pady": 6} frm_top = ttk.LabelFrame(self.root, text="Connection") frm_top.grid(row=0, column=0, sticky="nsew", **pad) ttk.Label(frm_top, text="Broker Host").grid(row=0, column=0, sticky="e") ttk.Entry(frm_top, textvariable=self.var_host, width=18).grid(row=0, column=1, sticky="w") ttk.Label(frm_top, text="Port").grid(row=0, column=2, sticky="e") ttk.Spinbox(frm_top, from_=1, to=65535, textvariable=self.var_port, width=8).grid(row=0, column=3, sticky="w") ttk.Label(frm_top, text="Topic").grid(row=1, column=0, sticky="e") ttk.Entry(frm_top, textvariable=self.var_topic, width=40).grid(row=1, column=1, columnspan=3, sticky="w") ttk.Label(frm_top, text="QoS").grid(row=2, column=0, sticky="e") ttk.Spinbox(frm_top, values=(0,1,2), textvariable=self.var_qos, width=8).grid(row=2, column=1, sticky="w") btn_conn = ttk.Button(frm_top, text="Connect", command=self.connect) btn_dis = ttk.Button(frm_top, text="Disconnect", command=self.disconnect) btn_conn.grid(row=2, column=2, sticky="we") btn_dis.grid(row=2, column=3, sticky="we") ttk.Label(frm_top, text="Status:").grid(row=3, column=0, sticky="e") ttk.Label(frm_top, textvariable=self.var_status).grid(row=3, column=1, columnspan=3, sticky="w") frm_vals = ttk.LabelFrame(self.root, text="Current State") frm_vals.grid(row=1, column=0, sticky="nsew", **pad) ttk.Label(frm_vals, text="BUTTON_Y").grid(row=0, column=0, sticky="e") ttk.Label(frm_vals, textvariable=self.var_button_y, width=20).grid(row=0, column=1, sticky="w") ttk.Label(frm_vals, text="Forward_RPM").grid(row=1, column=0, sticky="e") ttk.Label(frm_vals, textvariable=self.var_forward_rpm, width=20).grid(row=1, column=1, sticky="w") ttk.Label(frm_vals, text="UPDown_RPM").grid(row=2, column=0, sticky="e") ttk.Label(frm_vals, textvariable=self.var_updown_rpm, width=20).grid(row=2, column=1, sticky="w") frm_log = ttk.LabelFrame(self.root, text="Logs") frm_log.grid(row=2, column=0, sticky="nsew", **pad) frm_log.rowconfigure(0, weight=1) frm_log.columnconfigure(0, weight=1) self.txt_log = tk.Text(frm_log, height=14) self.txt_log.grid(row=0, column=0, sticky="nsew") scr = ttk.Scrollbar(frm_log, command=self.txt_log.yview) scr.grid(row=0, column=1, sticky="ns") self.txt_log.configure(yscrollcommand=scr.set) # 讓整體可縮放 self.root.rowconfigure(2, weight=1) self.root.columnconfigure(0, weight=1) # ---------- MQTT ---------- def connect(self): if self.connected: self._log("Already connected.") return host = self.var_host.get().strip() port = int(self.var_port.get()) topic = self.var_topic.get().strip() qos = int(self.var_qos.get()) if not host or not topic: messagebox.showwarning("Warning", "Host 與 Topic 不可為空") return # 建立 client self.client = mqtt.Client(client_id="gui_sub_plc", protocol=mqtt.MQTTv311) # 自動重連延遲 self.client.reconnect_delay_set(min_delay=1, max_delay=10) # Callbacks self.client.on_connect = self._on_connect self.client.on_message = self._on_message self.client.on_disconnect = self._on_disconnect try: self.client.connect(host, port, keepalive=60) except Exception as e: self._log(f"❌ Connect failed: {e}") self.var_status.set("Connect failed") return # 非阻塞背景 loop self.client.loop_start() self.var_status.set("Connecting...") self._log(f"🔌 Connecting to {host}:{port}, topic={topic}, qos={qos}") def disconnect(self): if self.client is None: return try: self.client.loop_stop() self.client.disconnect() except Exception as e: self._log(f"Disconnect error: {e}") def _on_connect(self, client, userdata, flags, reason_code, properties=None): if reason_code == 0: self.connected = True self.var_status.set("Connected") topic = self.var_topic.get().strip() qos = int(self.var_qos.get()) client.subscribe(topic, qos=qos) self._log(f"✅ Connected. Subscribed: {topic} (QoS={qos})") else: self.connected = False self.var_status.set(f"Connect failed ({reason_code})") self._log(f"❌ Connect failed. reason_code={reason_code}") def _on_disconnect(self, client, userdata, reason_code, properties=None): self.connected = False self.var_status.set(f"Disconnected ({reason_code})") self._log(f"🔌 Disconnected. reason_code={reason_code}") def _on_message(self, client, userdata, msg: mqtt.MQTTMessage): payload_raw = msg.payload.decode("utf-8", errors="replace") self._log(f"📥 Raw: {payload_raw}") # 先試標準 JSON try: data = json.loads(payload_raw) except json.JSONDecodeError: # 失敗再容錯解析 try: data = robust_json_parse(payload_raw) except Exception as e: self._log(f"❌ Parse error: {e}") return self._log(f"✅ Parsed: {data}") # 更新顯示(有傳才改) if "BUTTON_Y" in data: self.var_button_y.set(str(data["BUTTON_Y"])) if "Forward_RPM" in data: self.var_forward_rpm.set(str(data["Forward_RPM"])) if "UPDown_RPM" in data: self.var_updown_rpm.set(str(data["UPDown_RPM"])) # ---------- Utils ---------- def _log(self, text: str): self.txt_log.insert("end", text + "\n") self.txt_log.see("end") def on_close(self): try: if self.client: self.client.loop_stop() self.client.disconnect() except Exception: pass self.root.destroy() if __name__ == "__main__": root = tk.Tk() # Windows 讓標準字體稍微大一點 try: from ctypes import windll windll.shcore.SetProcessDpiAwareness(1) # 可忽略失敗 except Exception: pass MQTTGui(root) root.mainloop()