2026年9月15日 星期二

2026-09 mqtt 實驗1

 2026-09 mqtt 實驗1

MQTT (Message Queuing Telemetry Transport) 是一種輕量級的發佈/訂閱 (Publish/Subscribe) 訊息傳輸協定,專為網路頻寬有限、硬體資源受限或網路不穩定的物聯網 (IoT) 裝置與 M2M (Machine-to-Machine) 通訊而設計。

核心架構與運作機制

  • 發佈/訂閱模式 (Pub/Sub):傳送者(發佈者)與接收者(訂閱者)不直接對連,解耦了裝置間的時間與空間依賴。

  • 代理伺服器 (Broker):訊息的中繼站。負責接收發佈者的訊息、過濾並轉發給已訂閱該主題的客戶端(如 ESP32、Python 應用程式)。

  • 主題 (Topic):訊息的分類標籤,採階層式路徑設計(例如 home/bedroom/temperaturealex9ufo/2026/led)。

主要特性

  • 極輕量與高效能:標頭檔最小僅 2 Bytes,大幅降低網路封包負擔與裝置耗電量。

  • 基於 TCP/IP:確保資料傳輸的可靠性與順序性(預設 Port 為 1883,加密的 MQTTS 使用 Port 8883)。

  • 服務品質等級 (QoS, Quality of Service)

    • QoS 0 (At most once):最多一次,發後即忘(可能遺失)。

    • QoS 1 (At least once):至少一次,確保送達(可能重複)。

    • QoS 2 (Exactly once):恰好一次,確保不重覆且不遺失(傳送開銷最大)。

  • 遺言訊息 (Last Will and Testament, LWT):當裝置異常斷線時,Broker 會自動發佈事先預設好的訊息告知其他客戶端。

  • 保留訊息 (Retained Message):Broker 會儲存主題最後一筆訊息,新訂閱者連線時能立即取得當前最新的狀態。

典型應用場景

  • 物聯網與智慧家庭:感測器數據採集(溫濕度、電量)與家電遠端控制(如 ESP32 LED 控制)。

  • 車聯網與追蹤系統:車輛 GPS 定位與狀態回傳。

  • 即時通訊:部分行動裝置 App 的即時訊息推播服務。





















ESP32 程式功能說明

  • 網路與 MQTT 連線建立:開機後自動連線至 Wokwi 虛擬 Wi-Fi (Wokwi-GUEST),並連接至 mqttgo.io Broker (Port 1883)。連線成功後訂閱控制主題 alex9ufo/2026/led

  • 命令接收與模式切換 (Callback):收到指令後切換 currentMode 狀態,並立即向 alex9ufo/2026/ledstatus 主題回報最新狀態:

    • "on":將 GPIO 2 設為高電位(亮燈),回傳 "on_led"

    • "off":將 GPIO 2 設為低電位(熄滅),回傳 "off_led"

    • "flash":進入閃爍模式,回傳 "flash_led"

    • "timer(5Sec)":點亮 LED 並記錄起始時間,回傳 "timer(5Sec)_led"

  • 非阻塞式背景任務 (Loop):利用 millis() 時間差處理持續性動作,避免使用 delay() 導致 MQTT 斷線:

    • 閃爍模式:每 500 毫秒翻轉一次 GPIO 2 電位。

    • 定時模式:持續滿 5 秒後自動關閉 LED,將模式切回 OFF 並回傳 "off_led"

Python (Tkinter) 程式功能說明

  • 圖形化操作介面 (GUI):使用 Tkinter 打造控制面板,提供 ONOFFFLASHTimer (定時5秒) 四個控制按鈕。

  • 雙向 MQTT 通訊

    • 發送端:點擊按鈕時,將對應的指令字串(如 "on""flash")發送至 alex9ufo/2026/led 主題。

    • 接收端:訂閱 alex9ufo/2026/ledstatus 主題,即時接收 ESP32 回傳的當前狀態。

  • 狀態動態更新

    • MQTT 連線狀態:監聽 MQTT 連線狀態(on_connect / on_disconnect),在介面上以不同顏色顯示「連線中」、「已連線」或「已斷線」。

    • LED 實時狀態:接收到 ESP32 回傳訊息時,動態更新介面上的 LED 狀態 文字(例如 ESP32 定時 5 秒結束發送 "off_led" 時,介面也會同步更新為關閉狀態)。

  • 背景異步執行:採用 loop_start() 讓 MQTT 在獨立執行緒處理網路封包,確保 GUI 畫面順暢不卡頓。

Python+TKinter程式

import tkinter as tk

import paho.mqtt.client as mqtt


BROKER = "mqttgo.io"

PORT = 1883

TOPIC_PUB = "alex9ufo/2026/led"

# 修正重點:補上 /2026,確保與 ESP32 的 pub_topic 完全一致

TOPIC_SUB = "alex9ufo/2026/ledstatus"


# MQTT v2 Callback: 處理連線狀態

def on_connect(client, userdata, flags, reason_code, properties=None):

    if reason_code == 0:

        print(f"[系統日誌] 成功連線至 Broker,正在訂閱主題: {TOPIC_SUB}")

        mqtt_status_label.config(text="MQTT 連線狀態: 已連線", fg="green")

        client.subscribe(TOPIC_SUB)

    else:

        print(f"[錯誤日誌] 連線失敗,代碼: {reason_code}")

        mqtt_status_label.config(text=f"MQTT 連線失敗 (代碼: {reason_code})", fg="red")


# MQTT v2 Callback: 處理斷線狀態

def on_disconnect(client, userdata, disconnect_flags, reason_code, properties=None):

    print(f"[警告日誌] 已與 Broker 斷開連線,原因代碼: {reason_code}")

    mqtt_status_label.config(text="MQTT 連線狀態: 已斷線", fg="red")


# MQTT Callback: 接收 ESP32 回傳的狀態

def on_message(client, userdata, msg):

    try:

        payload = msg.payload.decode("utf-8")

        print(f"[收到訊息] 主題: {msg.topic} | 內容: {payload}")

        # 更新 UI

        led_status_label.config(text=f"LED 狀態: {payload}")

    except Exception as e:

        print(f"[錯誤日誌] 解析訊息失敗: {e}")


def send_cmd(cmd):

    print(f"[發送指令] 主題: {TOPIC_PUB} | 指令: {cmd}")

    client.publish(TOPIC_PUB, cmd)


# 初始化 MQTT (使用 Callback API v2)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)

client.on_connect = on_connect

client.on_disconnect = on_disconnect

client.on_message = on_message


# GUI 介面設計

root = tk.Tk()

root.title("ESP32 LED 控制器")

root.geometry("320x300")


# 顯示 MQTT 連線狀態

mqtt_status_label = tk.Label(root, text="MQTT 連線狀態: 連線中...", font=("Arial", 10), fg="orange")

mqtt_status_label.pack(pady=5)


# 顯示 LED 目前狀態

led_status_label = tk.Label(root, text="LED 狀態: 未知", font=("Arial", 12, "bold"), fg="blue")

led_status_label.pack(pady=10)


# 控制按鈕

btn_on = tk.Button(root, text="ON (開啟)", width=20, command=lambda: send_cmd("on"))

btn_on.pack(pady=5)


btn_off = tk.Button(root, text="OFF (關閉)", width=20, command=lambda: send_cmd("off"))

btn_off.pack(pady=5)


btn_flash = tk.Button(root, text="FLASH (閃爍)", width=20, command=lambda: send_cmd("flash"))

btn_flash.pack(pady=5)


btn_timer = tk.Button(root, text="Timer (定時5秒)", width=20, command=lambda: send_cmd("timer(5Sec)"))

btn_timer.pack(pady=5)


# 非阻塞方式連接 Broker

try:

    print(f"[系統日誌] 正在嘗試連線至 Broker: {BROKER}:{PORT}")

    client.connect_async(BROKER, PORT, 60)

    client.loop_start()

except Exception as e:

    print(f"[錯誤日誌] 連線異常: {e}")

    mqtt_status_label.config(text=f"連線異常: {e}", fg="red")


root.mainloop()





WOKWI   ESP32 程式

#include <WiFi.h>
#include <PubSubClient.h>

const char* ssid = "Wokwi-GUEST";
const char* password = "";

const char* mqtt_server = "mqttgo.io";
const int mqtt_port = 1883;

const char* sub_topic = "alex9ufo/2026/led";
const char* pub_topic = "alex9ufo/2026/ledstatus";

const int ledPin = 2;

WiFiClient espClient;
PubSubClient client(espClient);

enum Mode { OFF, ON, FLASH, TIMER };
Mode currentMode = OFF;

unsigned long lastFlashTime = 0;
unsigned long timerStartTime = 0;
bool ledState = LOW;

void setup_wifi() {
  delay(10);
  Serial.print("[Wi-Fi] 正在連線至 ");
  Serial.println(ssid);

  WiFi.begin(ssid, password);
  while (WiFi.status() != WL_CONNECTED) {
    delay(500);
    Serial.print(".");
  }

  Serial.println("\n[Wi-Fi] 已成功連線!");
  Serial.print("[Wi-Fi] IP 位址: ");
  Serial.println(WiFi.localIP());
}

void publishStatus(const char* payload) {
  client.publish(pub_topic, payload);
  Serial.print("[MQTT 發行] 主題: ");
  Serial.print(pub_topic);
  Serial.print(" | Payload: ");
  Serial.println(payload);
}

void callback(char* topic, byte* payload, unsigned int length) {
  String message;
  for (int i = 0; i < length; i++) {
    message += (char)payload[i];
  }

  Serial.print("[MQTT 接收] 主題: ");
  Serial.print(topic);
  Serial.print(" | Payload: ");
  Serial.println(message);

  if (message == "on") {
    currentMode = ON;
    digitalWrite(ledPin, HIGH);
    publishStatus("on_led");
  } else if (message == "off") {
    currentMode = OFF;
    digitalWrite(ledPin, LOW);
    publishStatus("off_led");
  } else if (message == "flash") {
    currentMode = FLASH;
    publishStatus("flash_led");
  } else if (message == "timer(5Sec)") {
    currentMode = TIMER;
    timerStartTime = millis();
    digitalWrite(ledPin, HIGH);
    publishStatus("timer(5Sec)_led");
  }
}

void reconnect() {
  while (!client.connected()) {
    Serial.print("[MQTT] 正在嘗試連線至 Broker: ");
    Serial.println(mqtt_server);
   
    String clientId = "ESP32Client-" + String(random(0xffff), HEX);
    if (client.connect(clientId.c_str())) {
      Serial.println("[MQTT] 連線成功!");
      Serial.print("[MQTT] 正在訂閱主題: ");
      Serial.println(sub_topic);
      client.subscribe(sub_topic);
    } else {
      Serial.print("[MQTT] 連線失敗, rc=");
      Serial.print(client.state());
      Serial.println(" 5 秒後重試...");
      delay(5000);
    }
  }
}

void setup() {
  Serial.begin(115200); // 初始化 Serial Monitor
 
  pinMode(ledPin, OUTPUT);
  digitalWrite(ledPin, LOW);
 
  setup_wifi();
  client.setServer(mqtt_server, mqtt_port);
  client.setCallback(callback);
}

void loop() {
  if (!client.connected()) {
    reconnect();
  }
  client.loop();

  unsigned long currentMillis = millis();

  if (currentMode == FLASH) {
    if (currentMillis - lastFlashTime >= 500) {
      lastFlashTime = currentMillis;
      ledState = !ledState;
      digitalWrite(ledPin, ledState);
    }
  } else if (currentMode == TIMER) {
    if (currentMillis - timerStartTime >= 5000) {
      digitalWrite(ledPin, LOW);
      currentMode = OFF;
      publishStatus("off_led");
    }
  }
}


2026年9月14日 星期一

EasyBuilder Pro 人機介面 HMI 編輯軟體)透過 Modbus TCP 控制研華 ADAM-6050

EasyBuilder Pro 人機介面 HMI 編輯軟體)透過 Modbus TCP 控制研華 ADAM-6050 












EasyBuilder Pro(威綸通 Weintek 人機介面 HMI 編輯軟體)透過 Modbus TCP 控制研華 ADAM-6050 的完整設定步驟如下:

一、 系統與網路參數設定

在進行 EasyBuilder Pro 設定前,請先確認兩者的網路參數在同一區域網路內:

  • ADAM-6050 IP192.168.50.122 / Port502

  • HMI (或 PC 模擬器) IP:建議設定為 192.168.50.X(例如 192.168.50.100

  • 子網路遮罩255.255.255.0

二、 EasyBuilder Pro 設備新增設定

  1. 開啟 EasyBuilder Pro 並建立新工程檔案(選擇對應的 HMI 型號或 PC 模擬)。

  2. 在上方選單點選 「常用」>「系統參數」(或雙擊右下角系統參數區)。

  3. 在「設備清單」分頁中,點擊右側的 「新增設備/伺服器...」

  4. 設定設備參數:

    • 設備名稱ADAM6050(可自訂)

    • 設備類型:選擇 MODBUS Server (ID Single) (TCP/IP)MODBUS RTU/TCP (Master)

    • 介面類型Ethernet

    • IP 位址:輸入 192.168.50.122

    • 埠號 (Port)502

    • 站號 (Device ID / Unit ID)1

  5. 點擊 「確定」 完成設備新增。

三、 Modbus 位址對照與元件設定

研華 ADAM-6050 在 EasyBuilder Pro 中的元件對照位址如下:

功能通道功能說明Modbus 位址類型EasyBuilder Pro 位址設定備註
DI 0 ~ DI 11數位輸入監控0x (Coil)1x (Discrete Input)0x - 1 ~ 0x - 12 (或 1x - 1)讀取 12 個通道狀態
DO 0 ~ DO 5數位輸出控制0x (Coil)0x - 17 ~ 0x - 22重點:起始位址從 17 開始 (Offset 16)

1. 設定 DI 監控燈號 (指示燈)

  • 上方選單點選 「物件」>「位元指示燈」

  • 讀取位址:選擇剛設定的設備 ADAM6050,位址類型選擇 0x(或 1x),位址填入 1(代表 DI 0)。

  • 複製該元件並將位址依序設為 2 (DI 1)、3 (DI 2)... 直至 12 (DI 11)。

2. 設定 DO 開關控制 (開關按鈕)

  • 上方選單點選 「物件」>「切換開關」 (Toggle Switch) 或 「位元按鈕」

  • 讀取/寫入位址:選擇設備 ADAM6050,位址類型選擇 0x,位址填入 17(代表 DO 0)。

  • 複製該元件並將位址依序設為 18 (DO 1)、19 (DO 2)... 直至 22 (DO 5)。

四、 測試與排查注意事項

  1. 關閉其他占用軟體:在執行 EasyBuilder Pro 在線模擬或 HMI 通訊前,務必先關閉研華 ADAMUtility 及 Python 程式,避免 Port 502 被獨佔導致 HMI 顯示「PLC No Response」(PLC 無回應)。

  2. 位址偏移說明:EasyBuilder Pro 的 Modbus 通訊通常為 1-based 位址格式,因此 ADAM-6050 的 DO 0(暫存器 0x0010 / 16)在 EasyBuilder Pro 中需設定為 0x - 17

ADAM 6050 ModBus TCP 控制程式

ADAM 6050  ModBus TCP 控制程式












一、 ADAM-6050 與 PC 網路設定說明

根據前面排查結果,您的硬體與網路參數如下:

  • ADAM-6050 IP192.168.50.122 / Port502

  • PC Wi-Fi IP192.168.50.136

  • 子網路遮罩 (Subnet Mask)255.255.255.0

  • 預設閘道 (Gateway)192.168.50.1

1. 網路實體與 IP 連線設定

  • 同網段原則:PC 的 IP (192.168.50.136) 與 ADAM-6050 (192.168.50.122) 前三碼必須完全相同 (192.168.50.x)。

  • 網卡切換建議

    • 若走 Wi-Fi 通訊:電腦已連上與 ADAM-6050 相同的 Wi-Fi 路由器,網路設定已正確。

    • 若用網線直連 PC 乙太網路卡:需進入「控制台 > 網路和共用中心 > 變更介面卡設定 > 乙太網路 2 > 內容 > IPv4」,將 PC 的實體網卡手動指定 IP(例如 192.168.50.200),避免出現 169.254.x.x (APIPA) 導致無法通訊。

2. Modbus 連線限制與常見卡關問題

  • 獨佔/連線數限制:ADAM-6050 支援的 Modbus TCP 連線數極少。執行 Python 程式前,請務必先關閉 Advantech ADAM Utility,否則 Utility 會佔用 Port 502,導致 Python 提示連線失敗或沒有回應。

  • DI (數位輸入) 接線特性

    • Dry Contact (無電壓接點):訊號線需短路閉合至 GND,邏輯電位才會轉變。

    • Wet Contact (有電壓接點):高電位 (ON) 需輸入 +10V ~ 30VDC 電壓;低電位 (OFF) 為 0 ~ 3VDC

二、 Python 程式內容與架構說明

程式包含三大核心層次:

1. 介面與切換 (GUI & Application State)

  • 採用 Tkinter 打造雙畫面架構:

    • 登入畫面 (Login Frame):驗證 Username (Admin) / Password (Admin),並提供動態修改 ADAM IP 與 Port 的輸入欄位。

    • 控制畫面 (Control Frame):顯示 12 組 DI 狀態燈、6 組 DO 狀態燈、全開/全關核取方塊(All Channels Output)、控制按鈕及退出的選項(Back to Login / Exit Application)。

2. 背景多線程輪詢 (Threading Polling Loop)

  • 避免 GUI 卡死:使用 threading.Thread 在背景開啟獨立線程,每隔 0.5 秒向 ADAM-6050 發送 Modbus 讀取請求。

  • 跨線程更新 UI:背景線程讀取到數據後,透過 self.after(0, self.update_gui_indicators) 將點燈與計數任務交還主 UI 線程執行,確保界面不會無回應。

3. Modbus TCP 映射與通訊協定 (Modbus Protocol & Mapping)

研華 ADAM-6050 的 Modbus Register 映射規則:

  • DI 0 ~ 11 (共 12 通道)

    • 使用 Modbus Function Code 02 (read_discrete_inputs) 或 Function Code 01 (read_coils)。

    • Modbus 位址:從 0 (對應 00001) 開始連續讀取 12 個 Bit。

  • DO 0 ~ 5 (共 6 通道)

    • 使用 Modbus Function Code 01 (read_coils) 與 Function Code 15 (write_coils)。

    • Modbus 位址:ADAM-6050 的 DO 在 Modbus 中有 16 位址偏移 (Offset 16),對應 00017~00022。讀取與寫入時傳遞的 address 必須帶入 16

import threading
import time
import tkinter as tk
from tkinter import messagebox

try:
    from pymodbus.client import ModbusTcpClient
except ImportError:
    ModbusTcpClient = None


class ADAM6050App(tk.Tk):

    def __init__(self):
        super().__init__()
        self.title("ADAM-6050 Modbus TCP Controller")
        self.geometry("750x700")
        self.configure(bg="#D4E6F1")

        # Modbus 設定
        self.host = "192.168.50.122"
        self.port = 502
        self.slave_id = 1
        self.client = None
        self.polling = False

        # 輪詢與錯誤計數
        self.poll_count = 0
        self.di_err_count = 0
        self.do_err_count = 0

        # 通道狀態 (12 個 DI, 6 個 DO)
        self.di_states = [False] * 12
        self.do_states = [False] * 6
        self.do_vars = [tk.BooleanVar(value=False) for _ in range(6)]
        self.all_channels_var = tk.BooleanVar(value=False)

        # 顯示登入畫面
        self.show_login_frame()

    # ================= 1. 登入畫面 =================
    def show_login_frame(self):
        self.clear_window()

        self.login_frame = tk.Frame(self, bg="#D4E6F1")
        self.login_frame.pack(expand=True)

        canvas = tk.Canvas(
            self.login_frame,
            width=90,
            height=90,
            bg="#D4E6F1",
            highlightthickness=0,
        )
        canvas.pack(pady=10)
        canvas.create_oval(5, 5, 85, 85, outline="#1B4F72", width=3)
        canvas.create_oval(30, 20, 60, 50, fill="#1B4F72", outline="")
        canvas.create_arc(
            15, 45, 75, 90, start=0, extent=180, fill="#1B4F72", outline=""
        )

        tk.Label(
            self.login_frame,
            text="ADAM-6050 IP",
            font=("Arial", 10, "bold"),
            bg="#D4E6F1",
        ).pack(anchor="w", padx=20)
        self.ip_entry = tk.Entry(
            self.login_frame, font=("Arial", 11), width=25, bd=2, relief="groove"
        )
        self.ip_entry.insert(0, self.host)
        self.ip_entry.pack(pady=(0, 8), padx=20)

        tk.Label(
            self.login_frame, text="Port", font=("Arial", 10, "bold"), bg="#D4E6F1"
        ).pack(anchor="w", padx=20)
        self.port_entry = tk.Entry(
            self.login_frame, font=("Arial", 11), width=25, bd=2, relief="groove"
        )
        self.port_entry.insert(0, str(self.port))
        self.port_entry.pack(pady=(0, 8), padx=20)

        tk.Label(
            self.login_frame, text="Username", font=("Arial", 10), bg="#D4E6F1"
        ).pack(anchor="w", padx=20)
        self.user_entry = tk.Entry(
            self.login_frame, font=("Arial", 11), width=25, bd=2, relief="groove"
        )
        self.user_entry.insert(0, "Admin")
        self.user_entry.pack(pady=(0, 8), padx=20)

        tk.Label(
            self.login_frame, text="Password", font=("Arial", 10), bg="#D4E6F1"
        ).pack(anchor="w", padx=20)
        self.pass_entry = tk.Entry(
            self.login_frame,
            font=("Arial", 11),
            show="*",
            width=25,
            bd=2,
            relief="groove",
        )
        self.pass_entry.insert(0, "Admin")
        self.pass_entry.pack(pady=(0, 15), padx=20)

        login_btn = tk.Button(
            self.login_frame,
            text="LOGIN",
            font=("Arial", 11, "bold"),
            bg="#1B4F72",
            fg="white",
            width=20,
            command=self.verify_login,
        )
        login_btn.pack(pady=5)

    def verify_login(self):
        username = self.user_entry.get().strip()
        password = self.pass_entry.get().strip()
        ip = self.ip_entry.get().strip()
        try:
            port = int(self.port_entry.get().strip())
        except ValueError:
            messagebox.showerror("Error", "Port 必須為數字!")
            return

        if username == "Admin" and password == "Admin":
            self.host = ip
            self.port = port

            if self.connect_modbus():
                self.show_control_frame()
            else:
                messagebox.showerror(
                    "Connection Error",
                    f"無法連線至 ADAM-6050 ({self.host}:{self.port})",
                )
        else:
            messagebox.showerror(
                "Error", "帳號或密碼錯誤!\n預設:Admin / Admin"
            )

    # ================= 2. ADAM-6050 控制畫面 =================
    def show_control_frame(self):
        self.clear_window()

        self.ctrl_frame = tk.Frame(self, bg="#C5D9E8", bd=2, relief="ridge")
        self.ctrl_frame.pack(fill="both", expand=True, padx=15, pady=15)

        # 1. DI Section
        di_header = tk.Label(
            self.ctrl_frame,
            text="Digital Input Values",
            font=("Arial", 12, "bold"),
            bg="#E67E22",
            fg="black",
        )
        di_header.pack(fill="x")

        self.di_status_lbl = tk.Label(
            self.ctrl_frame,
            text="Polling 0 times... DiErr2:0!",
            font=("Arial", 10),
            bg="#A9CCE3",
        )
        self.di_status_lbl.pack(fill="x")

        di_grid = tk.Frame(self.ctrl_frame, bg="#A9CCE3", bd=1, relief="solid")
        di_grid.pack(fill="x", padx=5, pady=5)

        self.di_canvas_list = []
        for i in range(12):
            col_frame = tk.Frame(
                di_grid, bg="#A9CCE3", bd=1, relief="solid", width=55, height=65
            )
            col_frame.grid(row=0, column=i, sticky="nsew", padx=1, pady=1)
            col_frame.grid_propagate(False)

            tk.Label(
                col_frame,
                text=f"DI\n{i}",
                font=("Arial", 9, "bold"),
                bg="#A9CCE3",
            ).pack()
            cv = tk.Canvas(
                col_frame,
                width=24,
                height=24,
                bg="#A9CCE3",
                highlightthickness=0,
            )
            cv.pack(pady=2)
            bulb = cv.create_oval(2, 2, 22, 22, fill="#BDC3C7", outline="#7F8C8D")
            self.di_canvas_list.append((cv, bulb))

        # 2. DO Read Section
        do_header = tk.Label(
            self.ctrl_frame,
            text="Digital Output Values",
            font=("Arial", 12, "bold"),
            bg="#E74C3C",
            fg="black",
        )
        do_header.pack(fill="x", pady=(10, 0))

        self.do_status_lbl = tk.Label(
            self.ctrl_frame,
            text="Polling 0 times... DoErr2:0!",
            font=("Arial", 10),
            bg="#A9CCE3",
        )
        self.do_status_lbl.pack(fill="x")

        do_grid = tk.Frame(self.ctrl_frame, bg="#A9CCE3", bd=1, relief="solid")
        do_grid.pack(fill="x", padx=5, pady=5)

        self.do_canvas_list = []
        for i in range(12):
            col_frame = tk.Frame(
                do_grid, bg="#A9CCE3", bd=1, relief="solid", width=55, height=65
            )
            col_frame.grid(row=0, column=i, sticky="nsew", padx=1, pady=1)
            col_frame.grid_propagate(False)

            if i < 6:
                tk.Label(
                    col_frame,
                    text=f"DO\n{i}",
                    font=("Arial", 9, "bold"),
                    bg="#A9CCE3",
                ).pack()
                cv = tk.Canvas(
                    col_frame,
                    width=24,
                    height=24,
                    bg="#A9CCE3",
                    highlightthickness=0,
                )
                cv.pack(pady=2)
                bulb = cv.create_oval(
                    2, 2, 22, 22, fill="#BDC3C7", outline="#7F8C8D"
                )
                self.do_canvas_list.append((cv, bulb))

        # 3. Output Control Section
        ctrl_header = tk.Frame(self.ctrl_frame, bg="#E74C3C")
        ctrl_header.pack(fill="x", pady=(10, 0))

        tk.Label(
            ctrl_header,
            text="All Channels Output:",
            font=("Arial", 11, "bold"),
            bg="#E74C3C",
            fg="white",
        ).pack(side="left", padx=10)
        chk_all = tk.Checkbutton(
            ctrl_header,
            variable=self.all_channels_var,
            bg="#E74C3C",
            activebackground="#E74C3C",
            command=self.toggle_all_channels,
        )
        chk_all.pack(side="left")

        do_ctrl_grid = tk.Frame(
            self.ctrl_frame, bg="#A9CCE3", bd=1, relief="solid"
        )
        do_ctrl_grid.pack(fill="x", padx=5, pady=5)

        for i in range(12):
            col_frame = tk.Frame(
                do_ctrl_grid,
                bg="#A9CCE3",
                bd=1,
                relief="solid",
                width=55,
                height=65,
            )
            col_frame.grid(row=0, column=i, sticky="nsew", padx=1, pady=1)
            col_frame.grid_propagate(False)

            if i < 6:
                chk = tk.Checkbutton(
                    col_frame, variable=self.do_vars[i], bg="#A9CCE3"
                )
                chk.pack()
                tk.Label(
                    col_frame,
                    text=f"DO\n{i}",
                    font=("Arial", 9, "bold"),
                    bg="#A9CCE3",
                ).pack()

        apply_btn = tk.Button(
            self.ctrl_frame,
            text="Apply Output",
            font=("Arial", 12, "bold"),
            bg="#E0E0E0",
            command=self.apply_output,
        )
        apply_btn.pack(pady=15, ipadx=20, ipady=5)

        nav_frame = tk.Frame(self.ctrl_frame, bg="#C5D9E8")
        nav_frame.pack(pady=5)

        back_btn = tk.Button(
            nav_frame,
            text="Back to Login",
            font=("Arial", 10, "underline"),
            fg="blue",
            bg="#C5D9E8",
            bd=0,
            cursor="hand2",
            command=self.logout,
        )
        back_btn.pack(side="left", padx=15)

        exit_btn = tk.Button(
            nav_frame,
            text="Exit Application",
            font=("Arial", 10, "bold"),
            fg="red",
            bg="#C5D9E8",
            bd=0,
            cursor="hand2",
            command=self.on_closing,
        )
        exit_btn.pack(side="left", padx=15)

    # ================= 3. MODBUS 修正邏輯 =================
    def connect_modbus(self):
        if ModbusTcpClient is None:
            return False
        try:
            self.client = ModbusTcpClient(
                host=self.host, port=self.port, timeout=2.0
            )
            if self.client.connect():
                self.polling = True
                self.poll_thread = threading.Thread(
                    target=self.poll_modbus_loop, daemon=True
                )
                self.poll_thread.start()
                return True
            return False
        except Exception:
            return False

    def _modbus_read_di(self):
        """相容不同 pymodbus 版本的 DI 讀取"""
        try:
            return self.client.read_discrete_inputs(
                address=0, count=12, slave=self.slave_id
            )
        except TypeError:
            try:
                return self.client.read_discrete_inputs(address=0, count=12)
            except Exception:
                return self.client.read_coils(address=0, count=12)

    def _modbus_read_do(self):
        """相容不同 pymodbus 版本的 DO 讀取 (ADAM-6050 DO 位址為 16)"""
        try:
            return self.client.read_coils(
                address=16, count=6, slave=self.slave_id
            )
        except TypeError:
            return self.client.read_coils(address=16, count=6)

    def _modbus_write_do(self, values):
        """相容不同 pymodbus 版本的 DO 寫入 (ADAM-6050 DO 位址為 16)"""
        try:
            return self.client.write_coils(
                address=16, values=values, slave=self.slave_id
            )
        except TypeError:
            return self.client.write_coils(address=16, values=values)

    def poll_modbus_loop(self):
        while self.polling:
            self.poll_count += 1

            if self.client and self.client.is_socket_open():
                try:
                    # 1. 讀取 DI
                    rr_di = self._modbus_read_di()
                    if rr_di and not rr_di.isError():
                        self.di_states = rr_di.bits[:12]
                    else:
                        self.di_err_count += 1

                    # 2. 讀取 DO
                    rr_do = self._modbus_read_do()
                    if rr_do and not rr_do.isError():
                        self.do_states = rr_do.bits[:6]
                    else:
                        self.do_err_count += 1

                except Exception:
                    self.di_err_count += 1
                    self.do_err_count += 1
            else:
                self.di_err_count += 1
                self.do_err_count += 1

            self.after(0, self.update_gui_indicators)
            time.sleep(0.5)

    def update_gui_indicators(self):
        if not hasattr(self, "di_status_lbl"):
            return

        self.di_status_lbl.config(
            text=f"Polling {self.poll_count} times... DiErr2:{self.di_err_count}!"
        )
        self.do_status_lbl.config(
            text=f"Polling {self.poll_count} times... DoErr2:{self.do_err_count}!"
        )

        for i, (cv, bulb) in enumerate(self.di_canvas_list):
            color = "#F1C40F" if self.di_states[i] else "#BDC3C7"
            cv.itemconfig(bulb, fill=color)

        for i, (cv, bulb) in enumerate(self.do_canvas_list):
            color = "#F1C40F" if self.do_states[i] else "#BDC3C7"
            cv.itemconfig(bulb, fill=color)

    def toggle_all_channels(self):
        state = self.all_channels_var.get()
        for var in self.do_vars:
            var.set(state)

    def apply_output(self):
        values = [var.get() for var in self.do_vars]
        if self.client and self.client.is_socket_open():
            try:
                res = self._modbus_write_do(values)
                if res and not res.isError():
                    messagebox.showinfo(
                        "Success", f"成功寫入 DO 狀態:\n{values}"
                    )
                else:
                    messagebox.showerror("Modbus Error", "寫入 DO 失敗!")
            except Exception as e:
                messagebox.showerror(
                    "Modbus Error", f"控制 ADAM-6050 發生錯誤:\n{e}"
                )
        else:
            messagebox.showwarning("Warning", "目前未連線至 ADAM-6050 設備!")

    def logout(self):
        self.polling = False
        if self.client:
            self.client.close()
        self.show_login_frame()

    def clear_window(self):
        for widget in self.winfo_children():
            widget.destroy()

    def on_closing(self):
        self.polling = False
        if self.client:
            self.client.close()
        self.destroy()


if __name__ == "__main__":
    app = ADAM6050App()
    app.protocol("WM_DELETE_WINDOW", app.on_closing)
    app.mainloop()

2026-09 mqtt 實驗1

 2026-09 mqtt 實驗1 MQTT (Message Queuing Telemetry Transport) 是一種輕量級的 發佈/訂閱 (Publish/Subscribe) 訊息傳輸協定,專為網路頻寬有限、硬體資源受限或網路不穩定的物聯網 (IoT) 裝置與 ...