ESP32 雙核心控制 LED 與 DHT22 溫濕度感測器 (Wokwi 模擬) EX10 -- Python GUI , MQTT , Zrok , Line Messaging API
功能類似 ngrok 的軟體和服務，它們各有優缺點，你可以根據自己的需求來選擇：
1. 開源且可自架的方案 (Open Source & Self-Hostable)
如果你追求更高的控制權、安全性和透明度，或者有長期、大量的隧道需求，自架服務會是很好的選擇。
Zrok 是一個類似 Tunnelmole/ngrok 的內網穿透工具，它能將你的本地服務暴露到公共網路。
好的，這次我們將把網路對接工具從 Tunnelmole 替換成 Zrok，來實現 LINE Bot 和本地 Python 應用程式之間的通訊。其他功能，包括 Tkinter GUI 和 MQTT 與 ESP32 的互動，都將保留。
Zrok 是一個類似 Tunnelmole/ngrok 的內網穿透工具，它能將你的本地服務暴露到公共網路。
核心組件概述 (更新為 Zrok)
Wokwi ESP32 模擬器 (Arduino 程式碼)：
MQTT Broker (MQTT 伺服器)：
Python GUI 應用程式 (Python 程式碼)：
包含 Tkinter GUI 來顯示感測器數據和控制 LED。
新增 Flask Webhook，現在將通過 Zrok 暴露到公共網路，以接收來自 LINE 平台的訊息。
包含 LINE Bot SDK 來處理 LINE 訊息和發送回覆。
Zrok (內網穿透工具)：
前置準備：安裝和設定 Zrok
Zrok 的設定比 Tunnelmole 稍微多一些，需要先在 Zrok 網站上註冊並獲取一個 zrok.json 設定檔。
1. 安裝 Zrok CLI 工具
下載 Zrok： 訪問 Zrok 的 GitHub Releases 頁面。根據你的作業系統下載對應的最新版本壓縮包。
例如，Windows 用戶下載 zrok_<版本號>_windows_amd64.zip。
macOS 用戶下載 zrok_<版本號>_darwin_amd64.tar.gz。
Linux 用戶下載 zrok_<版本號>_linux_amd64.tar.gz。
解壓縮： 將下載的壓縮包解壓縮到你喜歡的位置（例如，在 Windows 上解壓縮到 C:\zrok，在 macOS/Linux 上解壓縮到 ~/zrok）。
添加到 PATH (推薦)： 將 Zrok 可執行檔所在的目錄添加到你的系統 PATH 環境變數中，這樣你就可以在任何終端機中直接運行 zrok 命令。
2. 註冊 Zrok 帳戶並獲取 zrok.json
訪問 Zrok 網站： 打開瀏覽器，前往 https://zrok.io/。
註冊/登錄： 點擊 "Sign Up" 註冊一個新帳戶（可以使用 GitHub 登錄）。
建立環境 (Environment)： 登錄後，Zrok 會提示你建立一個新的環境。按照指示操作。
下載環境文件： 在成功建立環境後，Zrok 網站會顯示一個命令，類似於：
將這個命令複製並在你的終端機中執行。 這會在你的當前目錄下生成一個 zrok.json 文件。這個文件包含你的 Zrok 環境設定，請妥善保管，不要分享給他人。 這個 zrok.json 文件會自動被 Zrok CLI 工具檢測到。
程式碼說明與設定 (Python)
這份程式碼是基於你之前的版本，並重新整合了 Flask 和 LINE Bot 功能。
<Python程式>
import tkinter as tk
from tkinter import ttk, scrolledtext, messagebox
import paho.mqtt.client as mqtt
import threading
import time
import os
import sys
# Flask for Webhook (to receive LINE messages)
from flask import Flask, request, abort
# LINE Messaging API SDK v3 imports
from linebot.v3.webhook import WebhookHandler
from linebot.v3.webhooks import MessageEvent, TextMessageContent
from linebot.v3.messaging import Configuration, ApiClient, MessagingApi, ReplyMessageRequest, TextMessage as LineTextMessage
from linebot.exceptions import InvalidSignatureError, LineBotApiError
# --- Configuration Section ---
# MQTT Broker Settings (與 Wokwi ESP32 設定一致)
MQTT_BROKER = "broker.mqttgo.io" # 或 "mqtt.eclipseprojects.io"
MQTT_PORT = 1883
# MQTT Topics (MUST match Wokwi ESP32 code)
MQTT_LED_CONTROL_TOPIC = "wokwi/esp32/led/control"
MQTT_TEMP_TOPIC = "wokwi/esp32/dht/temperature"
MQTT_HUMID_TOPIC = "wokwi/esp32/dht/humidity"
MQTT_STATUS_TOPIC = "wokwi/esp32/status" # ESP32 發布上線狀態
# LINE Messaging API Settings (從 LINE Developers Console 取得)
# 建議從環境變數讀取以保護你的 Token 和 Secret
# 例如在 Linux/macOS:
# export LINE_CHANNEL_ACCESS_TOKEN='你的Channel Access Token'
# export LINE_CHANNEL_SECRET='你的Channel Secret'
LINE_CHANNEL_ACCESS_TOKEN = os.getenv('LINE_CHANNEL_ACCESS_TOKEN', '9hDdyyVKvNlZLVBgmVLAGAmSplmtM9Gs82NDov5z7tGSSP5wJKVRWkI5PZiwLBUw4e99/X3tx0z00+i1J/nwVnInuSs9tLZmiBrkdedf4hwqy1M/jldCgeJvIpt1+MWgFxxC4SUTvTL/AIyWq/X/ANAdB04t89/1O/w1cDnyilFU=')
LINE_CHANNEL_SECRET = os.getenv('LINE_CHANNEL_SECRET', '1ace0f9245c59c652708cd905235647e')
# Flask App port (Zrok 將會轉發到此埠)
FLASK_PORT = 5000
# --- Global variables for latest sensor data ---
latest_temperature = "N/A"
latest_humidity = "N/A"
# Flask App and LINE Bot API instances
app = Flask(__name__)
# LINE Bot SDK v3 Initialization
line_bot_configuration = Configuration(access_token=LINE_CHANNEL_ACCESS_TOKEN)
line_bot_api_client = ApiClient(line_bot_configuration)
line_bot_messaging_api = MessagingApi(line_bot_api_client)
handler = WebhookHandler(LINE_CHANNEL_SECRET)
tkinter_app_instance = None # Reference to the Tkinter application instance
# --- Tkinter Main Application Class ---
class IoTApp:
def __init__(self, master):
self.master = master
master.title("Wokwi ESP32 IoT Control & Monitor (Tkinter with LINE Bot via Zrok)")
master.geometry("700x650") # 增加視窗大小以容納更多內容
self.mqtt_client = None
self.mqtt_reconnect_timer = None # 用於延遲 MQTT 重連嘗試的計時器
self.create_widgets()
self.setup_mqtt()
# Start Flask Webhook service in a separate thread
self.line_webhook_thread = threading.Thread(target=self.start_flask_app, daemon=True)
self.line_webhook_thread.start()
global tkinter_app_instance
tkinter_app_instance = self
def create_widgets(self):
"""Creates all widgets in the Tkinter GUI."""
# --- LED Control Section ---
led_frame = ttk.LabelFrame(self.master, text="LED 控制", padding="10")
led_frame.pack(pady=10, padx=10, fill="x")
ttk.Button(led_frame, text="LED 開啟", command=lambda: self.publish_mqtt(MQTT_LED_CONTROL_TOPIC, "on")).pack(side="left", padx=5, pady=5)
ttk.Button(led_frame, text="LED 關閉", command=lambda: self.publish_mqtt(MQTT_LED_CONTROL_TOPIC, "off")).pack(side="left", padx=5, pady=5)
ttk.Button(led_frame, text="LED 閃爍", command=lambda: self.publish_mqtt(MQTT_LED_CONTROL_TOPIC, "flash")).pack(side="left", padx=5, pady=5)
ttk.Button(led_frame, text="LED 定時 (10秒)", command=lambda: self.publish_mqtt(MQTT_LED_CONTROL_TOPIC, "timer")).pack(side="left", padx=5, pady=5)
# --- DHT22 Sensor Data Display Section ---
dht_frame = ttk.LabelFrame(self.master, text="DHT22 感測器數據", padding="10")
dht_frame.pack(pady=10, padx=10, fill="x")
self.temp_label = ttk.Label(dht_frame, text=f"溫度: {latest_temperature}°C", font=("Arial", 14))
self.temp_label.pack(pady=5, anchor="w")
self.humid_label = ttk.Label(dht_frame, text=f"濕度: {latest_humidity}%", font=("Arial", 14))
self.humid_label.pack(pady=5, anchor="w")
# --- Status Display Section ---
status_frame = ttk.LabelFrame(self.master, text="連接狀態", padding="10")
status_frame.pack(pady=10, padx=10, fill="x")
self.mqtt_status_label = ttk.Label(status_frame, text="MQTT: 連接中...", font=("Arial", 10), foreground="blue")
self.mqtt_status_label.pack(pady=2, anchor="w")
self.flask_status_label = ttk.Label(status_frame, text="Flask: 啟動中...", font=("Arial", 10), foreground="blue")
self.flask_status_label.pack(pady=2, anchor="w")
self.line_status_label = ttk.Label(status_frame, text="LINE Webhook: 等待 Zrok 連接...", font=("Arial", 10), foreground="blue")
self.line_status_label.pack(pady=2, anchor="w")
ttk.Label(status_frame, text=f"Flask 本地埠: {FLASK_PORT}", font=("Arial", 9)).pack(pady=1, anchor="w")
ttk.Label(status_frame, text="請啟動 Zrok 將此埠暴露到公共網路！", font=("Arial", 9), foreground="purple").pack(pady=1, anchor="w")
# --- Message Log Section ---
log_frame = ttk.LabelFrame(self.master, text="訊息日誌", padding="10")
log_frame.pack(pady=10, padx=10, fill="both", expand=True)
self.log_text = scrolledtext.ScrolledText(log_frame, width=80, height=15, wrap=tk.WORD, font=("Consolas", 10))
self.log_text.pack(expand=True, fill="both")
self.log_text.config(state=tk.DISABLED) # Make read-only
def log_message(self, message, tag=None):
"""Appends a message to the log text area."""
self.master.after(0, self._append_log_message, message, tag)
def _append_log_message(self, message, tag):
"""Internal function to append log message (thread-safe)."""
self.log_text.config(state=tk.NORMAL)
self.log_text.insert(tk.END, message + "\n", tag)
self.log_text.see(tk.END) # Scroll to the end
self.log_text.config(state=tk.DISABLED)
# --- MQTT Related Functions ---
def setup_mqtt(self):
"""Sets up MQTT client callbacks and attempts to connect."""
self.mqtt_client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
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
# Initial connection attempt
self.attempt_mqtt_connection()
def attempt_mqtt_connection(self):
"""Attempts to connect to the MQTT broker."""
try:
self.update_mqtt_status("嘗試連接到 MQTT...", "blue")
self.log_message("MQTT: 嘗試連接到 Broker...", "info")
self.mqtt_client.connect(MQTT_BROKER, MQTT_PORT, 60)
# Run MQTT loop in a separate thread to avoid blocking the GUI
if not hasattr(self, 'mqtt_thread') or not self.mqtt_thread.is_alive():
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"錯誤: {e}", "red")
self.log_message(f"MQTT: 初始連接錯誤: {e}", "error")
self.schedule_mqtt_reconnect()
def on_mqtt_connect(self, client, userdata, flags, rc, properties):
"""Callback function when MQTT connects successfully."""
if rc == 0:
self.update_mqtt_status("已連接到 MQTT Broker！", "green")
self.log_message("MQTT: 已連接到 Broker！", "success")
client.subscribe(MQTT_TEMP_TOPIC)
client.subscribe(MQTT_HUMID_TOPIC)
client.subscribe(MQTT_STATUS_TOPIC) # 訂閱 ESP32 狀態
self.log_message(f"MQTT: 已訂閱 {MQTT_TEMP_TOPIC}, {MQTT_HUMID_TOPIC}, {MQTT_STATUS_TOPIC}", "info")
if self.mqtt_reconnect_timer:
self.master.after_cancel(self.mqtt_reconnect_timer)
self.mqtt_reconnect_timer = None
else:
self.update_mqtt_status(f"連接失敗 (代碼 {rc})。正在重試...", "orange")
self.log_message(f"MQTT: 連接失敗, 返回代碼 {rc}。正在重試...", "warning")
self.schedule_mqtt_reconnect()
def on_mqtt_disconnect(self, client, userdata, rc):
"""Callback function when MQTT disconnects."""
self.update_mqtt_status(f"已從 MQTT 斷開連接 (代碼 {rc})。正在重連...", "red")
self.log_message(f"MQTT: 已斷開連接, 結果代碼 {rc}。正在重連...", "error")
# Ensure that if the client disconnects, we re-attempt connection
self.schedule_mqtt_reconnect()
def schedule_mqtt_reconnect(self):
"""Schedules an MQTT reconnection attempt after a delay."""
# Only schedule if not already scheduled
if self.mqtt_reconnect_timer is None:
self.mqtt_reconnect_timer = self.master.after(5000, self.perform_mqtt_reconnect)
self.log_message("MQTT: 已安排 5 秒後重連。", "info")
else:
self.log_message("MQTT: 重連已安排。", "info")
def perform_mqtt_reconnect(self):
"""Performs the actual MQTT reconnection attempt."""
self.mqtt_reconnect_timer = None # Clear the timer as we are about to attempt connection
if not self.mqtt_client.is_connected():
self.log_message("MQTT: 嘗試重連...", "info")
try:
self.mqtt_client.reconnect() # This will trigger on_mqtt_connect if successful
except Exception as e:
self.update_mqtt_status(f"MQTT 重連失敗: {e}", "red")
self.log_message(f"MQTT: 重連失敗: {e}", "error")
self.schedule_mqtt_reconnect() # Schedule another reconnect if this one failed
else:
self.log_message("MQTT: 已連接, 無需重連。", "info")
def on_mqtt_message(self, client, userdata, msg):
"""Callback function when an MQTT message is received, updates sensor data."""
global latest_temperature, latest_humidity
payload = msg.payload.decode()
self.log_message(f"MQTT Rx: 主題='{msg.topic}', 載荷='{payload}'", "mqtt_rx")
if msg.topic == MQTT_TEMP_TOPIC:
latest_temperature = payload
self.master.after(0, self.update_dht_labels)
elif msg.topic == MQTT_HUMID_TOPIC:
latest_humidity = payload
self.master.after(0, self.update_dht_labels)
elif msg.topic == MQTT_STATUS_TOPIC:
self.log_message(f"ESP32 狀態: {payload}", "esp32_status")
def update_dht_labels(self):
"""Updates temperature and humidity display on Tkinter GUI."""
self.temp_label.config(text=f"溫度: {latest_temperature}°C")
self.humid_label.config(text=f"濕度: {latest_humidity}%")
def update_mqtt_status(self, text, color):
"""Updates MQTT status display on Tkinter GUI."""
self.master.after(0, lambda: self.mqtt_status_label.config(text=f"MQTT: {text}", foreground=color))
def update_flask_status(self, text, color):
"""Updates Flask status display on Tkinter GUI."""
self.master.after(0, lambda: self.flask_status_label.config(text=f"Flask: {text}", foreground=color))
def update_line_status(self, text, color):
"""Updates LINE Bot status display on Tkinter GUI."""
self.master.after(0, lambda: self.line_status_label.config(text=f"LINE Webhook: {text}", foreground=color))
def publish_mqtt(self, topic, payload):
"""Publishes an MQTT message to the specified topic."""
if self.mqtt_client and self.mqtt_client.is_connected():
try:
self.mqtt_client.publish(topic, payload)
self.log_message(f"MQTT Tx: 主題='{topic}', 載荷='{payload}'", "mqtt_tx")
self.update_mqtt_status(f"已發送指令: {payload}", "blue")
except Exception as e:
self.update_mqtt_status(f"發布錯誤: {e}", "red")
self.log_message(f"MQTT: 發布錯誤: {e}", "error")
else:
self.update_mqtt_status("MQTT 未連接！指令未發送。", "red")
self.log_message("MQTT: 客戶端未連接，無法發布。指令未發送。", "error")
self.schedule_mqtt_reconnect() # Attempt to reconnect if not connected
# --- LINE Bot Webhook Related Functions (Flask App in separate thread) ---
def start_flask_app(self):
"""Starts the Flask web service to listen for LINE Webhooks."""
if LINE_CHANNEL_ACCESS_TOKEN == 'YOUR_LINE_CHANNEL_ACCESS_TOKEN_HERE' or \
LINE_CHANNEL_SECRET == 'YOUR_LINE_CHANNEL_SECRET_HERE':
messagebox.showerror("LINE Token/Secret 錯誤", "請設定你的 LINE_CHANNEL_ACCESS_TOKEN 和 LINE_CHANNEL_SECRET！")
self.update_line_status("錯誤: Token/Secret 缺失！", "red")
self.log_message("LINE: 錯誤 - Channel Access Token 或 Secret 缺失。", "error")
return
self.update_flask_status(f"正在啟動於埠 {FLASK_PORT}...", "blue")
self.log_message(f"Flask: 應用程式正在啟動於埠 {FLASK_PORT}...", "info")
try:
# Flask app.run() is blocking, so it must be in a separate thread
app.run(host='0.0.0.0', port=FLASK_PORT, debug=False, use_reloader=False)
# This line will only be reached if app.run() exits, which usually means an error or intentional shutdown.
self.update_flask_status(f"已退出於埠 {FLASK_PORT}", "grey") # 不應正常發生
self.log_message(f"Flask: 應用程式已停止運行於埠 {FLASK_PORT}。", "warning")
except Exception as e:
self.update_flask_status(f"錯誤: {e}", "red")
self.log_message(f"Flask: 應用程式啟動錯誤: {e}", "error")
messagebox.showerror("Flask 應用程式錯誤", f"無法啟動 Flask 應用程式用於 LINE: {e}\n"
f"請確保 {FLASK_PORT} 埠沒有被占用，並且手動啟動 Zrok。")
# If Flask fails to start, we should terminate the Tkinter app as well.
self.master.after(100, self.master.destroy)
os._exit(1) # 強制退出並帶有錯誤代碼
def on_closing(self):
"""
Cleanup actions when the Tkinter window is closed.
Gracefully stops the MQTT client. Flask app, being a daemon thread, will automatically exit.
"""
self.log_message("應用程式: 正在關閉...", "info")
if self.mqtt_client:
self.log_message("MQTT: 正在斷開客戶端連接...", "info")
self.mqtt_client.disconnect()
if self.mqtt_reconnect_timer:
self.master.after_cancel(self.mqtt_reconnect_timer)
self.master.destroy()
os._exit(0) # 強制退出以確保所有線程終止
# --- LINE Webhook Handler (Outside IoTApp class, but calls its methods) ---
@app.route("/callback", methods=['POST'])
def line_webhook_callback():
signature = request.headers.get('X-Line-Signature') # Use .get() to avoid KeyError
body = request.get_data(as_text=True)
if tkinter_app_instance:
tkinter_app_instance.log_message(f"LINE Webhook: 收到請求。主體長度: {len(body)}", "line_rx")
# tkinter_app_instance.log_message(f"LINE Webhook Body: {body}", "line_rx") # 取消註釋用於調試原始主體
try:
# Update LINE status *before* handling, it will be updated again by handler
if tkinter_app_instance:
tkinter_app_instance.update_line_status("正在處理 Webhook...", "blue")
handler.handle(body, signature)
if tkinter_app_instance:
tkinter_app_instance.update_line_status("Webhook 處理完成 OK", "green")
return 'OK' # 成功處理後總是返回 200 OK
except InvalidSignatureError:
error_msg = "LINE Webhook: 無效簽名。請檢查 token/secret。"
if tkinter_app_instance:
tkinter_app_instance.log_message(error_msg, "error")
tkinter_app_instance.update_line_status("無效簽名！", "red")
print(error_msg) # 也打印到控制台以供查看
abort(400) # Bad Request
except LineBotApiError as e:
error_msg = f"LINE Webhook: 處理過程中發生 LineBotApiError: {e}"
if tkinter_app_instance:
tkinter_app_instance.log_message(error_msg, "error")
tkinter_app_instance.update_line_status(f"處理錯誤 (LINE API): {e}", "red")
print(error_msg)
abort(500) # Internal Server Error
except Exception as e:
error_msg = f"LINE Webhook: 一般處理錯誤: {e}"
if tkinter_app_instance:
tkinter_app_instance.log_message(error_msg, "error")
tkinter_app_instance.update_line_status(f"處理錯誤 (一般): {e}", "red")
print(error_msg)
abort(500) # Internal Server Error
@handler.add(MessageEvent, message=TextMessageContent)
def handle_line_message(event):
received_text = event.message.text.lower().strip()
user_id = event.source.user_id
if tkinter_app_instance:
tkinter_app_instance.log_message(f"LINE 訊息: 來自 {user_id}, 文本: '{received_text}'", "line_msg")
reply_text = "未知指令。請輸入 '開燈', '關燈', '閃爍', '計時', '溫度' 或 '濕度'。"
mqtt_command = None
if received_text == "開燈":
mqtt_command = "on"
reply_text = "LED 已開啟！"
elif received_text == "關燈":
mqtt_command = "off"
reply_text = "LED 已關閉！"
elif received_text == "閃爍":
mqtt_command = "flash"
reply_text = "LED 開始閃爍！"
elif received_text == "計時":
mqtt_command = "timer"
reply_text = "LED 已設定為 10 秒定時開關！"
elif received_text == "溫度":
reply_text = f"目前溫度為: {latest_temperature}°C"
elif received_text == "濕度":
reply_text = f"目前濕度為: {latest_humidity}%"
elif received_text == "狀態": # 新增的狀態指令
mqtt_status = "連線" if tkinter_app_instance and tkinter_app_instance.mqtt_client.is_connected() else "斷線"
# 這裡的 Flask 和 LINE 狀態會根據日誌顯示來判斷
flask_status_text = tkinter_app_instance.flask_status_label.cget("text").replace("Flask: ", "") if tkinter_app_instance else "未知"
line_status_text = tkinter_app_instance.line_status_label.cget("text").replace("LINE Webhook: ", "") if tkinter_app_instance else "未知"
reply_text = f"系統狀態:\nMQTT: {mqtt_status}\nFlask: {flask_status_text}\nLINE: {line_status_text}\n溫度: {latest_temperature}°C\n濕度: {latest_humidity}%"
if mqtt_command and tkinter_app_instance:
# Important: Call publish_mqtt via master.after to ensure it runs on the main Tkinter thread
# This avoids potential issues with MQTT client being used from a different thread (Flask's thread)
tkinter_app_instance.master.after(0, tkinter_app_instance.publish_mqtt, MQTT_LED_CONTROL_TOPIC, mqtt_command)
tkinter_app_instance.log_message(f"LINE 觸發 MQTT 發布: {mqtt_command}", "line_mqtt_tx")
# Reply message using linebot.v3.messaging.MessagingApi
try:
line_bot_messaging_api.reply_message_with_http_info(
ReplyMessageRequest(
reply_token=event.reply_token,
messages=[LineTextMessage(text=reply_text)]
)
)
if tkinter_app_instance:
tkinter_app_instance.log_message(f"LINE 回覆: '{reply_text}'", "line_reply")
tkinter_app_instance.update_line_status("訊息已處理並回覆。", "green")
except LineBotApiError as e:
error_msg = f"LINE: 回覆失敗 (LineBotApiError): {e.status_code} - {e.error.message}" # 更詳細的錯誤
if tkinter_app_instance:
tkinter_app_instance.log_message(error_msg, "error")
tkinter_app_instance.update_line_status(f"回覆錯誤 (LINE API): {e.status_code}", "red")
print(error_msg)
except Exception as e:
error_msg = f"LINE: 回覆失敗 (一般錯誤): {e}"
if tkinter_app_instance:
tkinter_app_instance.log_message(error_msg, "error")
tkinter_app_instance.update_line_status(f"回覆錯誤 (一般): {e}", "red")
print(error_msg)
# --- Main Program Entry Point ---
if __name__ == "__main__":
if LINE_CHANNEL_ACCESS_TOKEN == 'YOUR_LINE_CHANNEL_ACCESS_TOKEN_HERE' or \
LINE_CHANNEL_SECRET == 'YOUR_LINE_CHANNEL_SECRET_HERE':
print("錯誤: 請設定你的 LINE_CHANNEL_ACCESS_TOKEN 和 LINE_CHANNEL_SECRET。")
messagebox.showerror("LINE API 錯誤", "請設定你的 LINE_CHANNEL_ACCESS_TOKEN 和 LINE_CHANNEL_SECRET！")
sys.exit(1)
root = tk.Tk()
app_instance = IoTApp(root)
tkinter_app_instance = app_instance # 將應用程式實例設置為全局變數
# Configure tags for log messages
app_instance.log_text.tag_config("info", foreground="blue")
app_instance.log_text.tag_config("success", foreground="green")
app_instance.log_text.tag_config("warning", foreground="orange")
app_instance.log_text.tag_config("error", foreground="red")
app_instance.log_text.tag_config("mqtt_tx", foreground="purple")
app_instance.log_text.tag_config("mqtt_rx", foreground="darkgreen")
app_instance.log_text.tag_config("line_rx", foreground="darkblue")
app_instance.log_text.tag_config("line_msg", foreground="magenta")
app_instance.log_text.tag_config("line_reply", foreground="darkcyan")
app_instance.log_text.tag_config("line_mqtt_tx", foreground="brown")
app_instance.log_text.tag_config("esp32_status", foreground="darkorange")
root.protocol("WM_DELETE_WINDOW", app_instance.on_closing)
root.mainloop()
<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 中運行
// 如果你有不需要在特定核心上運行的小型、非阻塞操作，也可以放在這裡。
}
好的，這次我們將把網路對接工具從 Tunnelmole 替換成 Zrok，來實現 LINE Bot 和本地 Python 應用程式之間的通訊。其他功能，包括 Tkinter GUI 和 MQTT 與 ESP32 的互動，都將保留。
Zrok 是一個類似 Tunnelmole/ngrok 的內網穿透工具，它能將你的本地服務暴露到公共網路。
核心組件概述 (更新為 Zrok)
Wokwi ESP32 模擬器 (Arduino 程式碼)：
MQTT Broker (MQTT 伺服器)：
Python GUI 應用程式 (Python 程式碼)：
包含 Tkinter GUI 來顯示感測器數據和控制 LED。
新增 Flask Webhook，現在將通過 Zrok 暴露到公共網路，以接收來自 LINE 平台的訊息。
包含 LINE Bot SDK 來處理 LINE 訊息和發送回覆。
Zrok (內網穿透工具)：
前置準備：安裝和設定 Zrok
Zrok 的設定比 Tunnelmole 稍微多一些，需要先在 Zrok 網站上註冊並獲取一個 zrok.json 設定檔。
1. 安裝 Zrok CLI 工具
下載 Zrok： 訪問 Zrok 的 GitHub Releases 頁面。根據你的作業系統下載對應的最新版本壓縮包。
例如，Windows 用戶下載 zrok_<版本號>_windows_amd64.zip。
macOS 用戶下載 zrok_<版本號>_darwin_amd64.tar.gz。
Linux 用戶下載 zrok_<版本號>_linux_amd64.tar.gz。
解壓縮： 將下載的壓縮包解壓縮到你喜歡的位置（例如，在 Windows 上解壓縮到 C:\zrok，在 macOS/Linux 上解壓縮到 ~/zrok）。
添加到 PATH (推薦)： 將 Zrok 可執行檔所在的目錄添加到你的系統 PATH 環境變數中，這樣你就可以在任何終端機中直接運行 zrok 命令。
2. 註冊 Zrok 帳戶並獲取 zrok.json
訪問 Zrok 網站： 打開瀏覽器，前往 https://zrok.io/。
註冊/登錄： 點擊 "Sign Up" 註冊一個新帳戶（可以使用 GitHub 登錄）。
建立環境 (Environment)： 登錄後，Zrok 會提示你建立一個新的環境。按照指示操作。
下載環境文件： 在成功建立環境後，Zrok 網站會顯示一個命令，類似於：
將這個命令複製並在你的終端機中執行。 這會在你的當前目錄下生成一個 zrok.json 文件。這個文件包含你的 Zrok 環境設定，請妥善保管，不要分享給他人。 這個 zrok.json 文件會自動被 Zrok CLI 工具檢測到。
程式碼說明與設定 (Python)
這份程式碼是基於你之前的版本，並重新整合了 Flask 和 LINE Bot 功能。
import tkinter as tk
from tkinter import ttk, scrolledtext, messagebox
import paho.mqtt.client as mqtt
import threading
import time
import os
import sys
from flask import Flask, request, abort
from linebot.v3.webhook import WebhookHandler
from linebot.v3.webhooks import MessageEvent, TextMessageContent
from linebot.v3.messaging import Configuration, ApiClient, MessagingApi, ReplyMessageRequest, TextMessage as LineTextMessage
from linebot.exceptions import InvalidSignatureError, LineBotApiError
MQTT_BROKER = "broker.mqttgo.io"
MQTT_PORT = 1883
MQTT_LED_CONTROL_TOPIC = "wokwi/esp32/led/control"
MQTT_TEMP_TOPIC = "wokwi/esp32/dht/temperature"
MQTT_HUMID_TOPIC = "wokwi/esp32/dht/humidity"
MQTT_STATUS_TOPIC = "wokwi/esp32/status"
LINE_CHANNEL_ACCESS_TOKEN = os.getenv('LINE_CHANNEL_ACCESS_TOKEN', 'YOUR_LINE_CHANNEL_ACCESS_TOKEN_HERE')
LINE_CHANNEL_SECRET = os.getenv('LINE_CHANNEL_SECRET', 'YOUR_LINE_CHANNEL_SECRET_HERE')
FLASK_PORT = 5000
latest_temperature = "N/A"
latest_humidity = "N/A"
app = Flask(__name__)
line_bot_configuration = Configuration(access_token=LINE_CHANNEL_ACCESS_TOKEN)
line_bot_api_client = ApiClient(line_bot_configuration)
line_bot_messaging_api = MessagingApi(line_bot_api_client)
handler = WebhookHandler(LINE_CHANNEL_SECRET)
tkinter_app_instance = None
class IoTApp:
def __init__(self, master):
self.master = master
master.title("Wokwi ESP32 IoT Control & Monitor (Tkinter with LINE Bot via Zrok)")
master.geometry("700x650")
self.mqtt_client = None
self.mqtt_reconnect_timer = None
self.create_widgets()
self.setup_mqtt()
self.line_webhook_thread = threading.Thread(target=self.start_flask_app, daemon=True)
self.line_webhook_thread.start()
global tkinter_app_instance
tkinter_app_instance = self
def create_widgets(self):
"""Creates all widgets in the Tkinter GUI."""
led_frame = ttk.LabelFrame(self.master, text="LED 控制", padding="10")
led_frame.pack(pady=10, padx=10, fill="x")
ttk.Button(led_frame, text="LED 開啟", command=lambda: self.publish_mqtt(MQTT_LED_CONTROL_TOPIC, "on")).pack(side="left", padx=5, pady=5)
ttk.Button(led_frame, text="LED 關閉", command=lambda: self.publish_mqtt(MQTT_LED_CONTROL_TOPIC, "off")).pack(side="left", padx=5, pady=5)
ttk.Button(led_frame, text="LED 閃爍", command=lambda: self.publish_mqtt(MQTT_LED_CONTROL_TOPIC, "flash")).pack(side="left", padx=5, pady=5)
ttk.Button(led_frame, text="LED 定時 (10秒)", 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 感測器數據", padding="10")
dht_frame.pack(pady=10, padx=10, fill="x")
self.temp_label = ttk.Label(dht_frame, text=f"溫度: {latest_temperature}°C", font=("Arial", 14))
self.temp_label.pack(pady=5, anchor="w")
self.humid_label = ttk.Label(dht_frame, text=f"濕度: {latest_humidity}%", font=("Arial", 14))
self.humid_label.pack(pady=5, anchor="w")
status_frame = ttk.LabelFrame(self.master, text="連接狀態", padding="10")
status_frame.pack(pady=10, padx=10, fill="x")
self.mqtt_status_label = ttk.Label(status_frame, text="MQTT: 連接中...", font=("Arial", 10), foreground="blue")
self.mqtt_status_label.pack(pady=2, anchor="w")
self.flask_status_label = ttk.Label(status_frame, text="Flask: 啟動中...", font=("Arial", 10), foreground="blue")
self.flask_status_label.pack(pady=2, anchor="w")
self.line_status_label = ttk.Label(status_frame, text="LINE Webhook: 等待 Zrok 連接...", font=("Arial", 10), foreground="blue")
self.line_status_label.pack(pady=2, anchor="w")
ttk.Label(status_frame, text=f"Flask 本地埠: {FLASK_PORT}", font=("Arial", 9)).pack(pady=1, anchor="w")
ttk.Label(status_frame, text="請啟動 Zrok 將此埠暴露到公共網路！", font=("Arial", 9), foreground="purple").pack(pady=1, anchor="w")
log_frame = ttk.LabelFrame(self.master, text="訊息日誌", padding="10")
log_frame.pack(pady=10, padx=10, fill="both", expand=True)
self.log_text = scrolledtext.ScrolledText(log_frame, width=80, height=15, wrap=tk.WORD, font=("Consolas", 10))
self.log_text.pack(expand=True, fill="both")
self.log_text.config(state=tk.DISABLED)
def log_message(self, message, tag=None):
"""Appends a message to the log text area."""
self.master.after(0, self._append_log_message, message, tag)
def _append_log_message(self, message, tag):
"""Internal function to append log message (thread-safe)."""
self.log_text.config(state=tk.NORMAL)
self.log_text.insert(tk.END, message + "\n", tag)
self.log_text.see(tk.END)
self.log_text.config(state=tk.DISABLED)
def setup_mqtt(self):
"""Sets up MQTT client callbacks and attempts to connect."""
self.mqtt_client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
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
self.attempt_mqtt_connection()
def attempt_mqtt_connection(self):
"""Attempts to connect to the MQTT broker."""
try:
self.update_mqtt_status("嘗試連接到 MQTT...", "blue")
self.log_message("MQTT: 嘗試連接到 Broker...", "info")
self.mqtt_client.connect(MQTT_BROKER, MQTT_PORT, 60)
if not hasattr(self, 'mqtt_thread') or not self.mqtt_thread.is_alive():
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"錯誤: {e}", "red")
self.log_message(f"MQTT: 初始連接錯誤: {e}", "error")
self.schedule_mqtt_reconnect()
def on_mqtt_connect(self, client, userdata, flags, rc, properties):
"""Callback function when MQTT connects successfully."""
if rc == 0:
self.update_mqtt_status("已連接到 MQTT Broker！", "green")
self.log_message("MQTT: 已連接到 Broker！", "success")
client.subscribe(MQTT_TEMP_TOPIC)
client.subscribe(MQTT_HUMID_TOPIC)
client.subscribe(MQTT_STATUS_TOPIC)
self.log_message(f"MQTT: 已訂閱 {MQTT_TEMP_TOPIC}, {MQTT_HUMID_TOPIC}, {MQTT_STATUS_TOPIC}", "info")
if self.mqtt_reconnect_timer:
self.master.after_cancel(self.mqtt_reconnect_timer)
self.mqtt_reconnect_timer = None
else:
self.update_mqtt_status(f"連接失敗 (代碼 {rc})。正在重試...", "orange")
self.log_message(f"MQTT: 連接失敗, 返回代碼 {rc}。正在重試...", "warning")
self.schedule_mqtt_reconnect()
def on_mqtt_disconnect(self, client, userdata, rc):
"""Callback function when MQTT disconnects."""
self.update_mqtt_status(f"已從 MQTT 斷開連接 (代碼 {rc})。正在重連...", "red")
self.log_message(f"MQTT: 已斷開連接, 結果代碼 {rc}。正在重連...", "error")
self.schedule_mqtt_reconnect()
def schedule_mqtt_reconnect(self):
"""Schedules an MQTT reconnection attempt after a delay."""
if self.mqtt_reconnect_timer is None:
self.mqtt_reconnect_timer = self.master.after(5000, self.perform_mqtt_reconnect)
self.log_message("MQTT: 已安排 5 秒後重連。", "info")
else:
self.log_message("MQTT: 重連已安排。", "info")
def perform_mqtt_reconnect(self):
"""Performs the actual MQTT reconnection attempt."""
self.mqtt_reconnect_timer = None
if not self.mqtt_client.is_connected():
self.log_message("MQTT: 嘗試重連...", "info")
try:
self.mqtt_client.reconnect()
except Exception as e:
self.update_mqtt_status(f"MQTT 重連失敗: {e}", "red")
self.log_message(f"MQTT: 重連失敗: {e}", "error")
self.schedule_mqtt_reconnect()
else:
self.log_message("MQTT: 已連接, 無需重連。", "info")
def on_mqtt_message(self, client, userdata, msg):
"""Callback function when an MQTT message is received, updates sensor data."""
global latest_temperature, latest_humidity
payload = msg.payload.decode()
self.log_message(f"MQTT Rx: 主題='{msg.topic}', 載荷='{payload}'", "mqtt_rx")
if msg.topic == MQTT_TEMP_TOPIC:
latest_temperature = payload
self.master.after(0, self.update_dht_labels)
elif msg.topic == MQTT_HUMID_TOPIC:
latest_humidity = payload
self.master.after(0, self.update_dht_labels)
elif msg.topic == MQTT_STATUS_TOPIC:
self.log_message(f"ESP32 狀態: {payload}", "esp32_status")
def update_dht_labels(self):
"""Updates temperature and humidity display on Tkinter GUI."""
self.temp_label.config(text=f"溫度: {latest_temperature}°C")
self.humid_label.config(text=f"濕度: {latest_humidity}%")
def update_mqtt_status(self, text, color):
"""Updates MQTT status display on Tkinter GUI."""
self.master.after(0, lambda: self.mqtt_status_label.config(text=f"MQTT: {text}", foreground=color))
def update_flask_status(self, text, color):
"""Updates Flask status display on Tkinter GUI."""
self.master.after(0, lambda: self.flask_status_label.config(text=f"Flask: {text}", foreground=color))
def update_line_status(self, text, color):
"""Updates LINE Bot status display on Tkinter GUI."""
self.master.after(0, lambda: self.line_status_label.config(text=f"LINE Webhook: {text}", foreground=color))
def publish_mqtt(self, topic, payload):
"""Publishes an MQTT message to the specified topic."""
if self.mqtt_client and self.mqtt_client.is_connected():
try:
self.mqtt_client.publish(topic, payload)
self.log_message(f"MQTT Tx: 主題='{topic}', 載荷='{payload}'", "mqtt_tx")
self.update_mqtt_status(f"已發送指令: {payload}", "blue")
except Exception as e:
self.update_mqtt_status(f"發布錯誤: {e}", "red")
self.log_message(f"MQTT: 發布錯誤: {e}", "error")
else:
self.update_mqtt_status("MQTT 未連接！指令未發送。", "red")
self.log_message("MQTT: 客戶端未連接，無法發布。指令未發送。", "error")
self.schedule_mqtt_reconnect()
def start_flask_app(self):
"""Starts the Flask web service to listen for LINE Webhooks."""
if LINE_CHANNEL_ACCESS_TOKEN == 'YOUR_LINE_CHANNEL_ACCESS_TOKEN_HERE' or \
LINE_CHANNEL_SECRET == 'YOUR_LINE_CHANNEL_SECRET_HERE':
messagebox.showerror("LINE Token/Secret 錯誤", "請設定你的 LINE_CHANNEL_ACCESS_TOKEN 和 LINE_CHANNEL_SECRET！")
self.update_line_status("錯誤: Token/Secret 缺失！", "red")
self.log_message("LINE: 錯誤 - Channel Access Token 或 Secret 缺失。", "error")
return
self.update_flask_status(f"正在啟動於埠 {FLASK_PORT}...", "blue")
self.log_message(f"Flask: 應用程式正在啟動於埠 {FLASK_PORT}...", "info")
try:
app.run(host='0.0.0.0', port=FLASK_PORT, debug=False, use_reloader=False)
self.update_flask_status(f"已退出於埠 {FLASK_PORT}", "grey")
self.log_message(f"Flask: 應用程式已停止運行於埠 {FLASK_PORT}。", "warning")
except Exception as e:
self.update_flask_status(f"錯誤: {e}", "red")
self.log_message(f"Flask: 應用程式啟動錯誤: {e}", "error")
messagebox.showerror("Flask 應用程式錯誤", f"無法啟動 Flask 應用程式用於 LINE: {e}\n"
f"請確保 {FLASK_PORT} 埠沒有被占用，並且手動啟動 Zrok。")
self.master.after(100, self.master.destroy)
os._exit(1)
def on_closing(self):
"""
Cleanup actions when the Tkinter window is closed.
Gracefully stops the MQTT client. Flask app, being a daemon thread, will automatically exit.
"""
self.log_message("應用程式: 正在關閉...", "info")
if self.mqtt_client:
self.log_message("MQTT: 正在斷開客戶端連接...", "info")
self.mqtt_client.disconnect()
if self.mqtt_reconnect_timer:
self.master.after_cancel(self.mqtt_reconnect_timer)
self.master.destroy()
os._exit(0)
@app.route("/callback", methods=['POST'])
def line_webhook_callback():
signature = request.headers.get('X-Line-Signature')
body = request.get_data(as_text=True)
if tkinter_app_instance:
tkinter_app_instance.log_message(f"LINE Webhook: 收到請求。主體長度: {len(body)}", "line_rx")
try:
if tkinter_app_instance:
tkinter_app_instance.update_line_status("正在處理 Webhook...", "blue")
handler.handle(body, signature)
if tkinter_app_instance:
tkinter_app_instance.update_line_status("Webhook 處理完成 OK", "green")
return 'OK'
except InvalidSignatureError:
error_msg = "LINE Webhook: 無效簽名。請檢查 token/secret。"
if tkinter_app_instance:
tkinter_app_instance.log_message(error_msg, "error")
tkinter_app_instance.update_line_status("無效簽名！", "red")
print(error_msg)
abort(400)
except LineBotApiError as e:
error_msg = f"LINE Webhook: 處理過程中發生 LineBotApiError: {e}"
if tkinter_app_instance:
tkinter_app_instance.log_message(error_msg, "error")
tkinter_app_instance.update_line_status(f"處理錯誤 (LINE API): {e}", "red")
print(error_msg)
abort(500)
except Exception as e:
error_msg = f"LINE Webhook: 一般處理錯誤: {e}"
if tkinter_app_instance:
tkinter_app_instance.log_message(error_msg, "error")
tkinter_app_instance.update_line_status(f"處理錯誤 (一般): {e}", "red")
print(error_msg)
abort(500)
@handler.add(MessageEvent, message=TextMessageContent)
def handle_line_message(event):
received_text = event.message.text.lower().strip()
user_id = event.source.user_id
if tkinter_app_instance:
tkinter_app_instance.log_message(f"LINE 訊息: 來自 {user_id}, 文本: '{received_text}'", "line_msg")
reply_text = "未知指令。請輸入 '開燈', '關燈', '閃爍', '計時', '溫度' 或 '濕度'。"
mqtt_command = None
if received_text == "開燈":
mqtt_command = "on"
reply_text = "LED 已開啟！"
elif received_text == "關燈":
mqtt_command = "off"
reply_text = "LED 已關閉！"
elif received_text == "閃爍":
mqtt_command = "flash"
reply_text = "LED 開始閃爍！"
elif received_text == "計時":
mqtt_command = "timer"
reply_text = "LED 已設定為 10 秒定時開關！"
elif received_text == "溫度":
reply_text = f"目前溫度為: {latest_temperature}°C"
elif received_text == "濕度":
reply_text = f"目前濕度為: {latest_humidity}%"
elif received_text == "狀態":
mqtt_status = "連線" if tkinter_app_instance and tkinter_app_instance.mqtt_client.is_connected() else "斷線"
flask_status_text = tkinter_app_instance.flask_status_label.cget("text").replace("Flask: ", "") if tkinter_app_instance else "未知"
line_status_text = tkinter_app_instance.line_status_label.cget("text").replace("LINE Webhook: ", "") if tkinter_app_instance else "未知"
reply_text = f"系統狀態:\nMQTT: {mqtt_status}\nFlask: {flask_status_text}\nLINE: {line_status_text}\n溫度: {latest_temperature}°C\n濕度: {latest_humidity}%"
if mqtt_command and tkinter_app_instance:
tkinter_app_instance.master.after(0, tkinter_app_instance.publish_mqtt, MQTT_LED_CONTROL_TOPIC, mqtt_command)
tkinter_app_instance.log_message(f"LINE 觸發 MQTT 發布: {mqtt_command}", "line_mqtt_tx")
try:
line_bot_messaging_api.reply_message_with_http_info(
ReplyMessageRequest(
reply_token=event.reply_token,
messages=[LineTextMessage(text=reply_text)]
)
)
if tkinter_app_instance:
tkinter_app_instance.log_message(f"LINE 回覆: '{reply_text}'", "line_reply")
tkinter_app_instance.update_line_status("訊息已處理並回覆。", "green")
except LineBotApiError as e:
error_msg = f"LINE: 回覆失敗 (LineBotApiError): {e.status_code} - {e.error.message}"
if tkinter_app_instance:
tkinter_app_instance.log_message(error_msg, "error")
tkinter_app_instance.update_line_status(f"回覆錯誤 (LINE API): {e.status_code}", "red")
print(error_msg)
except Exception as e:
error_msg = f"LINE: 回覆失敗 (一般錯誤): {e}"
if tkinter_app_instance:
tkinter_app_instance.log_message(error_msg, "error")
tkinter_app_instance.update_line_status(f"回覆錯誤 (一般): {e}", "red")
print(error_msg)
if __name__ == "__main__":
if LINE_CHANNEL_ACCESS_TOKEN == 'YOUR_LINE_CHANNEL_ACCESS_TOKEN_HERE' or \
LINE_CHANNEL_SECRET == 'YOUR_LINE_CHANNEL_SECRET_HERE':
print("錯誤: 請設定你的 LINE_CHANNEL_ACCESS_TOKEN 和 LINE_CHANNEL_SECRET。")
messagebox.showerror("LINE API 錯誤", "請設定你的 LINE_CHANNEL_ACCESS_TOKEN 和 LINE_CHANNEL_SECRET！")
sys.exit(1)
root = tk.Tk()
app_instance = IoTApp(root)
tkinter_app_instance = app_instance
app_instance.log_text.tag_config("info", foreground="blue")
app_instance.log_text.tag_config("success", foreground="green")
app_instance.log_text.tag_config("warning", foreground="orange")
app_instance.log_text.tag_config("error", foreground="red")
app_instance.log_text.tag_config("mqtt_tx", foreground="purple")
app_instance.log_text.tag_config("mqtt_rx", foreground="darkgreen")
app_instance.log_text.tag_config("line_rx", foreground="darkblue")
app_instance.log_text.tag_config("line_msg", foreground="magenta")
app_instance.log_text.tag_config("line_reply", foreground="darkcyan")
app_instance.log_text.tag_config("line_mqtt_tx", foreground="brown")
app_instance.log_text.tag_config("esp32_status", foreground="darkorange")
root.protocol("WM_DELETE_WINDOW", app_instance.on_closing)
root.mainloop()
操作步驟 (一步一步來)
步驟 1: 準備 Python 環境
安裝 Python (如果尚未安裝)： 從 python.org 下載並安裝最新版本的 Python。安裝時記得勾選 "Add Python to PATH"。
安裝必要的 Python 庫： 打開你的命令提示字元 (CMD) 或終端機，執行以下指令：
tkinter 是 Python 標準庫的一部分，通常無需額外安裝。
步驟 2: 設定 Zrok
下載並解壓縮 Zrok CLI (參考上面 "前置準備" 區塊)。
在終端機中，導航到你希望存放 zrok.json 的目錄。 (例如，你的專案目錄)。
執行 Zrok 啟用命令： 在你的 Zrok 網站帳戶頁面找到 zrok enable YOUR_UNIQUE_KEY > zrok.json 類似的命令，並在終端機中執行它。這會生成 zrok.json 文件。
3. 設定 LINE Developers Console
登入你的 LINE Developers Console (https://developers.line.biz/console/)。
選擇你的 Provider 和 Channel。
進入 "Messaging API" 分頁。
找到 "Channel Access Token (long-lived)" 和 "Channel Secret"。複製它們。
更新 Python 程式碼： 將複製的 Token 和 Secret 替換到 Python 程式碼中 LINE_CHANNEL_ACCESS_TOKEN 和 LINE_CHANNEL_SECRET 的位置，或者設置為環境變數。
強烈建議使用環境變數來保護這些敏感資訊。
4. 運行 Wokwi ESP32 模擬器
如果你還沒有，按照上面 "Arduino (Wokwi ESP32) 程式碼" 的說明在 Wokwi 上設置好 ESP32 專案。
重要： 確保 Wokwi 程式碼中的 mqtt_client_id 是獨特的。
啟動 Wokwi 模擬。 觀察 Serial Monitor 確保 Wi-Fi 和 MQTT 連接正常，並開始發送溫濕度數據。
5. 運行 Python GUI 應用程式 (包含 Flask)
打開一個新的終端機視窗。
導航到你保存 line_iot_app_with_gui.py 檔案的目錄。
執行指令：
觀察 Tkinter 視窗：
"MQTT" 狀態應該顯示 "已連接到 MQTT Broker！" (綠色)。
"Flask" 狀態應該顯示 "正在啟動於埠 5000..." (藍色)。
"LINE Webhook" 狀態應該顯示 "等待 Zrok 連接..." (藍色)。
6. 啟動 Zrok 並獲取公共 URL
打開一個全新的終端機視窗 (不要關閉運行 Python 程式的視窗)。
確保你現在的工作目錄是之前生成 zrok.json 文件的那個目錄。
執行 Zrok 命令來暴露你的本地 Flask 服務：
Zrok 會在終端機中顯示一個公共 URL，類似於：
複製這個 Public URL (以 https:// 開頭的)。
請勿關閉這個運行 Zrok 的終端機視窗。
在桌面上 執行 bat檔案 zrok.bat 檔案內容如下
c:
cd C:\Users\User\Downloads\zrok_1.0.6_windows_amd64
zrok disable
zrok enable gCegmLgx1H2V8vzrok share public http://localhost:5000
7. 完成 LINE Developers Console 設定
回到你的 LINE Developers Console -> Messaging API 分頁。
在 "Webhook URL" 欄位中，粘貼你從 Zrok 複製的 https:// 公共 URL。
在貼上的 URL 末尾，手動加上 /callback。
確保 "Use webhooks" 設置為 Enabled (啟用)。
點擊 "Verify" (驗證) 按鈕。
8. 觀察日誌與測試
Python Tkinter 視窗的日誌區塊：
通過 LINE App 測試：
這看起來是一個結合物聯網 (IoT)、即時通訊和開發測試的複雜系統。
核心概念與角色
在深入控制流程之前，先了解一下每個組件在這裡扮演的角色：
Zrok: 是一個開源的零信任 (Zero Trust) 網路工具，類似 ngrok，用於將本地服務安全地公開到網際網路。它能讓你在本地運行的程式被外部網路存取。
LINE: 台灣最主流的即時通訊軟體。在這裡，它通常作為使用者介面 (UI)，接收指令和發送通知。
LINE Developers (LINE Develop): LINE 官方為開發者提供的平台，用於創建 LINE Bot、設定 Webhook URL、管理頻道等。這是你的 LINE Bot 與外部世界互動的橋樑。
MQTT: 一種輕量級的訊息發布/訂閱協議，專為物聯網設備設計。它非常適合在不穩定網路或資源有限的設備之間傳輸少量數據。通常需要一個 MQTT Broker（訊息代理伺服器）來協調訊息傳輸。
Python: 一種功能強大的程式語言，通常用於開發 LINE Bot 的後端邏輯、處理 MQTT 訊息、控制硬體設備等。
Wokwi: 一個線上模擬器，主要用於模擬 Arduino、ESP32 等微控制器和各種電子元件。它能讓你無需實體硬體就能測試物聯網專案。
典型的控制流程情境
基於這些組件，我們可以設想一個常見的控制流程：使用者透過 LINE 發送指令，經由 LINE Bot 處理後，透過 MQTT 控制 Wokwi 上的模擬硬體，並可能將結果或狀態回傳給 LINE。
以下是兩種主要情境的詳細控制流程：
情境一：LINE 控制 Wokwi 模擬設備 (例如：開燈/關燈)
這個流程描述了使用者如何從 LINE 發送指令來控制一個模擬設備（例如 Wokwi 上的一顆 LED）。
控制流程步驟說明：
使用者發送指令 (LINE -> LINE Bot):
LINE Bot 接收訊息並轉發 (LINE Bot -> LINE Developers -> Python 後端):
LINE 伺服器接收到使用者訊息後，會將這個訊息作為一個 Webhook 事件 發送到你在 LINE Developers 平台上設定的 Webhook URL。
這個 Webhook URL 實際上指向你的 Python 後端程式。
Zrok 在這裡扮演關鍵角色： 你的 Python 後端程式可能正在本地電腦上運行 (例如 http://localhost:5000)。由於 LINE 伺服器無法直接存取你的本地電腦，你會使用 Zrok 建立一個安全的隧道，將 http://localhost:5000 公開為一個可以被 LINE 伺服器存取的公開 URL (例如 https://your-zrok-id.zrok.io)。這個公開 URL 就是你設定在 LINE Developers 上的 Webhook URL。
Python 後端處理指令 (Python):
Python 後端程式接收到 LINE 的 Webhook 事件。
程式解析使用者發送的指令 (例如，判斷是 "開燈" 還是 "關燈")。
根據指令，Python 程式會建構一個 MQTT 訊息，包含要控制的目標設備 (例如 "LED") 和其狀態 (例如 "ON" 或 "OFF")。
Python 發布 MQTT 訊息 (Python -> Zrok -> MQTT Broker):
Python 程式使用 MQTT 客戶端庫（如 paho-mqtt）連接到一個 MQTT Broker。
這個 MQTT Broker 可以是公開的（例如 HiveMQ Cloud, Mosquitto），也可以是你自架的。
如果 MQTT Broker 在本地運行或需要穿透防火牆： Zrok 再次發揮作用，可以將本地的 MQTT Broker (例如 tcp://localhost:1883) 暴露到網際網路上，或者你的 Python 程式可以直接連線到一個公開的 MQTT Broker。
MQTT Broker 轉發訊息 (MQTT Broker -> Zrok -> Wokwi):
Wokwi 模擬器接收並執行 (Wokwi):
Wokwi 回傳狀態 (可選，Wokwi -> Zrok -> MQTT Broker -> Python):
Wokwi 上的模擬設備在執行完指令後，可以選擇發布一個狀態更新訊息到 MQTT Broker（例如 "LED 狀態：已開"）。
這個流程與步驟 4-5 類似，只不過這次發布者是 Wokwi 模擬器。
Python 後端程式也訂閱這個狀態主題，接收並處理這些回傳訊息。
Python 回覆使用者 (Python -> LINE Developers -> LINE Bot -> LINE):
情境二：Wokwi 狀態報告到 LINE (例如：溫濕度監測)
這個流程描述了 Wokwi 模擬的感測器如何定時或在狀態改變時，將數據發送到 LINE。
控制流程步驟說明：
Wokwi 監測數據並發布 MQTT (Wokwi -> Zrok -> MQTT Broker):
在 Wokwi 模擬器中，你的 ESP32/Arduino 程式碼會模擬讀取感測器數據（例如溫度、濕度）。
程式碼會定期（或在數據變化達到閾值時）將這些感測器數據封裝成 MQTT 訊息。
Wokwi 上的模擬設備會作為 MQTT 客戶端，透過 Zrok 隧道連接到 MQTT Broker，並發布這些訊息。
MQTT Broker 轉發訊息 (MQTT Broker -> Python):
Python 後端接收並處理 (Python):
Python 發送 LINE 通知 (Python -> LINE Developers -> LINE Bot -> LINE):
Zrok 在此流程中的關鍵作用
你會發現 Zrok 在這兩個情境中都扮演著至關重要的橋樑角色：
LINE Webhook 連線： 將本地運行的 Python 後端程式暴露給 LINE 伺服器，以便接收 Webhook 事件。
MQTT Broker/Client 連線： 如果你的 MQTT Broker 運行在本地網路，或者 Wokwi 上的模擬器需要連接到一個外部無法直接存取的 MQTT Broker，Zrok 就能建立隧道，讓它們之間能夠通訊。
透過 Zrok，你無需配置複雜的網路穿透（如埠轉發），就能輕鬆實現本地開發環境與外部服務的互動，極大地簡化了 LINE Bot 和 IoT 專案的開發與測試過程。
