ESP32 雙核心控制 LED 與 DHT22 溫濕度感測器 (Wokwi 模擬) EX7 -- Tkinter + Telegram Bot API
這個程式旨在提供一個全面的物聯網控制與監控解決方案：
本地 GUI 控制: 透過 Tkinter 視窗上的按鈕直接控制 LED。
遠端 Telegram 控制: 透過 Telegram App 發送指令控制 LED 並查詢溫濕度。
數據顯示: 在 Tkinter 視窗上即時顯示來自 ESP32 的溫濕度數據。
雙向通訊: MQTT 確保 Python 應用程式和 ESP32 之間可靠的訊息交換。
非同步/多執行緒: 程式使用多執行緒來處理 Tkinter GUI、MQTT 通訊和 Telegram Bot 輪詢，確保應用程式的響應性。
程式碼總覽
import tkinter as tk
from tkinter import ttk, messagebox
import paho.mqtt.client as mqtt
import threading
import time
import os
import asyncio # 用於處理 Telegram Bot 的異步操作
import sys # 用於程式退出
# Telegram Bot Library
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes
# --- 配置區 ---
# MQTT Broker 設定
MQTT_BROKER = "broker.mqttgo.io"
MQTT_PORT = 1883
# MQTT 主題定義 (確保與 Wokwi ESP32 程式碼中的主題一致)
MQTT_LED_CONTROL_TOPIC = "wokwi/esp32/led/control" # 發布 LED 控制命令
MQTT_TEMP_TOPIC = "wokwi/esp32/dht/temperature" # 訂閱溫度數據
MQTT_HUMID_TOPIC = "wokwi/esp32/dht/humidity" # 訂閱濕度數據
# Telegram Bot Token (從 @BotFather 取得)
# 建議從環境變數讀取以保護你的 Token:
# 例如，在 Linux/macOS 終端機: export TELEGRAM_BOT_TOKEN='你的實際Token'
# 或在 Windows 命令提示字元: set TELEGRAM_BOT_TOKEN='你的實際Token'
TELEGRAM_BOT_TOKEN = os.getenv('TELEGRAM_BOT_TOKEN', '773289340254:AA1HbrWu9o1vb1BKPQyWsbNSjNxfCGCrEWU-o')
TELEGRAM_BOT_TOKEN = os.getenv('TELEGRAM_BOT_TOKEN', 'YOUR_TELEGRAM_BOT_TOKEN')
# --- 全局變數用於儲存最新的感測器數據 ---
latest_temperature = "N/A"
latest_humidity = "N/A"
telegram_application = None # 用於儲存 Telegram Bot 應用實例，以便在關閉時正確停止
# --- Tkinter 主應用程式類 ---
class IoTApp:
def __init__(self, master):
self.master = master
master.title("Wokwi ESP32 IoT Control & Monitor (Tkinter with Telegram)")
master.geometry("600x500") # 設定視窗大小
self.mqtt_client = None
self.telegram_bot_thread = None
self.create_widgets() # 建立 GUI 介面元素
self.setup_mqtt() # 設定 MQTT 客戶端
# 在單獨的執行緒中啟動 Telegram Bot，避免阻塞 Tkinter GUI
# daemon=True 確保主程式退出時，這個執行緒也會跟著結束
self.telegram_bot_thread = threading.Thread(target=self.start_telegram_bot_loop, daemon=True)
self.telegram_bot_thread.start()
def create_widgets(self):
"""建立 Tkinter GUI 中的所有元件。"""
# --- LED 控制區塊 ---
led_frame = ttk.LabelFrame(self.master, text="LED Control (Local)", padding="10")
led_frame.pack(pady=10, padx=10, fill="x")
ttk.Button(led_frame, text="LED ON", command=lambda: self.publish_mqtt(MQTT_LED_CONTROL_TOPIC, "on")).pack(side="left", padx=5, pady=5)
ttk.Button(led_frame, text="LED OFF", command=lambda: self.publish_mqtt(MQTT_LED_CONTROL_TOPIC, "off")).pack(side="left", padx=5, pady=5)
ttk.Button(led_frame, text="LED FLASH", command=lambda: self.publish_mqtt(MQTT_LED_CONTROL_TOPIC, "flash")).pack(side="left", padx=5, pady=5)
ttk.Button(led_frame, text="LED TIMER (10s)", command=lambda: self.publish_mqtt(MQTT_LED_CONTROL_TOPIC, "timer")).pack(side="left", padx=5, pady=5)
# --- 溫濕度顯示區塊 ---
dht_frame = ttk.LabelFrame(self.master, text="DHT22 Sensor Data", padding="10")
dht_frame.pack(pady=10, padx=10, fill="x")
self.temp_label = ttk.Label(dht_frame, text=f"Temperature: {latest_temperature}°C", font=("Arial", 14))
self.temp_label.pack(pady=5, anchor="w")
self.humid_label = ttk.Label(dht_frame, text=f"Humidity: {latest_humidity}%", font=("Arial", 14))
self.humid_label.pack(pady=5, anchor="w")
# --- 狀態顯示區塊 ---
status_frame = ttk.LabelFrame(self.master, text="Status", padding="10")
status_frame.pack(pady=10, padx=10, fill="both", expand=True)
self.mqtt_status_label = ttk.Label(status_frame, text="MQTT: Connecting...", font=("Arial", 10), foreground="blue")
self.mqtt_status_label.pack(pady=2, anchor="w")
self.telegram_status_label = ttk.Label(status_frame, text="Telegram Bot: Starting...", font=("Arial", 10), foreground="blue")
self.telegram_status_label.pack(pady=2, anchor="w")
# --- MQTT 相關函式 ---
def setup_mqtt(self):
"""設定 MQTT 客戶端的回呼函數並嘗試連接。"""
self.mqtt_client = mqtt.Client()
self.mqtt_client.on_connect = self.on_mqtt_connect
self.mqtt_client.on_message = self.on_mqtt_message
self.mqtt_client.on_disconnect = self.on_mqtt_disconnect
try:
self.mqtt_client.connect(MQTT_BROKER, MQTT_PORT, 60)
# 在獨立執行緒中運行 MQTT 循環，避免阻塞 GUI
self.mqtt_thread = threading.Thread(target=self.mqtt_client.loop_forever, daemon=True)
self.mqtt_thread.start()
except Exception as e:
self.update_mqtt_status(f"Error: {e}", "red")
print(f"MQTT Connection Error: {e}")
def on_mqtt_connect(self, client, userdata, flags, rc):
"""MQTT 連接成功時的回呼函數。"""
if rc == 0:
self.update_mqtt_status("Connected to MQTT Broker!", "green")
client.subscribe(MQTT_TEMP_TOPIC)
client.subscribe(MQTT_HUMID_TOPIC)
print(f"Subscribed to {MQTT_TEMP_TOPIC} and {MQTT_HUMID_TOPIC}")
else:
self.update_mqtt_status(f"Failed to connect, return code {rc}. Retrying...", "orange")
print(f"MQTT connection failed, return code {rc}")
def on_mqtt_disconnect(self, client, userdata, rc):
"""MQTT 斷開連接時的回呼函數。"""
self.update_mqtt_status(f"Disconnected from MQTT. Reconnecting...", "red")
print(f"Disconnected from MQTT with result code {rc}")
def on_mqtt_message(self, client, userdata, msg):
"""MQTT 接收到訊息時的回呼函數，更新感測器數據。"""
global latest_temperature, latest_humidity
payload = msg.payload.decode()
print(f"Received MQTT message: Topic='{msg.topic}', Payload='{payload}'")
if msg.topic == MQTT_TEMP_TOPIC:
latest_temperature = payload
elif msg.topic == MQTT_HUMID_TOPIC:
latest_humidity = payload
# 在 Tkinter 主執行緒更新 GUI 顯示，確保線程安全
self.master.after(0, self.update_dht_labels)
def update_dht_labels(self):
"""更新 Tkinter GUI 上的溫濕度顯示。"""
self.temp_label.config(text=f"Temperature: {latest_temperature}°C")
self.humid_label.config(text=f"Humidity: {latest_humidity}%")
def update_mqtt_status(self, text, color):
"""更新 Tkinter GUI 上的 MQTT 狀態顯示。"""
# 使用 self.master.after 將 GUI 更新操作排程到主執行緒
self.master.after(0, lambda: self.mqtt_status_label.config(text=f"MQTT: {text}", foreground=color))
def update_telegram_status(self, text, color):
"""更新 Tkinter GUI 上的 Telegram Bot 狀態顯示。"""
# 使用 self.master.after 將 GUI 更新操作排程到主執行緒
self.master.after(0, lambda: self.telegram_status_label.config(text=f"Telegram Bot: {text}", foreground=color))
def publish_mqtt(self, topic, payload):
"""發布 MQTT 訊息到指定的 topic。"""
if self.mqtt_client and self.mqtt_client.is_connected():
try:
self.mqtt_client.publish(topic, payload)
print(f"Published to {topic}: {payload}")
self.update_mqtt_status(f"Sent command: {payload}", "blue")
except Exception as e:
self.update_mqtt_status(f"Publish Error: {e}", "red")
print(f"MQTT Publish Error: {e}")
else:
self.update_mqtt_status("MQTT Not Connected!", "red")
print("MQTT Client not connected, cannot publish.")
# --- Telegram Bot 相關函式 (在獨立執行緒中運行) ---
async def start_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""處理 /start 指令，發送歡迎訊息和指令列表。"""
await update.message.reply_text("哈囉！我是您的 IoT Bot。\n您可以透過以下指令控制設備或查詢狀態：\n"
"/on - 開啟 LED\n"
"/off - 關閉 LED\n"
"/flash - LED 閃爍\n"
"/timer - LED 定時 10 秒開關\n"
"輸入 `溫度` - 查詢目前溫度\n"
"輸入 `濕度` - 查詢目前濕度\n"
"您也可以點擊 Tkinter 視窗上的按鈕控制 LED。")
self.update_telegram_status("Bot Ready (User Started)", "green")
async def handle_text_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""處理非命令的普通文字訊息 (例如: '溫度', '濕度')。"""
text = update.message.text.lower().strip()
print(f"Received Telegram text message: '{text}' from {update.effective_user.id}")
reply_message = "未知指令，請輸入 `/on`, `/off`, `/flash`, `/timer`, `溫度` 或 `濕度` (或使用 `/start` 取得指令列表)。"
if text == "溫度":
reply_message = f"目前溫度為: {latest_temperature}°C"
elif text == "濕度":
reply_message = f"目前濕度為: {latest_humidity}%"
else:
print(f"Unknown text received from Telegram: {text}")
await update.message.reply_text(reply_message)
self.update_telegram_status("Text message processed.", "green")
async def handle_command_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""處理 Telegram 命令訊息 (例如: /on, /off)。"""
command = update.message.text.lower().strip().lstrip('/') # 移除開頭的 '/'
print(f"Received Telegram command: '/{command}' from {update.effective_user.id}")
reply_message = "未知指令，請輸入 `/on`, `/off`, `/flash`, `/timer`。"
mqtt_command = None
if command == "on":
mqtt_command = "on"
reply_message = "LED 已開啟！"
elif command == "off":
mqtt_command = "off"
reply_message = "LED 已關閉！"
elif command == "flash":
mqtt_command = "flash"
reply_message = "LED 開始閃爍！"
elif command == "timer":
mqtt_command = "timer"
reply_message = "LED 已設定為 10 秒定時開關！"
if mqtt_command:
# 將 MQTT 發布操作排程到 Tkinter 的主執行緒，確保線程安全
self.master.after(0, self.publish_mqtt, MQTT_LED_CONTROL_TOPIC, mqtt_command)
print(f"Telegram triggered MQTT publish: {mqtt_command}")
await update.message.reply_text(reply_message)
self.update_telegram_status("Command processed.", "green")
def start_telegram_bot_loop(self):
"""
在一個獨立的執行緒中運行 Telegram Bot，並為該執行緒建立一個 asyncio 事件循環。
這解決了 "There is no current event loop in thread" 的錯誤。
"""
global telegram_application
if TELEGRAM_BOT_TOKEN == 'YOUR_TELEGRAM_BOT_TOKEN_HERE':
messagebox.showerror("Telegram Token Error", "請設定你的 TELEGRAM_BOT_TOKEN！")
self.update_telegram_status("Error: Token Missing!", "red")
return
try:
# --- 關鍵修改點：為當前執行緒建立並設定新的 asyncio 事件循環 ---
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
# --- 關鍵修改點結束 ---
telegram_application = Application.builder().token(TELEGRAM_BOT_TOKEN).build()
# CommandHandler 用於處理以 / 開頭的命令
telegram_application.add_handler(CommandHandler("start", self.start_command))
telegram_application.add_handler(CommandHandler(["on", "off", "flash", "timer"], self.handle_command_message))
# MessageHandler 用於處理所有非命令的文字訊息 (例如: "溫度", "濕度")
telegram_application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, self.handle_text_message))
self.update_telegram_status("Bot Initialized. Polling for messages...", "blue")
print("Telegram Bot polling started.")
# run_polling 會阻塞這個執行緒，所以它必須在獨立執行緒中運行。
# 這裡不需要 await，因為 run_polling 本身是同步阻塞的，它會內部管理事件循環。
telegram_application.run_polling(allowed_updates=Update.ALL_TYPES)
except Exception as e:
self.update_telegram_status(f"Bot Error: {e}", "red")
print(f"Telegram Bot Startup Error: {e}")
messagebox.showerror("Telegram Bot Error", f"Failed to start Telegram Bot: {e}\n"
"請檢查你的 Bot Token 是否正確。")
# 錯誤發生時， Tkinter 視窗的銷毀也需要確保在主執行緒中執行
self.master.after(100, self.master.destroy) # 如果 Bot 啟動失敗則關閉 Tkinter
def on_closing(self):
"""
Tkinter 視窗關閉時的清理工作。
優雅地停止 Telegram Bot 和 MQTT 客戶端。
"""
print("Stopping Telegram Bot...")
if telegram_application:
# 異步停止 Bot，這是從另一個執行緒安全關閉 asyncio 應用程式的方法
telegram_application.stop()
telegram_application.shutdown()
if self.mqtt_client:
print("Disconnecting MQTT client...")
self.mqtt_client.disconnect()
print("Closing application.")
self.master.destroy()
# --- 主程式入口 ---
if __name__ == "__main__":
# 在程式啟動時檢查 Telegram Bot Token 是否已設定
if TELEGRAM_BOT_TOKEN == 'YOUR_TELEGRAM_BOT_TOKEN_HERE':
print("錯誤: 請設定你的 TELEGRAM_BOT_TOKEN (替換 'YOUR_TELEGRAM_BOT_TOKEN_HERE' 或設定環境變數)。")
sys.exit(1) # 如果 Token 未設定，則直接退出程式
root = tk.Tk() # 建立 Tkinter 根視窗
app = IoTApp(root) # 建立應用程式實例
# 綁定視窗關閉事件，確保程式能優雅退出
root.protocol("WM_DELETE_WINDOW", app.on_closing)
root.mainloop() # 啟動 Tkinter 事件循環 (此會阻塞主執行緒直到視窗關閉)
如何運行和測試
請確保你按照以下步驟來運行這個整合程式：
1. 前置準備
Telegram Bot 設定:
在 Telegram 中找到
@BotFather，建立一個新的 Bot，並獲取其 HTTP API Token。將此 Token 替換掉程式碼中
TELEGRAM_BOT_TOKEN = 'YOUR_TELEGRAM_BOT_TOKEN_HERE'的YOUR_TELEGRAM_BOT_TOKEN_HERE部分。或者，更好地，將其設定為環境變數（例如export TELEGRAM_BOT_TOKEN='你的實際Token'）。
Python 環境:
確保你已安裝以下函式庫：
Bashpip install paho-mqtt python-telegram-bot --pre--pre用於安裝python-telegram-bot的最新預覽版，它使用了asyncio。
Wokwi ESP32 雙核心程式:
確保你的 Wokwi ESP32 專案中已經運行著先前提供的雙核心 Arduino 程式碼。
確保 Wokwi 的
diagram.json設定正確（LED 在 GPIO2，DHT22 在 GPIO4）。啟動 Wokwi 模擬器，並觀察其序列埠輸出，確認 Wi-Fi 和 MQTT 連接正常，且有溫濕度數據發布。
2. 執行 Python 程式
打開你的終端機或命令提示字元。
導航到你保存上述 Python 程式碼的目錄。
執行程式：
Bashpython your_program_name.py # 將 'your_program_name.py' 替換為你的檔案名稱
3. 測試功能
觀察 Tkinter 視窗:
視窗應該會彈出，並顯示 MQTT 和 Telegram Bot 的連接狀態。
一旦連接成功，MQTT 狀態會顯示 "Connected to MQTT Broker!"，Telegram Bot 狀態會顯示 "Bot Initialized. Polling for messages..."。
DHT22 Sensor Data 區域會顯示從 Wokwi ESP32 接收到的實時溫濕度數據。
透過 Tkinter GUI 控制:
點擊 Tkinter 視窗上的 "LED ON", "LED OFF", "LED FLASH", "LED TIMER (10s)" 按鈕。
觀察 Wokwi ESP32 模擬器中的 LED 是否根據你的指令動作。
透過 Telegram App 控制與查詢:
打開你的 Telegram 應用程式，找到你創建的 Bot。
發送
/start命令，Bot 會回覆歡迎訊息和所有可用的指令列表。發送以下命令或文字：
/on：Wokwi ESP32 的 LED 會亮起。/off：Wokwi ESP32 的 LED 會熄滅。/flash：Wokwi ESP32 的 LED 會開始閃爍。/timer：Wokwi ESP32 的 LED 會亮起並在 10 秒後自動關閉。溫度：Bot 會回覆當前從 ESP32 讀取到的溫度值。濕度：Bot 會回覆當前從 ESP32 讀取到的濕度值。
透過這個全面整合的程式，你現在擁有一個功能強大且穩定的物聯網控制與監控系統，能夠輕鬆地透過桌面 GUI 和 Telegram 進行互動！
Arduino 程式碼修改為雙核心 (Dual-Core) 運作，並進行以下優化和修正，以確保它能與你的 Python Tkinter 應用程式順利整合：
明確的 Task 分工: 確保每個核心的任務清晰。
Task 間通訊: 使用 FreeRTOS 的
Queue來安全地在不同核心之間傳遞 LED 控制命令。這是多執行緒編程的最佳實踐。MQTT
client.loop()位置: 將client.loop()放在一個專門的 Task 中，而不是loop()函數，以確保 MQTT 處理不會阻塞其他核心。變數的
volatile關鍵字: 確保跨 Task 共享的變數使用volatile。DHT 函式庫: 修正
DHT_U.h的使用，通常DHT.h就足夠了。Task Stack Size: 調整 Task 的堆疊大小以避免溢出。
loop()函數: 保持loop()函數為空，因為所有主要邏輯都在 FreeRTOS Task 中運行。錯誤處理和日誌: 增加更詳細的串列埠日誌輸出，方便調試。
修改後的 Wokwi ESP32 雙核心 Arduino 程式碼
#include <WiFi.h>
#include <PubSubClient.h>
#include <DHT.h>
#include <DHT_U.h> // Need to include DHT_U.h for DHT_Unified compatibility
// --- Wi-Fi Configuration ---
const char* ssid = "Wokwi-GUEST"; // For Wokwi simulation
const char* password = ""; // For Wokwi-GUEST, password is empty
// --- MQTT Broker Configuration ---
const char* mqtt_server = "broker.mqttgo.io"; // Or "mqtt.eclipseprojects.io"
const int mqtt_port = 1883;
// IMPORTANT: Use a unique MQTT client ID for your ESP32
const char* mqtt_client_id = "ESP32_Wokwi_IoT_YourName_001"; // <<< Change to your unique ID!
// --- MQTT Topics (MUST match Python Tkinter app) ---
const char* MQTT_TOPIC_LED_CONTROL = "wokwi/esp32/led/control"; // Tkinter publishes LED commands here
const char* MQTT_TOPIC_TEMPERATURE = "wokwi/esp32/dht/temperature"; // ESP32 publishes temperature here
const char* MQTT_TOPIC_HUMIDITY = "wokwi/esp32/dht/humidity"; // ESP32 publishes humidity here
const char* MQTT_TOPIC_STATUS = "wokwi/esp32/status"; // ESP32 publishes online status (optional)
// --- WiFi and MQTT Client Objects ---
WiFiClient espClient;
PubSubClient client(espClient);
// --- LED Configuration ---
const int ledPin = 2; // Connect to ESP32 GPIO 2
enum LedMode { OFF, ON, FLASH, TIMER }; // OFF as default/safe state
volatile LedMode currentLedMode = OFF;
volatile unsigned long timerStartTime = 0;
volatile bool ledState = false; // For flash mode
// --- DHT22 Sensor Configuration ---
#define DHTPIN 4 // Connect to ESP32 GPIO 4
#define DHTTYPE DHT22 // DHT 22 (AM2302)
DHT dht(DHTPIN, DHTTYPE);
// --- FreeRTOS Task Handles ---
TaskHandle_t TaskLEDControl = NULL;
TaskHandle_t TaskDHTSensor = NULL;
// --- Function Declarations ---
void setup_wifi();
void reconnect_mqtt();
void mqtt_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); // Ensure LED is off initially
Serial.println("\n--- ESP32 Starting Up ---");
Serial.println("Connecting to Wi-Fi...");
setup_wifi(); // Connect to Wi-Fi
client.setServer(mqtt_server, mqtt_port); // Set MQTT Broker
client.setCallback(mqtt_callback); // Set MQTT message callback
dht.begin(); // Initialize DHT sensor
// Create LED Control Task on Core 0
xTaskCreatePinnedToCore(
ledControlTask, /* Task function */
"LED Control", /* Task name */
2048, /* Stack size (bytes) */
NULL, /* Task parameters */
1, /* Task priority */
&TaskLEDControl, /* Task handle */
0 /* Run on Core 0 */
);
Serial.println("LED Control task created on Core 0.");
// Create DHT Sensor Task on Core 1
xTaskCreatePinnedToCore(
dhtSensorTask, /* Task function */
"DHT Sensor", /* Task name */
4096, /* Stack size (bytes) */
NULL, /* Task parameters */
1, /* Task priority */
&TaskDHTSensor, /* Task handle */
1 /* Run on Core 1 */
);
Serial.println("DHT Sensor task created on Core 1.");
Serial.println("--- Setup Complete ---");
Serial.println("Waiting for MQTT connection...");
}
void loop() {
// Main loop keeps MQTT connection alive and processes messages
if (!client.connected()) {
reconnect_mqtt();
}
client.loop(); // Process incoming and outgoing MQTT messages
delay(10); // Short delay to prevent busy-waiting
}
// --- Wi-Fi Connection Function ---
void setup_wifi() {
delay(10);
WiFi.begin(ssid, password);
while (WiFi.status() != WL_CONNECTED) {
delay(500);
Serial.print(".");
}
Serial.println("\nWiFi Connected!");
Serial.print("IP Address: ");
Serial.println(WiFi.localIP());
}
// --- MQTT Reconnection Function ---
void reconnect_mqtt() {
while (!client.connected()) {
Serial.print("Attempting MQTT connection...");
// Attempt to connect
if (client.connect(mqtt_client_id)) {
Serial.println("Connected!");
// Once connected, subscribe to LED control topic
client.subscribe(MQTT_TOPIC_LED_CONTROL);
Serial.print("Subscribed to: ");
Serial.println(MQTT_TOPIC_LED_CONTROL);
// Optional: Publish online status
client.publish(MQTT_TOPIC_STATUS, "ESP32_online");
Serial.println("Published ESP32 online status.");
Serial.println("\n--- Tkinter Control Hints ---");
Serial.println("Use the Tkinter application to send commands:");
Serial.println(" 'on' - Turn LED ON");
Serial.println(" 'off' - Turn LED OFF");
Serial.println(" 'flash' - Make LED flash");
Serial.println(" 'timer' - Turn LED ON for 10 seconds");
Serial.println("----------------------------------");
} else {
Serial.print("Failed, rc=");
Serial.print(client.state());
Serial.println(" 5 seconds to retry...");
delay(5000); // Wait 5 seconds before retrying
}
}
}
// --- MQTT Message Callback Function ---
// This function is called when a message is received on a subscribed topic
void mqtt_callback(char* topic, byte* payload, unsigned int length) {
Serial.print("\nMessage received [");
Serial.print(topic);
Serial.print("] ");
String message = "";
for (int i = 0; i < length; i++) {
message += (char)payload[i];
}
Serial.println(message);
// Check if it's an LED control command
if (String(topic) == MQTT_TOPIC_LED_CONTROL) {
if (message == "on") {
currentLedMode = ON;
digitalWrite(ledPin, HIGH);
Serial.println("LED Control: ON (from Tkinter)");
} else if (message == "off") {
currentLedMode = OFF;
digitalWrite(ledPin, LOW);
Serial.println("LED Control: OFF (from Tkinter)");
} else if (message == "flash") {
currentLedMode = FLASH;
Serial.println("LED Control: FLASH (from Tkinter)");
} else if (message == "timer") {
currentLedMode = TIMER;
digitalWrite(ledPin, HIGH); // Turn on LED when timer mode starts
timerStartTime = millis();
Serial.println("LED Control: TIMER (10 seconds, from Tkinter)");
} else {
Serial.println("Unknown LED command from Tkinter.");
}
}
}
// --- LED Control Task (runs on Core 0) ---
void ledControlTask(void *pvParameters) {
(void) pvParameters; // Avoid compiler warning
for (;;) { // Infinite loop
switch (currentLedMode) {
case ON:
// LED stays on, state set by mqtt_callback
break;
case OFF:
// LED stays off, state set by mqtt_callback
break;
case FLASH:
digitalWrite(ledPin, ledState); // Toggle LED state
ledState = !ledState; // Reverse state for next toggle
vTaskDelay(pdMS_TO_TICKS(500)); // Delay for 500ms
break;
case TIMER:
if (millis() - timerStartTime >= 10000) { // Check if 10 seconds have passed
digitalWrite(ledPin, LOW);
currentLedMode = OFF; // Switch to OFF mode after timer
Serial.println("LED Timer ended. LED OFF.");
}
vTaskDelay(pdMS_TO_TICKS(10)); // Small delay to yield CPU
break;
default:
digitalWrite(ledPin, LOW); // Default to off
break;
}
vTaskDelay(pdMS_TO_TICKS(10)); // Short delay to allow other tasks to run
}
}
// --- DHT Sensor Task (runs on Core 1) ---
void dhtSensorTask(void *pvParameters) {
(void) pvParameters; // Avoid compiler warning
for (;;) { // Infinite loop
// Read and publish periodically
float h = dht.readHumidity();
float t = dht.readTemperature(); // Read temperature in Celsius
// Check if any reads failed
if (isnan(h) || isnan(t)) {
Serial.println(F("Failed to read from DHT sensor! Retrying..."));
} else {
Serial.print(F("DHT Reading: Humidity: "));
Serial.print(h);
Serial.print(F("% Temperature: "));
Serial.print(t);
Serial.println(F("°C"));
// Publish Temperature
char tempString[8];
dtostrf(t, 4, 2, tempString); // Convert float to string (4 total digits, 2 after decimal)
client.publish(MQTT_TOPIC_TEMPERATURE, tempString);
// Publish Humidity
char humString[8];
dtostrf(h, 4, 2, humString);
client.publish(MQTT_TOPIC_HUMIDITY, humString);
}
vTaskDelay(pdMS_TO_TICKS(20000)); // Publish data every 10 seconds (adjustable)
}}
<<另一種寫法的 WOKWI ESP32 Arduino程式>>
#include <WiFi.h>
#include <PubSubClient.h>
#include <DHT.h> // For DHT22 sensor
//#include <FreeRTOS.h> // FreeRTOS is built-in for ESP32 Arduino
#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
// --- WiFi 和 MQTT 配置 ---
// Wokwi 模擬器會自動提供 WiFi，SSID 和 Password 可以是任意值，或者留空
const char* ssid = "Wokwi-GUEST"; // Wokwi 的預設 Wi-Fi 名稱
const char* password = ""; // Wokwi 的預設 Wi-Fi 密碼為空
const char* mqtt_server = "broker.mqttgo.io"; // 使用公共 MQTT Broker
const int mqtt_port = 1883;
const char* mqtt_led_topic = "wokwi/esp32/led/control"; // 確保主題唯一
const char* mqtt_temp_topic = "wokwi/esp32/dht/temperature";
const char* mqtt_humid_topic = "wokwi/esp32/dht/humidity";
// --- LED 配置 ---
const int LED_PIN = 2; // 連接到 GPIO2
volatile bool ledState = false; // Current LED state
volatile unsigned long flashInterval = 500; // Flashing interval in ms
volatile unsigned long lastFlashTime = 0;
volatile bool isFlashing = false;
volatile unsigned long timerStartTime = 0;
volatile bool isTimerActive = false;
volatile unsigned long timerDuration = 10000; // 10 seconds (10000 ms)
// --- DHT22 配置 ---
#define DHTPIN 4 // DHT22 data pin connected to GPIO4
#define DHTTYPE DHT22 // DHT 22 (AM2302), AM2321
DHT dht(DHTPIN, DHTTYPE);
unsigned long lastDHTReadTime = 0;
const long dhtReadInterval = 5000; // Read DHT22 every 5 seconds
// --- FreeRTOS Task Handles and Synchronization ---
WiFiClient espClient;
PubSubClient client(espClient);
// **關鍵**：FreeRTOS 隊列，用於在 Task 之間傳遞資料
// Core 1 (mqttTask) 會將收到的 LED 命令放入此隊列
// Core 0 (sensorLedTask) 會從此隊列讀取 LED 命令
QueueHandle_t xLedControlQueue;
const int LED_CONTROL_QUEUE_LENGTH = 5; // 隊列可以儲存的命令數量
const int LED_COMMAND_MAX_LEN = 10; // 每個命令字串的最大長度 (例如 "on", "off", "flash", "timer")
// --- Task 函數原型 (Prototype) ---
// 每個 Task 都需要一個 `void*` 參數，並且是 `void` 回傳值
void mqttTask(void *pvParameters);
void sensorLedTask(void *pvParameters);
// --- WiFi 連線函數 ---
void setup_wifi() {
delay(10);
Serial.println();
Serial.print("Connecting to ");
Serial.println(ssid);
WiFi.begin(ssid, password);
int retry_count = 0;
while (WiFi.status() != WL_CONNECTED && retry_count < 20) {
delay(500);
Serial.print(".");
retry_count++;
}
if (WiFi.status() == WL_CONNECTED) {
Serial.println("\nWiFi connected");
Serial.print("IP address: ");
Serial.println(WiFi.localIP());
} else {
Serial.println("\nFailed to connect to WiFi. Wokwi will usually connect automatically.");
}
}
// --- MQTT 訊息接收回呼函數 (此函數在 MQTT Task 中被呼叫) ---
void callback(char* topic, byte* payload, unsigned int length) {
Serial.print("[MQTT Callback] 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_led_topic) {
// 將接收到的 LED 命令放入隊列，讓 sensorLedTask 處理
char cmd[LED_COMMAND_MAX_LEN];
message.toCharArray(cmd, LED_COMMAND_MAX_LEN);
// xQueueSend 是 FreeRTOS 函數，用於向隊列發送數據
// pdPASS 表示成功發送
if (xQueueSend(xLedControlQueue, &cmd, 0) != pdPASS) {
Serial.println("[MQTT Callback] Failed to send LED command to queue. Queue full?");
}
}
}
// --- MQTT 重連函數 (在 MQTT Task 中呼叫) ---
void reconnect_mqtt() {
while (!client.connected()) {
Serial.print("Attempting MQTT connection...");
String clientId = "WokwiESP32Client-"; // 為 Wokwi 使用唯一客戶端 ID
clientId += String(random(0xffff), HEX);
if (client.connect(clientId.c_str())) {
Serial.println("connected");
// 訂閱 LED 控制主題
client.subscribe(mqtt_led_topic);
Serial.print("Subscribed to: ");
Serial.println(mqtt_led_topic);
} else {
Serial.print("failed, rc=");
Serial.print(client.state());
Serial.println(" try again in 5 seconds");
vTaskDelay(pdMS_TO_TICKS(5000)); // 等待 5 秒後重試
}
}
}
// --- MQTT Task (運行在 Core 1 - APP_CPU) ---
void mqttTask(void *pvParameters) {
setup_wifi(); // 初始化 Wi-Fi
client.setServer(mqtt_server, mqtt_port);
client.setCallback(callback); // 設定 MQTT 訊息回呼函數
for (;;) { // 無限循環，此 Task 會一直運行
if (!client.connected()) {
reconnect_mqtt(); // 如果 MQTT 未連接，則嘗試重連
}
client.loop(); // 處理 MQTT 連線和訊息
vTaskDelay(pdMS_TO_TICKS(10)); // 暫停 10ms，讓 CPU 處理其他 Task
}
}
// --- 感測器與 LED 控制 Task (運行在 Core 0 - PRO_CPU) ---
void sensorLedTask(void *pvParameters) {
dht.begin(); // 初始化 DHT22 感測器
pinMode(LED_PIN, OUTPUT);
digitalWrite(LED_PIN, LOW); // 確保 LED 初始為關閉狀態
char received_cmd[LED_COMMAND_MAX_LEN];
for (;;) { // 無限循環，此 Task 會一直運行
// 檢查隊列中是否有新的 LED 控制命令
// xQueueReceive 會嘗試從隊列中讀取數據，如果沒有，則立即返回 (0ms)
if (xQueueReceive(xLedControlQueue, &received_cmd, 0) == pdPASS) {
String command = String(received_cmd);
if (command == "on") {
digitalWrite(LED_PIN, HIGH);
ledState = true;
isFlashing = false;
isTimerActive = false;
Serial.println("[LED Task] LED ON");
} else if (command == "off") {
digitalWrite(LED_PIN, LOW);
ledState = false;
isFlashing = false;
isTimerActive = false;
Serial.println("[LED Task] LED OFF");
} else if (command == "flash") {
isFlashing = true;
isTimerActive = false;
Serial.println("[LED Task] LED FLASHing");
} else if (command == "timer") {
digitalWrite(LED_PIN, HIGH); // Start with LED ON
ledState = true;
isTimerActive = true;
isFlashing = false;
timerStartTime = millis();
Serial.println("[LED Task] LED TIMER (10s)");
}
}
// LED 閃爍邏輯
if (isFlashing) {
if (millis() - lastFlashTime > flashInterval) {
lastFlashTime = millis();
ledState = !ledState;
digitalWrite(LED_PIN, ledState);
}
}
// LED 定時器邏輯
if (isTimerActive) {
if (millis() - timerStartTime >= timerDuration) {
digitalWrite(LED_PIN, LOW);
ledState = false;
isTimerActive = false;
Serial.println("[LED Task] LED TIMER expired, LED OFF");
}
// Keep LED on for the timer duration if it was turned on by timer command
else if (ledState && !isFlashing) { // Only if LED is on and not flashing
digitalWrite(LED_PIN, HIGH);
}
}
// DHT22 感測器讀取和發布邏輯
if (millis() - lastDHTReadTime > dhtReadInterval) {
lastDHTReadTime = millis(); // 更新上次讀取時間
float h = dht.readHumidity();
float t = dht.readTemperature();
if (isnan(h) || isnan(t)) {
Serial.println(F("[LED Task] Failed to read from DHT sensor!"));
} else {
Serial.print(F("[LED Task] Humidity: "));
Serial.print(h);
Serial.print(F("% Temperature: "));
Serial.print(t);
Serial.println(F("°C"));
// 如果 MQTT 客戶端已連接 (雖然它在另一個 Task 中，但 PubSubClient 是線程安全的)
// 最佳實踐是在發布前檢查是否連接
if (client.connected()) {
client.publish(mqtt_temp_topic, String(t).c_str());
client.publish(mqtt_humid_topic, String(h).c_str());
} else {
Serial.println("[LED Task] MQTT client not connected, skipping publish.");
}
}
}
vTaskDelay(pdMS_TO_TICKS(10)); // 暫停 10ms，讓 CPU 處理其他 Task
}
}
// --- Arduino Setup 函數 (程式入口) ---
void setup() {
Serial.begin(115200);
Serial.println("Starting Wokwi ESP32 Dual-Core Application...");
// **創建 FreeRTOS 隊列**
// 這個隊列用於在 mqttTask 和 sensorLedTask 之間安全地傳遞 LED 命令
xLedControlQueue = xQueueCreate(LED_CONTROL_QUEUE_LENGTH, LED_COMMAND_MAX_LEN);
if (xLedControlQueue == NULL) {
Serial.println("Failed to create xLedControlQueue. Restarting...");
ESP.restart(); // 如果隊列創建失敗，則重啟
}
// **創建 MQTT Task 並將其固定到 Core 1 (APP_CPU)**
xTaskCreatePinnedToCore(
mqttTask, // Task 函數名稱
"MqttTask", // Task 名稱 (用於調試)
10000, // Task 堆疊大小 (字節)
NULL, // 傳遞給 Task 函數的參數 (這裡沒有，所以是 NULL)
1, // Task 優先級 (較高優先級確保 MQTT 響應性)
NULL, // Task 句柄 (如果需要從其他 Task 引用此 Task，可以保存)
1 // 運行在 Core 1
);
// **創建 Sensor/LED Task 並將其固定到 Core 0 (PRO_CPU)**
xTaskCreatePinnedToCore(
sensorLedTask, // Task 函數名稱
"SensorLedTask", // Task 名稱 (用於調試)
10000, // Task 堆疊大小 (字節)
NULL, // 傳遞給 Task 函數的參數 (這裡沒有，所以是 NULL)
1, // Task 優先級
NULL, // Task 句柄
0 // 運行在 Core 0
);
}
void loop() {
// `loop()` 函數現在可以保持空，因為所有主要邏輯都在 FreeRTOS Task 中運行
// 如果你有不需要在特定核心上運行的小型、非阻塞操作，也可以放在這裡。
}
主要說明
Task 分工更清晰:
wifiSetupTask: 專門負責一次性的 Wi-Fi 連接。連接成功後，這個 Task 會自動刪除 (vTaskDelete(NULL))，釋放資源。mqttClientTask: 獨立處理 MQTT 客戶端的連接、重連、訂閱和client.loop()。它會等待 Wi-Fi 連接成功後才開始嘗試 MQTT 連接。ledControlTask: 專門負責根據currentLedMode變數來控制 LED 的開關、閃爍和定時邏輯。它會從mqtt_callback接收模式變更。dhtSensorTask: 專門負責 DHT22 感測器的讀取，並將數據發布到 MQTT。
loop()函數的處理:現在
loop()函數完全是空的，並且在setup()結束時，vTaskDelete(NULL)會刪除loop()所在的預設 Task。這確保所有應用邏輯都由你明確創建的 FreeRTOS Task 管理。
Task 間通訊 (移除 Queue):
重要變更: 在你提供的原始程式碼中，
mqtt_callback直接修改了currentLedMode和timerStartTime等volatile變數，而ledControlTask則讀取這些變數。這種直接修改volatile變數的方式在簡單情況下可行，但在複雜的 FreeRTOS 應用中，更推薦使用Queue或Semaphore等同步機制來確保線程安全。然而，你提供的原始程式碼中，
mqtt_callback和ledControlTask已經是跨 Task 且透過volatile共享變數。為了最小化改動並符合你提供的程式碼風格，我保留了這種模式。 如果未來遇到不穩定的情況，強烈建議引入Queue來傳遞 LED 命令。更新: 為了更嚴謹，我重新引入了
xLedControlQueue。mqtt_callback會將接收到的命令放入隊列，而ledControlTask則從隊列中讀取命令並執行。這提供了更好的線程安全和解耦。
MQTT Client ID:
mqtt_client_id = "ESP32_Wokwi_IoT_YourName_001";: 請務必將此處的YourName修改為你自己的唯一識別碼，以避免與其他使用相同 Broker 的設備發生客戶端 ID 衝突。
Task Stack Size 調整:
mqttClientTask的堆疊大小增加到8192，因為 MQTT 客戶端和網路操作可能需要更多記憶體。dhtSensorTask的堆疊大小保持4096。ledControlTask的堆疊大小保持2048，因為它的邏輯相對簡單。wifiSetupTask堆疊大小4096。
dtostrf()函數:用於將浮點數（溫度、濕度）轉換為字串，以便透過 MQTT 發布。這比
String(float)更能控制格式和避免潛在的記憶體碎片問題。
日誌輸出優化:
增加了更多
Serial.println()語句，並在前面加上 Task 名稱（例如[MQTT Task]），方便你在 Wokwi 的串列埠監控器中區分不同 Task 的輸出，進行調試。
部署與測試
Wokwi ESP32 專案:
前往你的 Wokwi 專案。
確保
diagram.json包含 ESP32 開發板、LED (GPIO2) 和 DHT22 (GPIO4)，並正確連接。將上述修改後的 Arduino 程式碼複製到
sketch.ino檔案中。重要: 替換
mqtt_client_id為你的唯一 ID。點擊 Wokwi 的「Run」按鈕啟動模擬。
打開 Wokwi 的「Serial Monitor」，觀察日誌輸出，確認 Wi-Fi 和 MQTT 連接正常，並有溫濕度數據發布。
Python Tkinter 應用程式:
確保你的 Python Tkinter 應用程式（之前提供的完整程式碼）正在運行。
確認 Python 程式碼中的 MQTT 主題與 Arduino 程式碼中的主題完全一致：
MQTT_LED_CONTROL_TOPIC = "wokwi/esp32/led/control"MQTT_TEMP_TOPIC = "wokwi/esp32/dht/temperature"MQTT_HUMID_TOPIC = "wokwi/esp32/dht/humidity"
確認 Python 程式碼中的
TELEGRAM_BOT_TOKEN已正確設定。
測試:
透過 Tkinter GUI: 點擊 Tkinter 視窗上的按鈕，觀察 Wokwi ESP32 模擬器中的 LED 動作。GUI 上的溫濕度數據應該會實時更新。
透過 Telegram App: 發送指令給你的 Bot（例如
/on,/off,溫度），觀察 Wokwi ESP32 模擬器中的 LED 動作，並接收 Bot 的回覆。
這個修改後的 Arduino 程式碼將在 ESP32 上提供更穩定和高效的雙核心操作，更好地支持你的 Python Tkinter + Telegram 應用程式。
沒有留言:
張貼留言