MQTT 觀念 (Wokwi ESP32 LED DHT22) + (MQTTx 設定)+(Python UI + Sqlite )
WOKWI程式
- 在 Wokwi 上建立專案：
- 前往
Wokwi 網站
- 點擊 "New Project" 並選擇 "Arduino ESP32" (或 "Arduino ESP8266"，如果使用 ESP8266)。
- 在
diagram.json 檔案中，按照上面提供的範例添加 ESP32/ESP8266 開發板、4 個 LED 和 DHT22 感測器，並將 LED 和 DHT22 連接到相應的引腳。
- 在
secrets.yaml 檔案中，填寫您的 Wi-Fi SSID 和密碼。
- 將上面的 Arduino 程式碼複製到
main.ino 檔案中。
- 安裝 Arduino 庫：
- 在 Arduino IDE (或 Wokwi 的內建庫管理器) 中，搜尋並安裝
PubSubClient, Adafruit Unified Sensor, DHT sensor library。
- 運行模擬：
- 點擊 Wokwi 介面上的 "Run" 按鈕。
- 您會在 Serial Monitor 中看到 Wi-Fi 連線進度，以及 MQTT 連線狀態。
- 測試 MQTT 控制：
- 您可以使用任何 MQTT 客戶端 (例如：MQTT Explorer, mosquitto_pub 等) 連接到
broker.mqttgo.io (端口 1883)。
- 訂閱
alex9ufo/ledstatus 和 alex9ufo/temphumi 主題，以查看設備發布的狀態和數據。
- 向
alex9ufo/ledcontrol 主題發布以下訊息來控制 LED：
1on, 2on, 3on, 4on (開啟個別 LED)1off, 2off, 3off, 4off (關閉個別 LED)allon (開啟所有 LED)alloff (關閉所有 LED)
- 測試 DHT22 讀取：
- 在 Wokwi 的 Serial Monitor 中，直接按下
Enter 鍵 (輸入一個空行)。
- 您會看到 ESP32/ESP8266 讀取 DHT22 的溫度和濕度數據，並發布到
alex9ufo/temphumi 主題。
- 前往
Wokwi 網站 - 點擊 "New Project" 並選擇 "Arduino ESP32" (或 "Arduino ESP8266"，如果使用 ESP8266)。
- 在
diagram.json檔案中，按照上面提供的範例添加 ESP32/ESP8266 開發板、4 個 LED 和 DHT22 感測器，並將 LED 和 DHT22 連接到相應的引腳。 - 在
secrets.yaml檔案中，填寫您的 Wi-Fi SSID 和密碼。 - 將上面的 Arduino 程式碼複製到
main.ino檔案中。
- 在 Arduino IDE (或 Wokwi 的內建庫管理器) 中，搜尋並安裝
PubSubClient,Adafruit Unified Sensor,DHT sensor library。
- 點擊 Wokwi 介面上的 "Run" 按鈕。
- 您會在 Serial Monitor 中看到 Wi-Fi 連線進度，以及 MQTT 連線狀態。
- 您可以使用任何 MQTT 客戶端 (例如：MQTT Explorer, mosquitto_pub 等) 連接到
broker.mqttgo.io(端口 1883)。 - 訂閱
alex9ufo/ledstatus和alex9ufo/temphumi主題，以查看設備發布的狀態和數據。 - 向
alex9ufo/ledcontrol主題發布以下訊息來控制 LED：1on,2on,3on,4on(開啟個別 LED)1off,2off,3off,4off(關閉個別 LED)allon(開啟所有 LED)alloff(關閉所有 LED)
- 在 Wokwi 的 Serial Monitor 中，直接按下
Enter鍵 (輸入一個空行)。 - 您會看到 ESP32/ESP8266 讀取 DHT22 的溫度和濕度數據，並發布到
alex9ufo/temphumi主題。
程式碼解釋：
- Wi-Fi 連線：
setup_wifi()函數負責連接到您在secrets.yaml中定義的 Wi-Fi 網路。 - MQTT 連線與回調：
PubSubClient用於處理 MQTT 通訊。client.setServer()設定 MQTT Broker 的地址和端口。client.setCallback(callback)設定了一個回調函數，當訂閱的主題收到訊息時會被觸發。reconnect()函數在 MQTT 連線斷開時嘗試重新連接，並訂閱alex9ufo/ledcontrol主題。
- LED 控制：
callback()函數解析收到的 MQTT 訊息 (1on,2off,allon等)。- 根據訊息內容，使用
digitalWrite()函數控制對應 LED 的狀態 (HIGH 為開，LOW 為關)。 - 控制 LED 後，發布
alex9ufo/ledstatus訊息通知 Broker LED 的狀態。
- DHT22 讀取與發布：
readAndPublishDHT22()函數使用dht.temperature().getEvent()和dht.humidity().getEvent()讀取 DHT22 的溫度和濕度。- 將讀取到的數據格式化成字串，並通過
client.publish()發布到alex9ufo/temphumi主題。
- Serial Monitor 監聽：
loop()函數中，Serial.available()和Serial.readStringUntil('\n')用於檢測 Serial Monitor 是否有輸入。- 當檢測到使用者按下
Enter鍵時 (輸入為空字串)，則呼叫readAndPublishDHT22()函數。
程式碼變動說明：
- 引入 FreeRTOS 庫：
C++
#include <freertos/FreeRTOS.h> #include <freertos/task.h> mqttPublishQueue：- 新增
QueueHandle_t mqttPublishQueue;作為一個 FreeRTOS 佇列句柄。 - 這個佇列用於在 Core 0 (處理訂閱) 和 Core 1 (處理發布) 之間傳遞需要發布的 MQTT 訊息。
- 新增
dhtReadTriggered,lastTemperature,lastHumidity：- 新增
volatile bool dhtReadTriggered = false;作為一個旗標，volatile關鍵字確保編譯器不會對其進行優化，因為它會被不同核心存取。 - 新增
volatile float lastTemperature = 0.0;和volatile float lastHumidity = 0.0;儲存最新的 DHT22 讀數。Core 0 讀取並更新這些值，Core 1 讀取並發布。
- 新增
mqttSubscribeCallback(Core 0 處理)：- 當收到 MQTT 控制訊息並操作 LED 後，不再直接使用
client.publish()。 - 取而代之的是，它將要發布的 LED 狀態訊息 (例如 "LED1 ON") 放入
mqttPublishQueue佇列中，由 Core 1 的發布任務來處理實際的發布操作：C++xQueueSend(mqttPublishQueue, &msgBuffer, portMAX_DELAY);
- 當收到 MQTT 控制訊息並操作 LED 後，不再直接使用
readDHT22()：- 這個函數現在僅負責讀取 DHT22 數據並更新
lastTemperature和lastHumidity。 - 讀取完成後，它會將
dhtReadTriggered設為true，通知 Core 1 有新的 DHT22 數據可以發布。
- 這個函數現在僅負責讀取 DHT22 數據並更新
core0Task(void * parameter)(運行在 Core 0)：- 這是專門為 Core 0 創建的 FreeRTOS 任務。
- 它負責處理 MQTT 連線管理 (
reconnect()) 和 訂閱訊息的處理 (client.loop())。 - 它也負責監聽 Serial Monitor 的輸入，當按下 Enter 時呼叫
readDHT22()。 vTaskDelay(10 / portTICK_PERIOD_MS);是一個短暫的延遲，讓 FreeRTOS 調度器有機會切換到其他任務。
core1Task(void * parameter)(運行在 Core 1)：- 這是專門為 Core 1 創建的 FreeRTOS 任務。
- 它負責從
mqttPublishQueue中接收要發布的訊息，並執行client.publish()。 - 它也定期檢查
dhtReadTriggered旗標，如果為true，則讀取lastTemperature和lastHumidity並發布到alex9ufo/temphumi，然後重置旗標。 vTaskDelay(50 / portTICK_PERIOD_MS);提供一個適度的延遲，避免無限迴圈耗盡 CPU 資源。
setup()函數的變更：- 在
setup()中，我們不再直接設定client.setCallback()，因為它會移到core0Task內部。 - 新增
mqttPublishQueue = xQueueCreate(10, sizeof(char[50]));來創建佇列。 - 最重要的是，使用
xTaskCreatePinnedToCore()函數來創建兩個任務，並將它們固定 (Pinned) 到指定的 CPU 核心：C++xTaskCreatePinnedToCore( core0Task, // 任務函數 "Core0Task", // 任務名稱 10000, // 任務堆疊大小 (位元組) NULL, // 傳遞給任務的參數 1, // 任務優先級 NULL, // 任務句柄 0 // 運行在 Core 0 ); xTaskCreatePinnedToCore( core1Task, // 任務函數 "Core1Task", // 任務名稱 10000, // 任務堆疊大小 (位元組) NULL, // 傳遞給任務的參數 1, // 任務優先級 NULL, // 任務句柄 1 // 運行在 Core 1 );
- 在
loop()函數的變更：loop()函數現在是空的，因為所有的主要邏輯都已經在 FreeRTOS 任務中運行。vTaskDelete(NULL);用於在setup()執行完畢後，將loop任務本身刪除，確保所有的執行都由我們創建的 FreeRTOS 任務來管理。
Python程式
import tkinter as tk
from tkinter import ttk, messagebox
import paho.mqtt.client as mqtt
import sqlite3
import datetime
import threading
import time
# --- MQTT 配置 ---
MQTT_BROKER = "broker.mqttgo.io"
MQTT_PORT = 1883
MQTT_CLIENT_ID = "python_tkinter_client_alex" # 確保這個 ID 是唯一的
MQTT_LED_CONTROL_TOPIC = "alex9ufo/ledcontrol"
MQTT_LED_STATUS_TOPIC = "alex9ufo/ledstatus"
MQTT_TEMP_HUMI_TOPIC = "alex9ufo/temphumi"
# --- SQLite 配置 ---
DB_NAME = "led_status.db"
INITIAL_ID = 10001 # 資料庫記錄的起始 ID
# --- 全局變數 ---
mqtt_client = None
db_conn = None
db_cursor = None
next_id = INITIAL_ID # 用於追蹤下一個可用的 ID
# --- Tkinter GUI 類別 ---
class MQTTLEDControlApp:
def __init__(self, master):
self.master = master
master.title("Wokwi LED & DHT22 Control")
master.geometry("800x700") # 調整視窗大小
# 初始化資料庫
self.init_db()
# 創建 GUI 元件
self.create_widgets()
# 初始化 MQTT 客戶端
self.init_mqtt()
# 程式啟動時，從資料庫載入並顯示所有內容
self.display_all_db_entries()
def init_db(self):
"""初始化 SQLite 資料庫，建立表格並設定下一個可用 ID。"""
global db_conn, db_cursor, next_id
try:
db_conn = sqlite3.connect(DB_NAME)
db_cursor = db_conn.cursor()
db_cursor.execute('''
CREATE TABLE IF NOT EXISTS led_status (
id INTEGER PRIMARY KEY,
status TEXT NOT NULL,
date TEXT NOT NULL,
time TEXT NOT NULL
)
''')
db_conn.commit()
# 檢查現有數據的最大 ID，確保新的 ID 從其遞增
db_cursor.execute(f"SELECT MAX(id) FROM led_status WHERE id >= {INITIAL_ID}")
max_id_result = db_cursor.fetchone()[0]
if max_id_result is not None:
next_id = max_id_result + 1
else:
next_id = INITIAL_ID # 如果資料庫是空的或者最大 ID 小於起始 ID
print(f"Database initialized. Next ID will be: {next_id}")
except sqlite3.Error as e:
messagebox.showerror("資料庫錯誤", f"初始化資料庫失敗: {e}")
self.master.destroy() # 若資料庫初始化失敗則終止程式
def init_mqtt(self):
"""初始化 MQTT 客戶端並嘗試連接 Broker。"""
global mqtt_client
# 指定 MQTT 協定版本為 VERSION1
mqtt_client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION1, client_id=MQTT_CLIENT_ID)
mqtt_client.on_connect = self.on_connect
mqtt_client.on_message = self.on_message
mqtt_client.on_disconnect = self.on_disconnect
try:
mqtt_client.connect(MQTT_BROKER, MQTT_PORT, 60) # 60 秒的 Keepalive 間隔
mqtt_client.loop_start() # 在背景執行緒中處理網路流量和回調
except Exception as e:
self.broker_status_label.config(text=f"Broker Status: 連線失敗 ({e})", style="Red.TLabel")
messagebox.showerror("MQTT 連線錯誤", f"無法連接到 MQTT Broker: {e}")
def create_widgets(self):
"""建立應用程式的所有 GUI 元件。"""
# --- Broker 狀態顯示 ---
status_frame = ttk.LabelFrame(self.master, text="MQTT Broker Status")
status_frame.pack(pady=10, padx=10, fill="x")
# 創建 ttk.Style 物件來設定 ttk.Label 的樣式
self.style = ttk.Style()
self.style.configure("Red.TLabel", foreground="red", font=("Arial", 12, "bold"))
self.style.configure("Green.TLabel", foreground="green", font=("Arial", 12, "bold"))
# 預設狀態為紅色
self.broker_status_label = ttk.Label(status_frame, text="Broker Status: 未連線", style="Red.TLabel")
self.broker_status_label.pack(pady=5)
# --- LED 控制區 ---
led_control_frame = ttk.LabelFrame(self.master, text="LED Control")
led_control_frame.pack(pady=10, padx=10, fill="x")
# 建立 4 個 LED 的開/關按鈕
self.led_buttons = [] # 儲存按鈕引用 (如果需要後期控制按鈕狀態)
for i in range(4):
frame = ttk.Frame(led_control_frame)
frame.pack(side="left", padx=10)
ttk.Label(frame, text=f"LED {i+1}").pack()
on_button = ttk.Button(frame, text="ON", command=lambda idx=i: self.publish_led_control(idx + 1, "on"))
on_button.pack(side="left", padx=5)
off_button = ttk.Button(frame, text="OFF", command=lambda idx=i: self.publish_led_control(idx + 1, "off"))
off_button.pack(side="left", padx=5)
self.led_buttons.append((on_button, off_button))
# 建立 "全部開" / "全部關" 按鈕
all_buttons_frame = ttk.Frame(led_control_frame)
all_buttons_frame.pack(side="right", padx=10)
ttk.Button(all_buttons_frame, text="All ON", command=lambda: self.publish_led_control(0, "allon")).pack(side="left", padx=5)
ttk.Button(all_buttons_frame, text="All OFF", command=lambda: self.publish_led_control(0, "alloff")).pack(side="left", padx=5)
# --- 溫濕度顯示區 ---
dht_frame = ttk.LabelFrame(self.master, text="DHT22 Data")
dht_frame.pack(pady=10, padx=10, fill="x")
self.temp_label = ttk.Label(dht_frame, text="溫度: N/A", font=("Arial", 14))
self.temp_label.pack(side="left", padx=20)
self.humi_label = ttk.Label(dht_frame, text="濕度: N/A", font=("Arial", 14))
self.humi_label.pack(side="left", padx=20)
# --- 資料庫控制區 ---
db_control_frame = ttk.LabelFrame(self.master, text="Database Operations")
db_control_frame.pack(pady=10, padx=10, fill="x")
# 輸入欄位
entry_frame = ttk.Frame(db_control_frame)
entry_frame.pack(pady=5)
ttk.Label(entry_frame, text="ID:").pack(side="left", padx=5)
self.id_entry = ttk.Entry(entry_frame, width=10)
self.id_entry.pack(side="left", padx=5)
ttk.Label(entry_frame, text="Status:").pack(side="left", padx=5)
self.status_entry = ttk.Entry(entry_frame, width=20)
self.status_entry.pack(side="left", padx=5)
# 按鈕
button_frame = ttk.Frame(db_control_frame)
button_frame.pack(pady=5)
ttk.Button(button_frame, text="新增一筆", command=self.add_db_entry).pack(side="left", padx=5)
ttk.Button(button_frame, text="更正一筆", command=self.update_db_entry).pack(side="left", padx=5)
ttk.Button(button_frame, text="查詢一筆", command=self.query_db_entry).pack(side="left", padx=5)
ttk.Button(button_frame, text="刪除一筆", command=self.delete_db_entry).pack(side="left", padx=5)
ttk.Button(button_frame, text="刪除所有資料", command=self.delete_all_db_entries).pack(side="left", padx=5)
ttk.Button(button_frame, text="刷新顯示", command=self.display_all_db_entries).pack(side="left", padx=5)
# 資料顯示區
self.db_display_text = tk.Text(self.master, wrap="word", height=15, width=90, font=("Courier New", 10))
self.db_display_text.pack(pady=10, padx=10, fill="both", expand=True)
self.db_display_text.config(state="disabled") # 預設為只讀
# --- MQTT 回調函數 ---
def on_connect(self, client, userdata, flags, rc):
"""當客戶端連接到 MQTT Broker 時調用。"""
if rc == 0:
# 連線成功，將狀態標籤設為綠色
self.broker_status_label.config(text="Broker Status: 已連線", style="Green.TLabel")
print("Connected to MQTT Broker!")
# 訂閱相關主題
client.subscribe(MQTT_LED_STATUS_TOPIC)
client.subscribe(MQTT_TEMP_HUMI_TOPIC)
print(f"Subscribed to {MQTT_LED_STATUS_TOPIC} and {MQTT_TEMP_HUMI_TOPIC}")
else:
# 連線失敗，將狀態標籤設為紅色
self.broker_status_label.config(text=f"Broker Status: 連線失敗, code {rc}", style="Red.TLabel")
print(f"Failed to connect, return code {rc}\n")
def on_message(self, client, userdata, msg):
"""當客戶端收到訂閱主題的訊息時調用。"""
topic = msg.topic
payload = msg.payload.decode() # 將位元組訊息解碼為字串
print(f"Received message - Topic: {topic}, Payload: {payload}")
if topic == MQTT_LED_STATUS_TOPIC:
self.process_led_status(payload)
elif topic == MQTT_TEMP_HUMI_TOPIC:
self.update_dht_display(payload)
def on_disconnect(self, client, userdata, rc):
"""當客戶端與 MQTT Broker 斷開連接時調用。"""
self.broker_status_label.config(text=f"Broker Status: 連線中斷 ({rc})", style="Red.TLabel")
print(f"Disconnected from MQTT Broker with code {rc}")
# 在新的執行緒中嘗試重新連接，避免阻塞主 GUI 執行緒
threading.Thread(target=self.reconnect_mqtt_async).start()
def reconnect_mqtt_async(self):
"""非同步地嘗試重新連接 MQTT Broker。"""
# 這是個簡單的重連邏輯，實際應用中可能需要更複雜的指數退避算法
time.sleep(5) # 等待 5 秒後重連
try:
mqtt_client.reconnect()
print("Attempting to reconnect MQTT...")
except Exception as e:
print(f"Reconnect failed: {e}")
self.broker_status_label.config(text=f"Broker Status: 重連失敗 ({e})", style="Red.TLabel")
def publish_led_control(self, led_num, action):
"""向 MQTT Broker 發布 LED 控制訊息。"""
if mqtt_client and mqtt_client.is_connected():
if led_num == 0: # 0 代表控制所有 LED (allon/alloff)
message = action
else:
# 組合訊息，例如 "1on", "2off"
message = f"{led_num}{action}"
mqtt_client.publish(MQTT_LED_CONTROL_TOPIC, message)
print(f"Published '{message}' to {MQTT_LED_CONTROL_TOPIC}")
else:
messagebox.showwarning("MQTT 錯誤", "未連接到 MQTT Broker，無法發送控制訊息。")
def process_led_status(self, status_message):
"""處理接收到的 LED 狀態訊息，並將其儲存到資料庫。"""
# 使用 master.after(0, ...) 在主執行緒中執行資料庫操作和 GUI 更新
# 這可以避免 Tkinter 相關的執行緒問題
self.master.after(0, self._update_db_and_display, status_message)
def _update_db_and_display(self, status_message):
"""實際執行資料庫儲存和 GUI 刷新的函數 (在主執行緒中)。"""
global next_id
current_time = datetime.datetime.now()
date_str = current_time.strftime("%Y-%m-%d")
time_str = current_time.strftime("%H:%M:%S")
try:
db_cursor.execute("INSERT INTO led_status (id, status, date, time) VALUES (?, ?, ?, ?)",
(next_id, status_message, date_str, time_str))
db_conn.commit()
print(f"Saved to DB: ID={next_id}, Status='{status_message}'")
next_id += 1 # ID 自動遞增
# 立即更新顯示所有資料庫內容
self.display_all_db_entries()
except sqlite3.Error as e:
messagebox.showerror("資料庫錯誤", f"儲存 LED 狀態失敗: {e}")
def update_dht_display(self, dht_data):
"""更新 GUI 上 DHT22 溫濕度顯示標籤。"""
# 預期格式範例: "Temperature: 25.50C, Humidity: 60.20%"
parts = dht_data.split(',')
temp_str = "溫度: N/A"
humi_str = "濕度: N/A"
for part in parts:
if "Temperature" in part:
temp_str = part.strip()
elif "Humidity" in part:
humi_str = part.strip()
# 在主執行緒中更新 GUI 標籤
self.master.after(0, lambda: self._update_dht_labels(temp_str, humi_str))
def _update_dht_labels(self, temp_str, humi_str):
"""實際更新溫濕度標籤的函數 (在主執行緒中)。"""
self.temp_label.config(text=temp_str)
self.humi_label.config(text=humi_str)
# --- 資料庫操作功能 ---
def display_all_db_entries(self):
"""從資料庫讀取所有記錄並顯示在 Text Widget 中。"""
try:
db_cursor.execute("SELECT * FROM led_status ORDER BY id ASC")
rows = db_cursor.fetchall()
self.db_display_text.config(state="normal") # 允許編輯 Text Widget
self.db_display_text.delete(1.0, tk.END) # 清空現有內容
if not rows:
self.db_display_text.insert(tk.END, "資料庫中沒有任何記錄。\n")
else:
# 標題行
self.db_display_text.insert(tk.END, f"{'ID':<6}{'Status':<20}{'Date':<15}{'Time':<12}\n")
self.db_display_text.insert(tk.END, "-"*53 + "\n") # 分隔線
for row in rows:
self.db_display_text.insert(tk.END, f"{row[0]:<6}{row[1]:<20}{row[2]:<15}{row[3]:<12}\n")
self.db_display_text.config(state="disabled") # 設回只讀
except sqlite3.Error as e:
messagebox.showerror("資料庫錯誤", f"讀取資料庫失敗: {e}")
def add_db_entry(self):
"""根據使用者輸入新增一筆資料到資料庫。"""
global next_id
status = self.status_entry.get().strip() # 取得狀態輸入框內容
if not status:
messagebox.showwarning("輸入錯誤", "請輸入狀態訊息。")
return
current_time = datetime.datetime.now()
date_str = current_time.strftime("%Y-%m-%d")
time_str = current_time.strftime("%H:%M:%S")
try:
db_cursor.execute("INSERT INTO led_status (id, status, date, time) VALUES (?, ?, ?, ?)",
(next_id, status, date_str, time_str))
db_conn.commit()
messagebox.showinfo("成功", f"新增記錄成功，ID: {next_id}")
next_id += 1 # ID 自動遞增
self.display_all_db_entries() # 刷新顯示
self.status_entry.delete(0, tk.END) # 清空狀態輸入框
except sqlite3.Error as e:
messagebox.showerror("資料庫錯誤", f"新增記錄失敗: {e}")
def update_db_entry(self):
"""根據使用者輸入的 ID 更新指定記錄的狀態。"""
record_id_str = self.id_entry.get().strip()
new_status = self.status_entry.get().strip()
if not record_id_str or not new_status:
messagebox.showwarning("輸入錯誤", "請輸入 ID 和新的狀態。")
return
try:
record_id = int(record_id_str)
db_cursor.execute("UPDATE led_status SET status = ? WHERE id = ?", (new_status, record_id))
db_conn.commit()
if db_cursor.rowcount > 0: # 檢查是否有記錄被更新
messagebox.showinfo("成功", f"ID {record_id} 的記錄已更新。")
self.display_all_db_entries() # 刷新顯示
else:
messagebox.showwarning("未找到", f"未找到 ID {record_id} 的記錄。")
except ValueError:
messagebox.showerror("輸入錯誤", "ID 必須是數字。")
except sqlite3.Error as e:
messagebox.showerror("資料庫錯誤", f"更新記錄失敗: {e}")
def query_db_entry(self):
"""根據使用者輸入的 ID 查詢並顯示單筆記錄。"""
record_id_str = self.id_entry.get().strip()
if not record_id_str:
messagebox.showwarning("輸入錯誤", "請輸入要查詢的 ID。")
return
try:
record_id = int(record_id_str)
db_cursor.execute("SELECT * FROM led_status WHERE id = ?", (record_id,))
row = db_cursor.fetchone() # 取得查詢結果的第一行
self.db_display_text.config(state="normal")
self.db_display_text.delete(1.0, tk.END)
if row:
self.db_display_text.insert(tk.END, "查詢結果:\n")
self.db_display_text.insert(tk.END, f"{'ID':<6}{'Status':<20}{'Date':<15}{'Time':<12}\n")
self.db_display_text.insert(tk.END, "-"*53 + "\n")
self.db_display_text.insert(tk.END, f"{row[0]:<6}{row[1]:<20}{row[2]:<15}{row[3]:<12}\n")
else:
self.db_display_text.insert(tk.END, f"未找到 ID {record_id} 的記錄。\n")
self.db_display_text.config(state="disabled")
except ValueError:
messagebox.showerror("輸入錯誤", "ID 必須是數字。")
except sqlite3.Error as e:
messagebox.showerror("資料庫錯誤", f"查詢記錄失敗: {e}")
def delete_db_entry(self):
"""根據使用者輸入的 ID 刪除單筆記錄。"""
record_id_str = self.id_entry.get().strip()
if not record_id_str:
messagebox.showwarning("輸入錯誤", "請輸入要刪除的 ID。")
return
try:
record_id = int(record_id_str)
if messagebox.askyesno("確認刪除", f"確定要刪除 ID {record_id} 的記錄嗎？"):
db_cursor.execute("DELETE FROM led_status WHERE id = ?", (record_id,))
db_conn.commit()
if db_cursor.rowcount > 0: # 檢查是否有記錄被刪除
messagebox.showinfo("成功", f"ID {record_id} 的記錄已刪除。")
self.display_all_db_entries() # 刷新顯示
else:
messagebox.showwarning("未找到", f"未找到 ID {record_id} 的記錄。")
except ValueError:
messagebox.showerror("輸入錯誤", "ID 必須是數字。")
except sqlite3.Error as e:
messagebox.showerror("資料庫錯誤", f"刪除記錄失敗: {e}")
def delete_all_db_entries(self):
"""刪除資料庫中的所有記錄並重置 ID。"""
if messagebox.askyesno("確認刪除所有", "確定要刪除所有資料嗎？此操作不可恢復！"):
try:
db_cursor.execute("DELETE FROM led_status")
db_conn.commit()
messagebox.showinfo("成功", "所有資料已刪除。")
global next_id
next_id = INITIAL_ID # 重置下一個 ID
self.display_all_db_entries() # 刷新顯示
except sqlite3.Error as e:
messagebox.showerror("資料庫錯誤", f"刪除所有資料失敗: {e}")
def on_closing(self):
"""處理視窗關閉事件，關閉 MQTT 和資料庫連線。"""
if mqtt_client:
mqtt_client.loop_stop() # 停止 MQTT 客戶端循環
mqtt_client.disconnect() # 斷開 MQTT 連線
if db_conn:
db_conn.close() # 關閉資料庫連線
self.master.destroy() # 銷毀 Tkinter 視窗
# --- 主程式進入點 ---
if __name__ == "__main__":
root = tk.Tk()
app = MQTTLEDControlApp(root)
# 設定視窗關閉時的回調函數
root.protocol("WM_DELETE_WINDOW", app.on_closing)
root.mainloop() # 啟動 Tkinter 事件循環
//===============================================
這個程式將包含以下功能：
- MQTT 連線與控制：
- 連接到
broker.mqttgo.io。 - 發布 LED 控制訊息 (
alex9ufo/ledcontrol)。 - 訂閱 LED 狀態訊息 (
alex9ufo/ledstatus)。 - 訂閱溫濕度訊息 (
alex9ufo/temphumi)。
- 連接到
- SQLite 資料庫操作：
- 建立
led_status.db資料庫，包含id,status,date,time欄位。 id從10001開始自動遞增作為主索引鍵。- 當收到
alex9ufo/ledstatus訊息時，自動儲存到資料庫。 - 提供 新增、更新、查詢、刪除單筆 和 刪除所有 資料的功能。
- 程式啟動時，將所有資料顯示在 Tkinter 畫面上。
- 建立
- Tkinter GUI 介面：
- 顯示 Broker 連線狀態。
- 4 個 LED 的開/關按鈕，以及 "全部開" / "全部關" 按鈕。
- 即時顯示接收到的溫度和濕度。
- 資料庫操作按鈕和輸入框。
- 一個文字區域 (Text Widget) 用於顯示資料庫內容。
- Wokwi 設置： 確保您的 Wokwi 專案（Arduino ESP32 或 ESP8266）正在運行，並且：
- 它連接到相同的 MQTT Broker (
broker.mqttgo.io)。 - 它訂閱
alex9ufo/ledcontrol主題以接收 LED 控制訊息 (1on,alloff等)。 - 它發布 LED 狀態到
alex9ufo/ledstatus(例如，當 LED 開啟或關閉時發送 "LED1 ON" 或 "ALL LEDs OFF")。 - 它發布 DHT22 溫濕度數據到
alex9ufo/temphumi(格式應為 "Temperature: XX.XXC, Humidity: YY.YY%")。
- 它連接到相同的 MQTT Broker (
程式功能概覽：
- MQTT 連線： 程式啟動後會自動嘗試連接到 MQTT Broker。連線狀態會顯示在視窗頂部。
- LED 控制：
- 視窗左側有獨立的 LED 1 到 LED 4 的 "ON" / "OFF" 按鈕。
- 還有 "All ON" 和 "All OFF" 按鈕，可以一次性控制所有 LED。
- 點擊這些按鈕，程式會向
alex9ufo/ledcontrol主題發布對應的訊息。
- DHT22 數據顯示：
- 視窗中部會實時顯示從
alex9ufo/temphumi主題接收到的溫度和濕度數據。
- 視窗中部會實時顯示從
- SQLite 資料庫：
- 程式會在同一個目錄下創建一個名為
led_status.db的 SQLite 資料庫檔案。 - 當收到
alex9ufo/ledstatus訊息時，狀態會自動連同當前日期和時間儲存到資料庫中。 - 資料庫操作區： 您可以手動執行：
- 新增一筆： 在 "Status" 框中輸入內容，會以自動遞增的 ID 新增記錄。
- 更正一筆： 輸入 "ID" 和新的 "Status" 來更新指定記錄。
- 查詢一筆： 輸入 "ID" 來查詢並顯示單筆記錄。
- 刪除一筆： 輸入 "ID" 來刪除指定記錄。
- 刪除所有資料： 清空資料庫中的所有記錄。
- 刷新顯示： 手動重新載入並顯示所有資料庫記錄。
- 資料顯示區： 視窗底部會顯示資料庫中的所有記錄，您可以滾動查看。
- 程式會在同一個目錄下創建一個名為
沒有留言:
張貼留言