ESP32 雙核心控制 LED 與 DHT22 溫濕度感測器 (Wokwi 模擬) EX10 -- Python GUI , MQTT , Zrok , Line Messaging API
功能類似 ngrok 的軟體和服務，它們各有優缺點，你可以根據自己的需求來選擇：
1. 開源且可自架的方案 (Open Source & Self-Hostable)
如果你追求更高的控制權、安全性和透明度，或者有長期、大量的隧道需求，自架服務會是很好的選擇。
Tunnelmole
優點：完全開源（客戶端和伺服器端），提供免費的公共 URL，支援自架。對於看重透明度和控制的開發者來說是個理想的選擇。它提供了簡單的安裝方式，並且支援 Node.js。
缺點：如果自架，需要自己維護伺服器。
Cloudflare Tunnel (cloudflared)
優點：由 Cloudflare 提供，高度可靠且功能強大。免費版有基本功能，付費版提供更多進階特性，如自訂網域、更多協定支援、整合 Cloudflare 的安全功能 (Zero Trust)。設定相對簡單，且能穿透防火牆。
缺點：需要一個 Cloudflare 帳戶。免費版隧道的 URL 每次重新連線可能會變動，或有 24 小時的連線時間限制，若要自訂網域和長期穩定連線需要付費。
frp (Fast Reverse Proxy)
優點：這是一個功能非常強大的開源反向代理工具，支援 HTTP, HTTPS, TCP, UDP 等多種協定。你可以將它部署在自己的伺服器上，實現高度客製化和無限制的使用。
缺點：需要自己租用一台有公網 IP 的伺服器並進行配置，對新手來說學習曲線較高。
Localtunnel
優點：一個簡單易用的開源工具，提供免費的公共 URL。安裝和使用都非常直觀。
缺點：功能相對單一，可能不支援所有 ngrok 的進階功能，且公共伺服器的穩定性可能不如商業服務。
Serveo
優點：基於 SSH，無需安裝任何客戶端軟體，只需一個 SSH 命令即可建立隧道。非常方便快捷。
缺點：穩定性可能不如商業服務，且功能相對基礎。
sish
優點：另一個基於 SSH 的開源工具，可自架，支援 HTTP/HTTPS/WS/WSS/TCP 隧道。
缺點：同 Serveo，穩定性取決於自架的伺服器。
Piko
優點：專為生產環境設計的開源反向代理服務，適合需要長期穩定運行的場景，特別是 Kubernetes 環境。
缺點：更偏向於生產環境部署，對於一次性或簡單的本地開發測試可能略顯複雜。
Zrok
優點：一個開源且基於零信任網路 (OpenZiti) 的 ngrok 替代品，提供免費的 SaaS 版本和自架選項。強調安全和私密性，支援 HTTP/HTTPS、TCP/UDP 隧道和檔案分享。
缺點：可能對於快速測試來說功能有點過於豐富。
Zrok 是一個類似 Tunnelmole/ngrok 的內網穿透工具，它能將你的本地服務暴露到公共網路。
好的，這次我們將把網路對接工具從 Tunnelmole 替換成 Zrok，來實現 LINE Bot 和本地 Python 應用程式之間的通訊。其他功能，包括 Tkinter GUI 和 MQTT 與 ESP32 的互動，都將保留。
Zrok 是一個類似 Tunnelmole/ngrok 的內網穿透工具，它能將你的本地服務暴露到公共網路。
核心組件概述 (更新為 Zrok)
Wokwi ESP32 模擬器 (Arduino 程式碼)：
保持不變，負責 DHT22 感測器數據讀取、LED 控制以及與 MQTT Broker 通訊。
MQTT Broker (MQTT 伺服器)：
保持不變，作為 ESP32 和 Python GUI 應用程式之間數據交換的中心。
Python GUI 應用程式 (Python 程式碼)：
包含 Tkinter GUI 來顯示感測器數據和控制 LED。
新增 Flask Webhook，現在將通過 Zrok 暴露到公共網路，以接收來自 LINE 平台的訊息。
包含 LINE Bot SDK 來處理 LINE 訊息和發送回覆。
Zrok (內網穿透工具)：
取代 Tunnelmole，將你本地運行 Flask 應用的 5000 埠暴露到網際網路。
提供一個公共的 HTTPS URL，供 LINE Developers Console 配置。
前置準備：安裝和設定 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命令。Windows: 搜尋 "環境變數"，編輯 "Path" 系統變數，添加 Zrok 可執行檔所在的路徑。
macOS/Linux: 將
zrok可執行檔移動到/usr/local/bin或在你~/.bashrc或~/.zshrc中添加路徑。
2. 註冊 Zrok 帳戶並獲取 zrok.json
訪問 Zrok 網站： 打開瀏覽器，前往
https://zrok.io/ 註冊/登錄： 點擊 "Sign Up" 註冊一個新帳戶（可以使用 GitHub 登錄）。
建立環境 (Environment)： 登錄後，Zrok 會提示你建立一個新的環境。按照指示操作。
下載環境文件： 在成功建立環境後，Zrok 網站會顯示一個命令，類似於：
Bashzrok enable YOUR_UNIQUE_KEY > zrok.json將這個命令複製並在你的終端機中執行。 這會在你的當前目錄下生成一個
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()
好的，這次我們將把網路對接工具從 Tunnelmole 替換成 Zrok，來實現 LINE Bot 和本地 Python 應用程式之間的通訊。其他功能，包括 Tkinter GUI 和 MQTT 與 ESP32 的互動，都將保留。
Zrok 是一個類似 Tunnelmole/ngrok 的內網穿透工具，它能將你的本地服務暴露到公共網路。
核心組件概述 (更新為 Zrok)
Wokwi ESP32 模擬器 (Arduino 程式碼)：
保持不變，負責 DHT22 感測器數據讀取、LED 控制以及與 MQTT Broker 通訊。
MQTT Broker (MQTT 伺服器)：
保持不變，作為 ESP32 和 Python GUI 應用程式之間數據交換的中心。
Python GUI 應用程式 (Python 程式碼)：
包含 Tkinter GUI 來顯示感測器數據和控制 LED。
新增 Flask Webhook，現在將通過 Zrok 暴露到公共網路，以接收來自 LINE 平台的訊息。
包含 LINE Bot SDK 來處理 LINE 訊息和發送回覆。
Zrok (內網穿透工具)：
取代 Tunnelmole，將你本地運行 Flask 應用的 5000 埠暴露到網際網路。
提供一個公共的 HTTPS URL，供 LINE Developers Console 配置。
前置準備：安裝和設定 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命令。Windows: 搜尋 "環境變數"，編輯 "Path" 系統變數，添加 Zrok 可執行檔所在的路徑。
macOS/Linux: 將
zrok可執行檔移動到/usr/local/bin或在你~/.bashrc或~/.zshrc中添加路徑。
2. 註冊 Zrok 帳戶並獲取 zrok.json
訪問 Zrok 網站： 打開瀏覽器，前往
https://zrok.io/ 註冊/登錄： 點擊 "Sign Up" 註冊一個新帳戶（可以使用 GitHub 登錄）。
建立環境 (Environment)： 登錄後，Zrok 會提示你建立一個新的環境。按照指示操作。
下載環境文件： 在成功建立環境後，Zrok 網站會顯示一個命令，類似於：
Bashzrok enable YOUR_UNIQUE_KEY > zrok.json將這個命令複製並在你的終端機中執行。 這會在你的當前目錄下生成一個
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
# 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', 'YOUR_LINE_CHANNEL_ACCESS_TOKEN_HERE')
LINE_CHANNEL_SECRET = os.getenv('LINE_CHANNEL_SECRET', 'YOUR_LINE_CHANNEL_SECRET_HERE')
# 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()
操作步驟 (一步一步來)
步驟 1: 準備 Python 環境
安裝 Python (如果尚未安裝)： 從
python.org 安裝必要的 Python 庫： 打開你的命令提示字元 (CMD) 或終端機，執行以下指令：
Bashpip install paho-mqtt flask line-bot-sdk==3.7.0 # 或者更新版本，但確保是 v3.xtkinter是 Python 標準庫的一部分，通常無需額外安裝。
步驟 2: 設定 Zrok
下載並解壓縮 Zrok CLI (參考上面 "前置準備" 區塊)。
在終端機中，導航到你希望存放
zrok.json的目錄。 (例如，你的專案目錄)。執行 Zrok 啟用命令： 在你的 Zrok 網站帳戶頁面找到
zrok enable YOUR_UNIQUE_KEY > zrok.json類似的命令，並在終端機中執行它。這會生成zrok.json文件。確保
zrok.json文件與你運行 Python 應用程式的目錄相同或在 Zrok 工具的搜尋路徑中。
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的位置，或者設置為環境變數。PythonLINE_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')強烈建議使用環境變數來保護這些敏感資訊。
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檔案的目錄。執行指令：
Bashpython line_iot_app_with_gui.py觀察 Tkinter 視窗：
"MQTT" 狀態應該顯示 "已連接到 MQTT Broker！" (綠色)。
"Flask" 狀態應該顯示 "正在啟動於埠 5000..." (藍色)。
"LINE Webhook" 狀態應該顯示 "等待 Zrok 連接..." (藍色)。
6. 啟動 Zrok 並獲取公共 URL
打開一個全新的終端機視窗 (不要關閉運行 Python 程式的視窗)。
確保你現在的工作目錄是之前生成
zrok.json文件的那個目錄。執行 Zrok 命令來暴露你的本地 Flask 服務：
Bashzrok share public http://localhost:5000public表示你會得到一個公共可訪問的 URL。http://localhost:5000是你的 Flask 應用程式正在監聽的地址和埠。
Zrok 會在終端機中顯示一個公共 URL，類似於：
Success! Your share is now active. Public URL: https://some-random-string.share.zrok.io複製這個
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。正確範例：
https://some-random-string.share.zrok.io/callback
確保 "Use webhooks" 設置為 Enabled (啟用)。
點擊 "Verify" (驗證) 按鈕。
8. 觀察日誌與測試
LINE Developers Console "Verify" 結果：
如果顯示 "Success"：表示 LINE 平台能夠成功連接到你的 Flask 應用並收到 200 OK 回應。
如果顯示 "404 Not Found"：最常見原因是 Webhook URL 末尾沒有加
/callback，或者 Zrok 的 URL 有問題（例如 Zrok 共享服務已停止）。如果顯示 "Timeout occurred..."：Zrok 可能沒有正常運行，或者你的防火牆阻止了 Zrok 的連接。
如果顯示 "500 Internal Server Error"：請求到達了，但 Python 程式碼在處理時出錯。查看 Python 終端機和 Tkinter 日誌。
注意：首次驗證時，可能會出現
InvalidSignatureError，這是正常的，因為驗證請求沒有正確的 LINE 簽名。只要後續正常訊息能處理就行。
Python Tkinter 視窗的日誌區塊：
當你點擊 LINE Developers Console 中的 "Verify" 時，你應該在 Tkinter 視窗的日誌區塊中看到類似 "LINE Webhook: 收到請求。" 的訊息。
"LINE Webhook" 的狀態應該會更新。
通過 LINE App 測試：
打開你的 LINE App，向你的 Bot 發送訊息（例如 "開燈"、"關燈"、"溫度"、"濕度"、"狀態"）。
觀察 Wokwi 模擬器中的 LED 是否響應。
觀察 Python Tkinter 應用程式的日誌區塊和 LINE Bot 的回覆。
這看起來是一個結合物聯網 (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）。
graph TD
A[使用者在LINE應用程式] -->|發送控制指令| B(LINE Bot)
B -->|透過LINE Developers的Webhook URL| C(LINE Bot 後端程式 - Python)
C -->|處理指令 (e.g., "開燈", "關燈")| D(Python程式發布MQTT訊息)
D -->|透過Zrok隧道傳送MQTT訊息| E(外部可存取的MQTT Broker)
E -->|MQTT Broker接收並轉發訊息| F(Wokwi模擬器上的MQTT客戶端 - e.g., ESP32/Arduino)
F -->|根據MQTT訊息控制模擬硬體| G(Wokwi模擬的LED或其他執行器)
G -->|硬體狀態變化 (可選)| H(Wokwi客戶端發布狀態MQTT訊息)
H -->|透過Zrok隧道傳送狀態訊息| E
E -->|MQTT Broker轉發狀態訊息| I(Python程式訂閱MQTT狀態訊息)
I -->|處理狀態並生成回覆訊息| J(Python程式發送LINE回覆訊息)
J -->|透過LINE Developers API| B
B -->|顯示回覆訊息| A
控制流程步驟說明：
使用者發送指令 (LINE -> LINE Bot):
使用者在手機上的 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):
MQTT Broker 收到 Python 發布的訊息後，會將此訊息轉發給所有訂閱了相關主題 (Topic) 的客戶端。
Wokwi 模擬器接收並執行 (Wokwi):
在 Wokwi 模擬器中，你的 ESP32/Arduino 程式碼 會作為一個 MQTT 客戶端，訂閱特定的 MQTT 主題。
當 Wokwi 上的模擬設備收到來自 MQTT Broker 的訊息時，其程式碼會解析訊息，並執行對應的操作，例如改變 LED 的狀態（從 OFF 到 ON，或從 ON 到 OFF）。
Wokwi 回傳狀態 (可選，Wokwi -> Zrok -> MQTT Broker -> Python):
Wokwi 上的模擬設備在執行完指令後，可以選擇發布一個狀態更新訊息到 MQTT Broker（例如 "LED 狀態：已開"）。
這個流程與步驟 4-5 類似，只不過這次發布者是 Wokwi 模擬器。
Python 後端程式也訂閱這個狀態主題，接收並處理這些回傳訊息。
Python 回覆使用者 (Python -> LINE Developers -> LINE Bot -> LINE):
Python 後端程式在成功執行指令或接收到設備回傳的狀態後，會透過 LINE Developers 提供的 Messaging API，向使用者發送一個回覆訊息（例如 "燈已開啟"）。
使用者在 LINE 應用程式中看到回覆。
情境二：Wokwi 狀態報告到 LINE (例如：溫濕度監測)
這個流程描述了 Wokwi 模擬的感測器如何定時或在狀態改變時，將數據發送到 LINE。
graph TD
A[Wokwi模擬器上的感測器數據] -->|Wokwi客戶端程式讀取數據| B(Wokwi模擬器上的MQTT客戶端 - e.g., ESP32/Arduino)
B -->|根據數據變化或定時發布MQTT訊息| C(Wokwi客戶端發布MQTT狀態訊息)
C -->|透過Zrok隧道傳送MQTT訊息| D(外部可存取的MQTT Broker)
D -->|MQTT Broker轉發狀態訊息| E(Python程式訂閱MQTT狀態訊息)
E -->|處理狀態數據 (e.g., 溫度、濕度)| F(Python程式發送LINE通知訊息)
F -->|透過LINE Developers API| G(LINE Bot)
G -->|顯示通知訊息| H[使用者在LINE應用程式]
控制流程步驟說明：
Wokwi 監測數據並發布 MQTT (Wokwi -> Zrok -> MQTT Broker):
在 Wokwi 模擬器中，你的 ESP32/Arduino 程式碼會模擬讀取感測器數據（例如溫度、濕度）。
程式碼會定期（或在數據變化達到閾值時）將這些感測器數據封裝成 MQTT 訊息。
Wokwi 上的模擬設備會作為 MQTT 客戶端，透過 Zrok 隧道連接到 MQTT Broker，並發布這些訊息。
MQTT Broker 轉發訊息 (MQTT Broker -> Python):
MQTT Broker 收到 Wokwi 發布的訊息後，轉發給所有訂閱了相關主題的客戶端。
Python 後端接收並處理 (Python):
你的 Python 後端程式作為另一個 MQTT 客戶端，訂閱了 Wokwi 發布的感測器數據主題。
Python 程式接收並解析這些感測器數據。
Python 發送 LINE 通知 (Python -> LINE Developers -> LINE Bot -> LINE):
根據接收到的感測器數據，Python 程式可以決定是否需要向使用者發送 LINE 通知。例如，如果溫度超過某個閾值，就發送一條警告訊息。
Python 程式透過 LINE Developers 的 Messaging API 發送通知訊息。
使用者在 LINE 應用程式中收到通知。
Zrok 在此流程中的關鍵作用
你會發現 Zrok 在這兩個情境中都扮演著至關重要的橋樑角色：
LINE Webhook 連線： 將本地運行的 Python 後端程式暴露給 LINE 伺服器，以便接收 Webhook 事件。
MQTT Broker/Client 連線： 如果你的 MQTT Broker 運行在本地網路，或者 Wokwi 上的模擬器需要連接到一個外部無法直接存取的 MQTT Broker，Zrok 就能建立隧道，讓它們之間能夠通訊。
透過 Zrok，你無需配置複雜的網路穿透（如埠轉發），就能輕鬆實現本地開發環境與外部服務的互動，極大地簡化了 LINE Bot 和 IoT 專案的開發與測試過程。
沒有留言:
張貼留言