alex9ufo 聰明人求知心切: Adam Modbus

顯示具有 Adam Modbus 標籤的文章。顯示所有文章

2026年4月25日星期六

Adam DO5 DI12 + Python + Node-Red

Adam DO*5 DI*12 + Python + Node-Red

為了讓您的系統穩定運行，建議的最終接線如下：

功能	ADAM 端子	對接目標
DI 輸入 (乾接點)	`DI0` ~ `DI11`	開關的一端
DI 共同端	DI.COM	開關的另一端 (接 GND)
DO 輸出	`DO0` ~ `DO5`	負載 (如繼電器、指示燈)
DO 共同端	DO.COM	外部電源的+極 (Open collector)

Python 程式

import tkinter as tk

from tkinter import messagebox

from pyModbusTCP.client import ModbusClient

import paho.mqtt.client as mqtt

import json

import threading

import time

# --- Topic 定義 (兩者皆為發布) ---

TOPIC_DI_PUBLISH = "alex9ufo/adam6050/input" # 程式發行 DI 狀態

TOPIC_DO_PUBLISH = "alex9ufo/adam6050/output" # 程式發行 DO 狀態

PUBLISH_INTERVAL = 5 # 每 5 秒強制發行一次

class Adam6050DualPublishHmi:

def __init__(self, root):

self.root = root

self.root.title("ADAM-6050 雙向發行控制面板 (alex9ufo)")

self.root.geometry("400x450")

# 1. 通訊初始化

self.modbus_client = ModbusClient(port=502, timeout=2.0, auto_open=False)

try:

self.mqtt_client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION1)

except:

self.mqtt_client = mqtt.Client()

self.mqtt_client.on_connect = self.on_mqtt_connect

self.mqtt_client.on_disconnect = self.on_mqtt_disconnect

# 2. 狀態與緩存

self.modbus_active = False

self.mqtt_active = False

self.last_di_data = [] # 用於偵測 DI 變化

self.last_pub_time = 0 # 上次發行時間

self.di_labels = []

self.do_vars = []

self.setup_ui()

self.root.protocol("WM_DELETE_WINDOW", self.on_closing)

def setup_ui(self):

# --- 連線設定區域 ---

cfg_frame = tk.LabelFrame(self.root, text="通訊設定", padx=10, pady=10)

cfg_frame.pack(padx=15, pady=10, fill="x")

# Adam IP

tk.Label(cfg_frame, text="Adam IP:").grid(row=0, column=0)

self.adam_ip_entry = tk.Entry(cfg_frame, width=15)

self.adam_ip_entry.insert(0, "192.168.50.122")

self.adam_ip_entry.grid(row=0, column=1)

self.adam_btn = tk.Button(cfg_frame, text="連線 Adam", command=self.toggle_adam, width=10)

self.adam_btn.grid(row=0, column=2, padx=5)

self.adam_led = tk.Canvas(cfg_frame, width=15, height=15, highlightthickness=0)

self.adam_led.grid(row=0, column=3)

self.adam_circle = self.adam_led.create_oval(2,2,13,13, fill="gray")

# MQTT Broker

tk.Label(cfg_frame, text="Broker:").grid(row=1, column=0)

self.mqtt_entry = tk.Entry(cfg_frame, width=15)

self.mqtt_entry.insert(0, "broker.mqtt-dashboard.com")

self.mqtt_entry.grid(row=1, column=1)

self.mqtt_btn = tk.Button(cfg_frame, text="連線 MQTT", command=self.toggle_mqtt, width=10)

self.mqtt_btn.grid(row=1, column=2, padx=5)

self.mqtt_led = tk.Canvas(cfg_frame, width=15, height=15, highlightthickness=0)

self.mqtt_led.grid(row=1, column=3)

self.mqtt_circle = self.mqtt_led.create_oval(2,2,13,13, fill="gray")

# --- DI 顯示區域 ---

di_frame = tk.LabelFrame(self.root, text="DI 狀態 (乾接點: DI 短路至 GND 觸發)", padx=10, pady=10)

di_frame.pack(padx=15, pady=5, fill="both", expand=True)

for i in range(12):

lbl = tk.Label(di_frame, text=f"DI_{i:02d}", width=8, height=2, bg="#333333", fg="white", font=("Arial", 9, "bold"))

lbl.grid(row=i // 4, column=i % 4, padx=5, pady=5)

self.di_labels.append(lbl)

# --- DO 控制區域 ---

do_frame = tk.LabelFrame(self.root, text="DO 控制 (點擊即發行至 /output)", padx=10, pady=10)

do_frame.pack(padx=15, pady=5, fill="both", expand=True)

for i in range(6):

var = tk.BooleanVar()

# 當畫面上勾選框改變時，執行 publish_do_and_sync

chk = tk.Checkbutton(do_frame, text=f"DO_{i:02d}", variable=var, command=self.publish_do_and_sync)

chk.grid(row=i // 3, column=i % 3, padx=15, pady=10)

self.do_vars.append(var)

# --- Adam 控制邏輯 ---

def toggle_adam(self):

if not self.modbus_active:

self.modbus_client.host = self.adam_ip_entry.get().strip()

if self.modbus_client.open():

self.modbus_active = True

self.adam_btn.config(text="斷開 Adam", bg="#ffcccc")

threading.Thread(target=self.adam_polling, daemon=True).start()

else:

messagebox.showerror("錯誤", "無法連線到 ADAM-6050")

else:

self.modbus_active = False

self.modbus_client.close()

self.adam_btn.config(text="連線 Adam", bg="#f0f0f0")

self.set_led(self.adam_led, self.adam_circle, "gray")

def adam_polling(self):

"""偵測實體 DI 變化並發行"""

while self.modbus_active:

di_data = self.modbus_client.read_discrete_inputs(0, 12)

if di_data is not None:

self.set_led(self.adam_led, self.adam_circle, "#00FF00")

# UI 更新顏色

for i, val in enumerate(di_data):

self.di_labels[i].config(bg="#4CAF50" if val else "#333333")

# 偵測 DI 變化，若有變則立即發行

if di_data != self.last_di_data:

self.publish_status(TOPIC_DI_PUBLISH, "DI_STATUS", "di_all", di_data)

self.last_di_data = di_data

else:

self.set_led(self.adam_led, self.adam_circle, "#F44336")

time.sleep(0.5)

# --- MQTT 邏輯 ---

def toggle_mqtt(self):

if not self.mqtt_active:

try:

self.mqtt_client.connect(self.mqtt_entry.get().strip(), 1883, 60)

self.mqtt_client.loop_start()

# 啟動定時發行執行緒 (5秒一次)

threading.Thread(target=self.publish_timer, daemon=True).start()

except:

messagebox.showerror("錯誤", "無法連線到 MQTT Broker")

else:

self.mqtt_active = False

self.mqtt_client.loop_stop()

def publish_timer(self):

"""每 5 秒強制發行一次 DI 與 DO 的全狀態"""

while self.mqtt_active:

# 強制發行 DI

if self.last_di_data:

self.publish_status(TOPIC_DI_PUBLISH, "DI_SNAPSHOT", "di_all", self.last_di_data)

# 強制發行 DO

do_states = [var.get() for var in self.do_vars]

self.publish_status(TOPIC_DO_PUBLISH, "DO_SNAPSHOT", "do_all", do_states)

time.sleep(PUBLISH_INTERVAL)

def publish_do_and_sync(self):

"""當使用者操作 DO 時，同步實體設備並發行訊息"""

do_states = [var.get() for var in self.do_vars]

# 1. 實體 Modbus 控制

if self.modbus_active and self.modbus_client.is_open:

self.modbus_client.write_multiple_coils(16, do_states)

# 2. 發行到 alex9ufo/adam6050/output

self.publish_status(TOPIC_DO_PUBLISH, "DO_CHANGE", "do_all", do_states)

def publish_status(self, topic, msg_type, key_name, data_list):

"""通用的發行 JSON 函式"""

if self.mqtt_active:

payload = {

"adam_ip": self.adam_ip_entry.get().strip(),

"type": msg_type,

key_name: data_list,

"timestamp": time.strftime("%H:%M:%S")

}

self.mqtt_client.publish(topic, json.dumps(payload), qos=0)

print(f"Published to {topic}: {msg_type}")

# --- 通用處理 ---

def on_mqtt_connect(self, client, userdata, flags, rc):

if rc == 0:

self.mqtt_active = True

self.mqtt_btn.config(text="斷開 MQTT", bg="#ffcccc")

self.set_led(self.mqtt_led, self.mqtt_circle, "#00FF00")

else:

self.set_led(self.mqtt_led, self.mqtt_circle, "red")

def on_mqtt_disconnect(self, client, userdata, rc):

self.mqtt_active = False

self.mqtt_btn.config(text="連線 MQTT", bg="#f0f0f0")

self.set_led(self.mqtt_led, self.mqtt_circle, "gray")

def set_led(self, canvas, item, color):

canvas.itemconfig(item, fill=color)

def on_closing(self):

self.modbus_active = False

self.mqtt_active = False

self.modbus_client.close()

self.mqtt_client.disconnect()

self.root.destroy()

if __name__ == "__main__":

root = tk.Tk()

app = Adam6050DualPublishHmi(root)

root.mainloop()

Node-RED 邏輯流程設計

我們需要兩個核心邏輯：

DI 監測流：訂閱 .../input -> 解析 di_all 陣列 -> 分送至 12 個 LED。
DO 監測流：訂閱 .../output -> 解析 do_all 陣列 -> 分送至 6 個 LED。

Function 節點 (解析 DI)：使用一個 Function 節點將陣列拆分成多個輸出訊息。

JavaScript
// 將 di_all 陣列內容分別傳送到 12 個輸出埠
var di = msg.payload.di_all;
var messages = [];
for (var i = 0; i < 12; i++) {
    messages.push({ payload: di[i] });
}
return messages;

重要設定：在該 Function 節點的設定下方，將 Outputs 數量改為 12。

UI LED 節點：

拖入 12 個 ui_led 節點。
將 Function 節點的 12 個輸出點分別連線到這 12 個 LED。
設定 LED：顏色（ON: green, OFF: gray）。

Function 節點 (解析 DO)：

JavaScript
var do_list = msg.payload.do_all;
var messages = [];
for (var i = 0; i < 6; i++) {
    messages.push({ payload: do_list[i] });
}
return messages;

將 Outputs 數量改為 6。

Node-Red程式

2026年4月9日星期四

Python Modbus 控制 ADAM-6050 18-ch Isolated Digital I/O Module

Python Modbus 控制 ADAM-6050 18-ch Isolated Digital I/O Module

import tkinter as tk

from tkinter import messagebox

from pymodbus.client import ModbusTcpClient

import threading

import time

# --- 配置區 ---

ADAM_IP = '192.168.1.128'

ADAM_PORT = 502

DI_COUNT = 12

DO_COUNT = 6

DI_START_ADDR = 0

DO_START_ADDR = 16

class Adam6050App:

def __init__(self, root):

self.root = root

self.root.title("ADAM-6050 控制監控 (相容性修正版)")

self.root.geometry("450x480")

self.client = ModbusTcpClient(ADAM_IP, port=ADAM_PORT)

self.di_labels = []

self.do_vars = []

self.conn_status_var = tk.StringVar(value="連線檢查中...")

self.setup_ui()

self.running = True

self.thread = threading.Thread(target=self.update_loop, daemon=True)

self.thread.start()

def setup_ui(self):

# 狀態欄

status_frame = tk.Frame(self.root, bd=1, relief=tk.GROOVE, padx=10, pady=5)

status_frame.pack(side=tk.BOTTOM, fill=tk.X)

self.status_led = tk.Canvas(status_frame, width=15, height=15, highlightthickness=0)

self.status_led.pack(side=tk.LEFT, padx=5)

self.led_circle = self.status_led.create_oval(2, 2, 13, 13, fill="gray")

self.status_label = tk.Label(status_frame, textvariable=self.conn_status_var, font=("Arial", 9, "bold"))

self.status_label.pack(side=tk.LEFT)

# DI 顯示

di_frame = tk.LabelFrame(self.root, text="DI 狀態 (Digital Input)", padx=10, pady=10)

di_frame.pack(padx=10, pady=10, fill=tk.BOTH, expand=True)

for i in range(DI_COUNT):

lbl = tk.Label(di_frame, text=f"DI_{i:02d}", width=8, height=2,

bg="#333333", fg="white", font=("Arial", 9))

lbl.grid(row=i // 4, column=i % 4, padx=5, pady=5)

self.di_labels.append(lbl)

# DO 控制

do_frame = tk.LabelFrame(self.root, text="DO 控制 (Digital Output)", padx=10, pady=10)

do_frame.pack(padx=10, pady=10, fill=tk.BOTH, expand=True)

for i in range(DO_COUNT):

var = tk.BooleanVar()

# 注意：這裡使用 command 綁定時，先讀取目前畫面值再寫入

chk = tk.Checkbutton(do_frame, text=f"DO_{i:02d}", variable=var,

command=lambda idx=i: self.toggle_do(idx))

chk.grid(row=i // 3, column=i % 3, padx=15, pady=10)

self.do_vars.append(var)

def update_status_ui(self, is_connected):

if is_connected:

self.status_led.itemconfig(self.led_circle, fill="#00FF00")

self.conn_status_var.set(f"已連線: {ADAM_IP}")

self.status_label.config(fg="#006400")

else:

self.status_led.itemconfig(self.led_circle, fill="red")

self.conn_status_var.set(f"連線中斷: {ADAM_IP}")

self.status_label.config(fg="red")

for lbl in self.di_labels: lbl.config(bg="#333333")

def toggle_do(self, index):

if self.client.is_socket_open():

try:

val = self.do_vars[index].get()

# 關鍵：使用具名參數指定 address 和 value

self.client.write_coil(address=DO_START_ADDR + index, value=val)

except Exception as e:

print(f"寫入失敗: {e}")

else:

messagebox.showwarning("警告", "設備未連線")

def update_loop(self):

while self.running:

connected = self.client.connect()

self.update_status_ui(connected)

if connected:

try:

# 關鍵：強制使用具名參數 address=, count=

# 避免版本將第二個位置參數誤判為 slave/unit

di_res = self.client.read_discrete_inputs(address=DI_START_ADDR, count=DI_COUNT)

if di_res and not di_res.isError():

for i in range(DI_COUNT):

bg_color = "#4CAF50" if di_res.bits[i] else "#333333"

self.di_labels[i].config(bg=bg_color)

do_res = self.client.read_coils(address=DO_START_ADDR, count=DO_COUNT)

if do_res and not do_res.isError():

for i in range(DO_COUNT):

self.do_vars[i].set(do_res.bits[i])

except Exception as e:

print(f"輪詢通訊錯誤: {e}")

time.sleep(0.5)

def on_closing(self):

self.running = False

self.client.close()

self.root.destroy()

if __name__ == "__main__":

root = tk.Tk()

app = Adam6050App(root)

root.protocol("WM_DELETE_WINDOW", app.on_closing)

root.mainloop()

這段程式碼結合了 Tkinter（圖形介面）與 Pymodbus（工業通訊協定），實現了對 ADAM-6050 模組的即時監控與控制。

以下是逐行的詳細拆解說明：

1. 匯入庫與基礎配置

Python
import tkinter as tk                  # 建立 GUI 視窗的標準庫
from tkinter import messagebox        # 彈出警告對話框用
from pymodbus.client import ModbusTcpClient  # 實作 Modbus TCP 通訊的主工具
import threading                      # 讓通訊在背景執行，避免視窗卡死
import time                           # 控制更新頻率（延時）

# --- 配置區 ---
ADAM_IP = '192.168.1.128'             # ADAM 模組的 IP 地址
ADAM_PORT = 502                       # Modbus TCP 標準通訊埠
DI_COUNT = 12                         # ADAM-6050 有 12 個輸入點
DO_COUNT = 6                          # ADAM-6050 有 6 個輸出點
DI_START_ADDR = 0                     # DI 在 Modbus 中的起始地址
DO_START_ADDR = 16                    # DO 在 Modbus 中的起始地址 (0x11 hex = 17, 0-based 為 16)

2. 初始化類別與連線

Python
class Adam6050App:
    def __init__(self, root):
        self.root = root
        self.root.title("ADAM-6050 控制監控 (相容性修正版)")
        self.root.geometry("450x480")
        
        # 建立 Modbus TCP 客戶端物件，但此時尚未真正連線
        self.client = ModbusTcpClient(ADAM_IP, port=ADAM_PORT)
        
        self.di_labels = []           # 儲存 12 個 DI 標籤物件以便後續改顏色
        self.do_vars = []             # 儲存 6 個 Checkbutton 的布林狀態
        self.conn_status_var = tk.StringVar(value="連線檢查中...") # 狀態欄文字

        self.setup_ui()               # 呼叫 UI 配置函式
        
        # 啟動「多執行緒」：讓 update_loop 在背景跑，不影響 UI 操作
        self.running = True
        self.thread = threading.Thread(target=self.update_loop, daemon=True)
        self.thread.start()

3. UI 界面配置 (`setup_ui`)

這裡將視窗分為三個部分：

DI 顯示區：用標籤（Label）當作指示燈。
DO 控制區：用勾選框（Checkbutton）切換開關。
狀態欄：顯示目前是否有連上設備。

Python
    def setup_ui(self):
        # 狀態欄 (底部)
        status_frame = tk.Frame(self.root, bd=1, relief=tk.GROOVE, padx=10, pady=5)
        status_frame.pack(side=tk.BOTTOM, fill=tk.X)
        
        # 繪製一個小圓圈當作 LED 燈
        self.status_led = tk.Canvas(status_frame, width=15, height=15, highlightthickness=0)
        self.status_led.pack(side=tk.LEFT, padx=5)
        self.led_circle = self.status_led.create_oval(2, 2, 13, 13, fill="gray")
        
        self.status_label = tk.Label(status_frame, textvariable=self.conn_status_var, font=("Arial", 9, "bold"))
        self.status_label.pack(side=tk.LEFT)

        # DI 顯示區 (Grid 佈局 3x4)
        di_frame = tk.LabelFrame(self.root, text="DI 狀態 (Digital Input)", padx=10, pady=10)
        di_frame.pack(padx=10, pady=10, fill=tk.BOTH, expand=True)

        for i in range(DI_COUNT):
            lbl = tk.Label(di_frame, text=f"DI_{i:02d}", width=8, height=2, 
                           bg="#333333", fg="white", font=("Arial", 9))
            lbl.grid(row=i // 4, column=i % 4, padx=5, pady=5)
            self.di_labels.append(lbl)

        # DO 控制區 (Grid 佈局 2x3)
        do_frame = tk.LabelFrame(self.root, text="DO 控制 (Digital Output)", padx=10, pady=10)
        do_frame.pack(padx=10, pady=10, fill=tk.BOTH, expand=True)

        for i in range(DO_COUNT):
            var = tk.BooleanVar() # 綁定 Checkbutton 狀態
            # 當點擊勾選框時，觸發 toggle_do 並傳入編號
            chk = tk.Checkbutton(do_frame, text=f"DO_{i:02d}", variable=var, 
                                 command=lambda idx=i: self.toggle_do(idx))
            chk.grid(row=i // 3, column=i % 3, padx=15, pady=10)
            self.do_vars.append(var)

4. 核心邏輯：控制與更新

這部分負責與硬體溝通。

更新連線狀態 UI：

Python
    def update_status_ui(self, is_connected):
        if is_connected:
            self.status_led.itemconfig(self.led_circle, fill="#00FF00") # 變綠燈
            self.conn_status_var.set(f"已連線: {ADAM_IP}")
        else:
            self.status_led.itemconfig(self.led_circle, fill="red")     # 變紅燈
            self.conn_status_var.set(f"連線中斷: {ADAM_IP}")

控制 DO 輸出：

Python
    def toggle_do(self, index):
        if self.client.is_socket_open():
            try:
                val = self.do_vars[index].get() # 取得 UI 上的勾選狀態 (True/False)
                # write_coil: 寫入 Modbus 線圈 (Digital Output)
                self.client.write_coil(address=DO_START_ADDR + index, value=val)
            except Exception as e:
                print(f"寫入失敗: {e}")

背景更新迴圈 (update_loop)： 這是程式的大腦，每 0.5 秒跑一次。

Python
    def update_loop(self):
        while self.running:
            connected = self.client.connect() # 嘗試與 ADAM 連線
            self.update_status_ui(connected)

            if connected:
                try:
                    # 讀取 DI (Discrete Inputs): 取得 00001 開始的 12 個狀態
                    di_res = self.client.read_discrete_inputs(address=DI_START_ADDR, count=DI_COUNT)
                    if di_res and not di_res.isError():
                        for i in range(DI_COUNT):
                            # bits[i] 為 True 代表有輸入，燈變綠色 (#4CAF50)
                            bg_color = "#4CAF50" if di_res.bits[i] else "#333333"
                            self.di_labels[i].config(bg=bg_color)

                    # 讀取 DO (Coils): 讀回目前 ADAM 的輸出狀態同步到 UI
                    do_res = self.client.read_coils(address=DO_START_ADDR, count=DO_COUNT)
                    if do_res and not do_res.isError():
                        for i in range(DO_COUNT):
                            self.do_vars[i].set(do_res.bits[i]) # 同步勾選框
                            
                except Exception as e:
                    print(f"輪詢通訊錯誤: {e}")
            
            time.sleep(0.5) # 休息半秒，避免 CPU 負載過高

5. 安全結束

Python
    def on_closing(self):
        self.running = False    # 停止 while 迴圈
        self.client.close()     # 關閉 Modbus TCP 連線
        self.root.destroy()     # 關閉視窗

總結程式運作流程：

啟動：開啟 UI 視窗並開啟一個後台小幫手（執行緒）。
連線：後台小幫手不斷嘗試連線到 192.168.1.128。
監控：連線成功後，每 0.5 秒問 ADAM：「DI 現在狀況如何？」、「DO 現在開還是關？」，然後更新畫面的顏色和勾選框。
控制：當你手動點擊 DO 勾選框，程式立刻發送指令告訴 ADAM：「把第 N 個輸出點打開/關閉」。

訂閱：文章 (Atom)

2026年4月25日 星期六

Adam DO*5 DI*12 + Python + Node-Red

Node-RED 邏輯流程設計

2026年4月9日 星期四

Python Modbus 控制 ADAM-6050 18-ch Isolated Digital I/O Module

1. 匯入庫與基礎配置

2. 初始化類別與連線

3. UI 界面配置 (setup_ui)

4. 核心邏輯：控制與更新

5. 安全結束

總結程式運作流程：

2026-09 mqtt 實驗3

2026年4月25日星期六

Adam DO5 DI12 + Python + Node-Red

2026年4月9日星期四

3. UI 界面配置 (`setup_ui`)