alex9ufo 聰明人求知心切: Adam DO*5 DI*12 + 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程式

alex9ufo 聰明人求知心切

2026年4月25日星期六

Adam DO5 DI12 + Python + Node-Red

Node-RED 邏輯流程設計

沒有留言:

張貼留言

Adam DO5 DI12 + Python + Node-Red

2026年4月25日 星期六

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

Node-RED 邏輯流程設計

沒有留言:

張貼留言

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

2026年4月25日星期六

Adam DO5 DI12 + Python + Node-Red

Adam DO5 DI12 + Python + Node-Red