使用Python Tkinter 人機介面控制 透過MQTT控制ESP32 內建LED ON. OFF , FLASH ,TIMER(20S)
工具: Python Thonny , ESP32 VS Code
Python Tkinter Thonny Code
import tkinter as tk
from tkinter import ttk
import paho.mqtt.client as mqtt
import time
# --- MQTT 設定 ---
MQTT_BROKER = "broker.mqtt-dashboard.com"
MQTT_PORT = 1883
TOPIC_LED_CONTROL = "alex9ufo/rfid/led"
TOPIC_LED_STATUS = "alex9ufo/rfid/ledStatus"
# TOPIC_RFID_UID = "alex9ufo/rfid/UID" # 本次範例未使用，但保留
class LedControllerApp:
def __init__(self, master):
self.master = master
master.title("MQTT LED 控制器")
self.led_status_var = tk.StringVar(value="狀態: 離線")
self.timer_label_var = tk.StringVar(value="計時器: 閒置")
self.timer_end_time = 0
self.is_flashing = False
# --- 設定 MQTT 客戶端 ---
self.client = mqtt.Client()
self.client.on_connect = self.on_connect
self.client.on_message = self.on_message
self.connect_mqtt()
# --- 建立 GUI ---
self.create_widgets()
# 啟動計時器更新迴圈
self.master.after(100, self.update_timer)
def connect_mqtt(self):
try:
print(f"嘗試連線到 MQTT Broker: {MQTT_BROKER}")
self.client.connect(MQTT_BROKER, MQTT_PORT, 60)
self.client.loop_start() # 啟動背景執行緒處理網路流量
except Exception as e:
print(f"MQTT 連線失敗: {e}")
self.led_status_var.set("狀態: 連線失敗")
def on_connect(self, client, userdata, flags, rc):
if rc == 0:
print("MQTT 連線成功!")
client.subscribe(TOPIC_LED_STATUS)
self.led_status_var.set("狀態: 已連線, LED未知")
else:
print(f"連線失敗，返回碼 {rc}")
self.led_status_var.set(f"狀態: 連線失敗 {rc}")
def on_message(self, client, userdata, msg):
topic = msg.topic
payload = msg.payload.decode()
print(f"收到訊息 - Topic: {topic}, Payload: {payload}")
if topic == TOPIC_LED_STATUS:
self.led_status_var.set(f"狀態: LED {payload.upper()}")
def send_command(self, command):
"""發送 LED 控制指令到 MQTT"""
if self.client.is_connected():
self.client.publish(TOPIC_LED_CONTROL, command, qos=1)
print(f"發送指令: {command}")
# 特殊處理 Timer 指令
if command == "timer_20":
self.timer_end_time = time.time() + 20
self.timer_label_var.set("計時器: 20秒倒數中...")
else:
self.timer_end_time = 0 # 取消計時器顯示
self.timer_label_var.set("計時器: 閒置")
else:
print("錯誤：MQTT 未連線，無法發送指令。")
def update_timer(self):
"""更新計時器顯示"""
if self.timer_end_time > 0:
remaining = int(self.timer_end_time - time.time())
if remaining > 0:
self.timer_label_var.set(f"計時器: 剩餘 {remaining} 秒")
self.master.after(1000, self.update_timer)
else:
self.timer_end_time = 0
self.timer_label_var.set("計時器: 結束 (LED OFF)")
# 不需要發送 OFF，因為 Arduino 會自行關閉
else:
self.master.after(1000, self.update_timer)
def create_widgets(self):
# 狀態標籤
ttk.Label(self.master, textvariable=self.led_status_var, font=('Arial', 12)).pack(pady=10)
ttk.Label(self.master, textvariable=self.timer_label_var, font=('Arial', 10)).pack(pady=5)
# 控制按鈕
button_frame = ttk.Frame(self.master)
button_frame.pack(pady=20, padx=10)
# ON
ttk.Button(button_frame, text="LED ON", command=lambda: self.send_command("on")).pack(side=tk.LEFT, padx=5, pady=5)
# OFF
ttk.Button(button_frame, text="LED OFF", command=lambda: self.send_command("off")).pack(side=tk.LEFT, padx=5, pady=5)
# FLASH
ttk.Button(button_frame, text="LED FLASH", command=lambda: self.send_command("flash")).pack(side=tk.LEFT, padx=5, pady=5)
# TIMER 20 SEC
ttk.Button(button_frame, text="TIMER (20s)", command=lambda: self.send_command("timer_20")).pack(side=tk.LEFT, padx=5, pady=5)
def on_closing(self):
"""關閉應用程式時清理資源"""
print("關閉應用程式...")
self.client.loop_stop()
self.client.disconnect()
self.master.destroy()
if __name__ == "__main__":
# 確保您已安裝 paho-mqtt: pip install paho-mqtt
root = tk.Tk()
app = LedControllerApp(root)
root.protocol("WM_DELETE_WINDOW", app.on_closing)
root.mainloop()
ESP32 VS-Code
#include <Arduino.h>
// 根據您的板卡調整 LED 腳位
// ESP32 通常是 GPIO 2 (板載 LED)
// ESP8266 NodeMCU 通常是 D4 (GPIO 2, 板載 LED)
#ifdef ESP32
#include <WiFi.h>
const int LED_PIN = 2;
#else
#include <ESP8266WiFi.h>
const int LED_PIN = D4; // D4 是 GPIO2
#endif
#include <PubSubClient.h>
// --- Wi-Fi 設定 ---
const char* ssid = "alex9ufo";
const char* password = "alex9981";
// --- MQTT 設定 ---
const char* MQTT_BROKER = "broker.mqtt-dashboard.com";
const int MQTT_PORT = 1883;
const char* TOPIC_LED_CONTROL = "alex9ufo/rfid/led";
const char* TOPIC_LED_STATUS = "alex9ufo/rfid/ledStatus";
// const char* TOPIC_RFID_UID = "alex9ufo/rfid/UID"; // 本次範例未使用
WiFiClient espClient;
PubSubClient client(espClient);
// LED 狀態變數
enum LedMode { OFF, ON, FLASH, TIMER_ACTIVE };
LedMode currentMode = OFF;
unsigned long previousMillis = 0;
const long flashInterval = 500; // 閃爍間隔 500ms
unsigned long timerStartTime = 0;
const unsigned long timerDuration = 20000; // 20 秒 (20000 毫秒)
// --- 函式宣告 ---
void setup_wifi();
void callback(char* topic, byte* payload, unsigned int length);
void reconnect();
void publishLedStatus(const char* status);
void handleTimer();
void setup() {
Serial.begin(115200);
pinMode(LED_PIN, OUTPUT);
digitalWrite(LED_PIN, LOW); // 預設關閉
setup_wifi();
client.setServer(MQTT_BROKER, MQTT_PORT);
client.setCallback(callback);
}
void loop() {
if (!client.connected()) {
reconnect();
}
client.loop();
// 處理閃爍邏輯 (FLASH)
if (currentMode == FLASH) {
unsigned long currentMillis = millis();
if (currentMillis - previousMillis >= flashInterval) {
previousMillis = currentMillis;
int ledState = digitalRead(LED_PIN);
digitalWrite(LED_PIN, !ledState);
}
}
// 處理計時器邏輯 (TIMER)
handleTimer();
}
/**
* @brief 連接 Wi-Fi 網路
*/
void setup_wifi() {
delay(10);
Serial.println();
Serial.print("連接到 ");
Serial.println(ssid);
WiFi.begin(ssid, password);
while (WiFi.status() != WL_CONNECTED) {
delay(500);
Serial.print(".");
}
Serial.println("");
Serial.println("WiFi 連線成功");
Serial.print("IP 位址: ");
Serial.println(WiFi.localIP());
}
/**
* @brief MQTT 訊息回呼函式
*/
void callback(char* topic, byte* payload, unsigned int length) {
Serial.print("收到訊息 [");
Serial.print(topic);
Serial.print("]: ");
String message;
for (int i = 0; i < length; i++) {
message += (char)payload[i];
}
Serial.println(message);
if (strcmp(topic, TOPIC_LED_CONTROL) == 0) {
if (message == "on") {
currentMode = ON;
digitalWrite(LED_PIN, HIGH);
publishLedStatus("ON");
} else if (message == "off") {
currentMode = OFF;
digitalWrite(LED_PIN, LOW);
publishLedStatus("OFF");
} else if (message == "flash") {
currentMode = FLASH;
publishLedStatus("FLASH");
} else if (message == "timer_20") {
currentMode = TIMER_ACTIVE;
timerStartTime = millis(); // 啟動計時器
digitalWrite(LED_PIN, HIGH); // Timer 開始時 LED ON
publishLedStatus("TIMER_20S");
}
}
}
/**
* @brief 重新連線到 MQTT Broker
*/
void reconnect() {
while (!client.connected()) {
Serial.print("嘗試 MQTT 連線...");
// 嘗試使用 MAC 位址作為客戶端 ID
String clientId = "ESP_Client_";
clientId += String(random(0xffff), HEX);
if (client.connect(clientId.c_str())) {
Serial.println("成功!");
// 訂閱控制主題
client.subscribe(TOPIC_LED_CONTROL);
// 發布初始狀態
publishLedStatus("READY");
} else {
Serial.print("失敗, rc=");
Serial.print(client.state());
Serial.println(" 10 秒後重試");
delay(10000);
}
}
}
/**
* @brief 發布 LED 當前狀態
*/
void publishLedStatus(const char* status) {
if (client.connected()) {
client.publish(TOPIC_LED_STATUS, status);
Serial.print("發布狀態: ");
Serial.println(status);
}
}
/**
* @brief 處理計時器結束邏輯
*/
void handleTimer() {
if (currentMode == TIMER_ACTIVE) {
if (millis() - timerStartTime >= timerDuration) {
// 計時器結束
currentMode = OFF;
digitalWrite(LED_PIN, LOW); // 關閉 LED
publishLedStatus("OFF_TIMER_DONE");
Serial.println("20 秒計時結束，LED 關閉。");
}
}
}
沒有留言:
張貼留言