2025-09 上學期<作業1> MQTT + WOKWI ESP32 (DHT22,LED) + MQTTX + MQTTGO.io + Python TKInter
作業1,2 每位同學都要繳交
作業3,4 以組為單位1至3人
作業執行結果上傳到YT上 影片超連結 mail 到 alex9ufo@gmail.com
(分數 依照繳交的先後次序)
執行後結果 https://www.youtube.com/watch?v=FY32EG7rwMc
# MQTT 設定
BROKER_HOST = "broker.mqttgo.io" PORT = 1883
LED_TOPIC = "wokwi/led/control"
TEMP_TOPIC = "wokwi/dht/temperature"
HUMIDITY_TOPIC = "wokwi/dht/humidity"
WOKWI ESP32
#include <WiFi.h>
#include <PubSubClient.h>
#include <DHT.h>
#include <DHT_U.h> // 需要同時包含 DHT_U.h
// --- Wi-Fi 設定 ---
const char* ssid = "Wokwi-GUEST";
const char* password = "";
// --- MQTT 設定 ---
const char* mqtt_server = "broker.mqttgo.io"; // 或 "mqtt.eclipseprojects.io"
const int mqtt_port = 1883;
const char* mqtt_client_id = "ESP32_Wokwi_Client";
// MQTT 主題
const char* mqtt_topic_led_control = "wokwi/led/control";
const char* mqtt_topic_temperature = "wokwi/dht/temperature";
const char* mqtt_topic_humidity = "wokwi/dht/humidity";
WiFiClient espClient;
PubSubClient client(espClient);
// --- LED 設定 ---
const int ledPin = 2; // 連接到 GPIO 2
enum LedMode { ON, OFF, FLASH, TIMER };
volatile LedMode currentLedMode = OFF;
volatile unsigned long timerStartTime = 0;
volatile bool ledState = false; // 用於閃爍模式
// --- DHT22 設定 ---
#define DHTPIN 4 // 連接到 GPIO 4
#define DHTTYPE DHT22 // DHT 22 (AM2302), AM2321
DHT dht(DHTPIN, DHTTYPE);
// --- 任務句柄 ---
TaskHandle_t TaskLEDControl = NULL;
TaskHandle_t TaskDHTSensor = NULL;
// --- 函式宣告 ---
void setup_wifi();
void reconnect_mqtt();
void callback(char* topic, byte* payload, unsigned int length);
void ledControlTask(void *pvParameters);
void dhtSensorTask(void *pvParameters);
void setup() {
Serial.begin(115200);
pinMode(ledPin, OUTPUT);
digitalWrite(ledPin, LOW); // 確保初始是關閉的
setup_wifi();
client.setServer(mqtt_server, mqtt_port);
client.setCallback(callback);
dht.begin(); // 初始化 DHT 感測器
// 創建 LED 控制任務，運行在 Core 0
xTaskCreatePinnedToCore(
ledControlTask, /* 任務函式 */
"LED Control", /* 任務名稱 */
2048, /* 堆疊大小 (字節) */
NULL, /* 任務參數 */
1, /* 任務優先級 */
&TaskLEDControl, /* 任務句柄 */
0 /* 運行在 Core 0 */
);
// 創建 DHT 感測器任務，運行在 Core 1
xTaskCreatePinnedToCore(
dhtSensorTask, /* 任務函式 */
"DHT Sensor", /* 任務名稱 */
4096, /* 堆疊大小 (字節) */
NULL, /* 任務參數 */
1, /* 任務優先級 */
&TaskDHTSensor, /* 任務句柄 */
1 /* 運行在 Core 1 */
);
}
void loop() {
// 主循環中只負責維持 MQTT 連線
if (!client.connected()) {
reconnect_mqtt();
}
client.loop(); // 處理 MQTT 訊息
delay(10); // 短暫延遲，避免佔用太多 CPU
}
// --- Wi-Fi 連線函式 ---
void setup_wifi() {
delay(10);
Serial.println();
Serial.print("Connecting to ");
Serial.println(ssid);
WiFi.begin(ssid, password);
while (WiFi.status() != WL_CONNECTED) {
delay(500);
Serial.print(".");
}
Serial.println("");
Serial.println("WiFi connected");
Serial.print("IP address: ");
Serial.println(WiFi.localIP());
}
// --- MQTT 重連函式 ---
void reconnect_mqtt() {
while (!client.connected()) {
Serial.print("Attempting MQTT connection...");
// 嘗試連線
if (client.connect(mqtt_client_id)) {
Serial.println("connected");
// 訂閱 LED 控制主題
client.subscribe(mqtt_topic_led_control);
Serial.print("Subscribed to: ");
Serial.println(mqtt_topic_led_control);
} else {
Serial.print("failed, rc=");
Serial.print(client.state());
Serial.println(" try again in 5 seconds");
// 等待 5 秒後重試
delay(5000);
}
}
}
// --- MQTT 訊息回調函式 ---
void callback(char* topic, byte* payload, unsigned int length) {
Serial.print("Message arrived [");
Serial.print(topic);
Serial.print("] ");
String message = "";
for (int i = 0; i < length; i++) {
message += (char)payload[i];
}
Serial.println(message);
if (String(topic) == mqtt_topic_led_control) {
if (message == "on") {
currentLedMode = ON;
digitalWrite(ledPin, HIGH);
Serial.println("LED Mode: ON");
} else if (message == "off") {
currentLedMode = OFF;
digitalWrite(ledPin, LOW);
Serial.println("LED Mode: OFF");
} else if (message == "flash") {
currentLedMode = FLASH;
Serial.println("LED Mode: FLASH");
} else if (message == "timer") {
currentLedMode = TIMER;
digitalWrite(ledPin, HIGH); // 定時模式開始時先開啟 LED
timerStartTime = millis();
Serial.println("LED Mode: TIMER (10s)");
} else {
Serial.println("Unknown LED command.");
}
}
}
// --- LED 控制任務 (運行在 Core 0) ---
void ledControlTask(void *pvParameters) {
(void) pvParameters; // 避免編譯器警告
for (;;) { // 無限循環
switch (currentLedMode) {
case ON:
// LED 保持亮著，由 callback 函式設置
break;
case OFF:
// LED 保持熄滅，由 callback 函式設置
break;
case FLASH:
digitalWrite(ledPin, ledState);
ledState = !ledState;
vTaskDelay(pdMS_TO_TICKS(500)); // 每 500ms 改變一次狀態
break;
case TIMER:
if (millis() - timerStartTime >= 10000) { // 10 秒
digitalWrite(ledPin, LOW);
currentLedMode = OFF; // 定時結束後轉為 OFF 模式
Serial.println("LED Timer finished. LED OFF.");
}
vTaskDelay(pdMS_TO_TICKS(10)); // 短暫延遲
break;
default:
digitalWrite(ledPin, LOW); // 預設為關閉
break;
}
vTaskDelay(pdMS_TO_TICKS(10)); // 短暫延遲，讓其他任務有機會執行
}
}
// --- DHT 感測器任務 (運行在 Core 1) ---
void dhtSensorTask(void *pvParameters) {
(void) pvParameters; // 避免編譯器警告
for (;;) { // 無限循環
delay(2000); // 每 2 秒讀取一次數據，避免頻繁讀取導致錯誤
float h = dht.readHumidity();
float t = dht.readTemperature(); // 讀取攝氏溫度
// 檢查是否讀取失敗，如果是則嘗試重讀
if (isnan(h) || isnan(t)) {
Serial.println(F("Failed to read from DHT sensor!"));
} else {
Serial.print(F("Humidity: "));
Serial.print(h);
Serial.print(F("% Temperature: "));
Serial.print(t);
Serial.println(F("°C"));
// 發布溫度
char tempString[8];
dtostrf(t, 4, 2, tempString); // 浮點數轉字串
client.publish(mqtt_topic_temperature, tempString);
// 發布濕度
char humString[8];
dtostrf(h, 4, 2, humString); // 浮點數轉字串
delay(250);
client.publish(mqtt_topic_humidity, humString);
}
vTaskDelay(pdMS_TO_TICKS(5000)); // 每 5 秒執行一次，避免 MQTT 發布過於頻繁
}
}
https://broker.mqttgo.io/的設定 (Broker + 推播 Publish + 訂閱 Subscriptions ) 要先連線再設定
MQTTX設定 (載點 https://mqttx.app/downloads )
PYTHON + TKInter
下載 https://thonny.org/ <<python>>
安裝套件
使用 Python Tkinter 庫的 MQTT 應用程式。這個程式將具備以下功能：
連接到 MQTT Broker (broker.mqttgo.io)。
提供使用者介面來發送指令，控制 LED 燈（開啟、關閉、閃爍、定時）。
訂閱並顯示來自兩個不同主題的溫度和濕度資料。
所有這些功能都將整合在一個單一的 Tkinter 視窗中。
==========================================================
import tkinter as tk
from tkinter import ttk
import paho.mqtt.client as mqtt
import threading
import json
# MQTT 設定
BROKER_HOST = "broker.mqttgo.io"
PORT = 1883
LED_TOPIC = "wokwi/led/control"
TEMP_TOPIC = "wokwi/dht/temperature"
HUMIDITY_TOPIC = "wokwi/dht/humidity"
class MqttApp(tk.Tk):
def __init__(self):
super().__init__()
self.title("MQTT LED and Sensor Monitor")
self.geometry("400x300")
self.resizable(False, False)
self.mqtt_client = mqtt.Client(protocol=mqtt.MQTTv311)
self.mqtt_client.on_connect = self.on_connect
self.mqtt_client.on_message = self.on_message
self.setup_ui()
self.connect_mqtt()
def setup_ui(self):
# 建立框架
main_frame = ttk.Frame(self, padding="10")
main_frame.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S))
self.columnconfigure(0, weight=1)
self.rowconfigure(0, weight=1)
# LED 控制區
led_frame = ttk.LabelFrame(main_frame, text="LED Control", padding="10")
led_frame.grid(row=0, column=0, padx=5, pady=5, sticky="ew")
ttk.Button(led_frame, text="ON", command=lambda: self.publish_led_command("on")).pack(fill="x", pady=2)
ttk.Button(led_frame, text="OFF", command=lambda: self.publish_led_command("off")).pack(fill="x", pady=2)
ttk.Button(led_frame, text="FLASH", command=lambda: self.publish_led_command("flash")).pack(fill="x", pady=2)
ttk.Button(led_frame, text="TIMER", command=lambda: self.publish_led_command("timer")).pack(fill="x", pady=2)
# 感應器數據顯示區
sensor_frame = ttk.LabelFrame(main_frame, text="Sensor Data", padding="10")
sensor_frame.grid(row=1, column=0, padx=5, pady=10, sticky="ew")
ttk.Label(sensor_frame, text="Temperature:").grid(row=0, column=0, sticky="w", pady=2)
self.temp_label = ttk.Label(sensor_frame, text="-- °C")
self.temp_label.grid(row=0, column=1, sticky="w", padx=5)
ttk.Label(sensor_frame, text="Humidity:").grid(row=1, column=0, sticky="w", pady=2)
self.humidity_label = ttk.Label(sensor_frame, text="-- %")
self.humidity_label.grid(row=1, column=1, sticky="w", padx=5)
# 狀態顯示區
self.status_label = ttk.Label(main_frame, text="Status: Disconnected", foreground="red")
self.status_label.grid(row=2, column=0, pady=10)
def connect_mqtt(self):
try:
self.mqtt_client.connect(BROKER_HOST, PORT)
# 啟動一個新線程來處理 MQTT 網路迴圈
mqtt_thread = threading.Thread(target=self.mqtt_loop, daemon=True)
mqtt_thread.start()
except Exception as e:
self.status_label.config(text=f"Status: Connection failed. {e}", foreground="red")
def mqtt_loop(self):
self.mqtt_client.loop_forever()
def on_connect(self, client, userdata, flags, rc):
if rc == 0:
print("Connected to MQTT Broker!")
self.status_label.config(text="Status: Connected", foreground="green")
# 連接成功後訂閱主題
self.mqtt_client.subscribe(TEMP_TOPIC)
self.mqtt_client.subscribe(HUMIDITY_TOPIC)
else:
print(f"Failed to connect, return code {rc}\n")
self.status_label.config(text="Status: Connection failed", foreground="red")
def on_message(self, client, userdata, msg):
topic = msg.topic
payload = msg.payload.decode()
print(f"Received message on topic: {topic} with payload: {payload}")
if topic == TEMP_TOPIC:
self.update_temperature(payload)
elif topic == HUMIDITY_TOPIC:
self.update_humidity(payload)
def update_temperature(self, temp_str):
try:
temp_value = float(temp_str)
self.temp_label.config(text=f"{temp_value:.1f} °C")
except ValueError:
self.temp_label.config(text="Invalid data")
def update_humidity(self, humidity_str):
try:
humidity_value = float(humidity_str)
self.humidity_label.config(text=f"{humidity_value:.1f} %")
except ValueError:
self.humidity_label.config(text="Invalid data")
def publish_led_command(self, command):
try:
self.mqtt_client.publish(LED_TOPIC, command)
print(f"Published command: '{command}' to topic '{LED_TOPIC}'")
except Exception as e:
print(f"Failed to publish message: {e}")
if __name__ == "__main__":
app = MqttApp()
app.mainloop()
==========================================================
程式碼說明
MqttApp 類別：這個類別繼承自 tk.Tk，是整個應用程式的主視窗。
__init__ 方法：初始化視窗、設定標題和大小，並創建 MQTT 客戶端實例。接著，它呼叫 setup_ui 建立使用者介面，並呼叫 connect_mqtt 連接到 MQTT Broker。
setup_ui 方法：這個方法負責建立所有 Tkinter 元件，包括按鈕和標籤，並使用 ttk 模組來獲得更現代的外觀。
connect_mqtt 方法：嘗試連接到指定的 MQTT Broker。為了確保 GUI 不會被 MQTT 的網路迴圈阻塞，它使用了一個新的**線程（threading）**來執行 mqtt_client.loop_forever()。
on_connect 方法：這是 MQTT 連接成功後的回呼函數。它會將狀態標籤更新為 "Connected"，並立即訂閱溫度和濕度主題。
on_message 方法：這是當收到訊息時觸發的回呼函數。它檢查訊息的主題，並根據不同的主題將數據傳遞給 update_temperature 或 update_humidity 方法來更新介面上的標籤。
publish_led_command 方法：當您點擊 LED 控制按鈕時，這個方法會將對應的命令（on, off, flash, timer）發佈到 wokwi/led/control 主題。
update_temperature 和 update_humidity 方法：這些方法負責解析收到的字串數據，將其轉換為浮點數，並更新介面上的文字顯示。
