全台空氣品質智慧雙向監控系統 (Python & Node-RED IoT Solution)
import tkinter as tk
from tkinter import ttk
import requests
import json
import random
import threading
import time
from urllib.parse import quote
import paho.mqtt.client as mqtt
# 關閉 SSL verify=False 時產生的 InsecureRequestWarning 警告
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# --- 全局設定 ---
API_KEY = "aa1bc2bf-24c2-4c55-980f-a12075c2caec"
DEFAULT_SITE = "基隆"
MQTT_BROKER = "broker.hivemq.com"
MQTT_PORT = 1883
SUB_TOPIC = "alex9ufo/sitename" # 1. 訂閱來源：接收 Node-RED 傳來的測站名稱
PUB_TOPIC = "alex9ufo/weather" # 發行來源：傳送完整 JSON 資料給 Node-RED
UPDATE_INTERVAL = 900 # 15 分鐘 (秒)
# 全台 84 個觀測站清單資料 (測站名稱, 縣市)
ALL_SITES_DATA = [
("基隆", "基隆市"), ("汐止", "新北市"), ("新店", "新北市"), ("土城", "新北市"),
("板橋", "新北市"), ("新莊", "新北市"), ("菜寮", "新北市"), ("林口", "新北市"),
("淡水", "新北市"), ("士林", "臺北市"), ("中山", "臺北市"), ("萬華", "臺北市"),
("古亭", "臺北市"), ("松山", "臺北市"), ("大同", "臺北市"), ("桃園", "桃園市"),
("大園", "桃園市"), ("觀音", "桃園市"), ("平鎮", "桃園市"), ("龍潭", "桃園市"),
("湖口", "新竹縣"), ("竹東", "新竹縣"), ("新竹", "新竹市"), ("頭份", "苗栗縣"),
("苗栗", "苗栗縣"), ("三義", "苗栗縣"), ("豐原", "臺中市"), ("沙鹿", "臺中市"),
("大里", "臺中市"), ("忠明", "臺中市"), ("西屯", "臺中市"), ("彰化", "彰化縣"),
("線西", "彰化縣"), ("二林", "彰化縣"), ("南投", "南投縣"), ("斗六", "雲林縣"),
("崙背", "雲林縣"), ("新港", "嘉義縣"), ("朴子", "嘉義縣"), ("臺西", "雲林縣"),
("嘉義", "嘉義市"), ("新營", "臺南市"), ("善化", "臺南市"), ("安南", "臺南市"),
("臺南", "臺南市"), ("美濃", "高雄市"), ("橋頭", "高雄市"), ("仁武", "高雄市"),
("鳳山", "高雄市"), ("大寮", "高雄市"), ("林園", "高雄市"), ("楠梓", "高雄市"),
("左營", "高雄市"), ("前金", "高雄市"), ("前鎮", "高雄市"), ("小港", "高雄市"),
("屏東", "屏東縣"), ("潮州", "屏東縣"), ("恆春", "屏東縣"), ("臺東", "臺東縣"),
("花蓮", "花蓮縣"), ("陽明", "臺北市"), ("宜蘭", "宜蘭縣"), ("冬山", "宜蘭縣"),
("三重", "新北市"), ("中壢", "桃園市"), ("竹山", "南投縣"), ("永和", "新北市"),
("復興", "高雄市"), ("埔里", "南投縣"), ("馬祖", "連江縣"), ("金門", "金門縣"),
("馬公", "澎湖縣"), ("關山", "臺東縣"), ("麥寮", "雲林縣"), ("富貴角", "新北市"),
("大城", "彰化縣"), ("臺灣大道", "臺中市"), ("員林", "彰化縣"), ("林森", "臺南市"),
("南投（鹿谷）", "南投縣"), ("屏東（琉球）", "屏東縣"), ("新北（樹林）", "新北市"), ("屏東（枋山）", "屏東縣")
]
# 建立 站名 -> 縣市 快速查詢字典
SITE_COUNTY_MAP = {site: county for site, county in ALL_SITES_DATA}
# 欄位中文名稱對照表
FIELD_NAMES = {
"sitename": "測站名稱",
"county": "縣市",
"aqi": "AQI 指標",
"pollutant": "主要污染物",
"status": "狀態",
"so2": "二氧化硫 (SO2 ppb)",
"co": "一氧化碳 (CO ppm)",
"o3": "臭氧 (O3 ppb)",
"o3_8hr": "臭氧8小時 (O3 8hr ppb)",
"pm10": "懸浮微粒 (PM10 ug/m3)",
"pm2.5": "細懸浮微粒 (PM2.5 ug/m3)",
"no2": "二氧化氮 (NO2 ppb)",
"nox": "氮氧化物 (NOx ppb)",
"no": "一氧化氮 (NO ppb)",
"wind_speed": "風速 (m/s)",
"wind_direc": "風向 (deg)",
"publishtime": "發布時間",
"co_8hr": "一氧化碳8小時 (CO 8hr)",
"pm2.5_avg": "PM2.5 移動平均",
"pm10_avg": "PM10 移動平均",
"so2_avg": "SO2 移動平均",
"longitude": "經度",
"latitude": "緯度",
"siteid": "測站代碼"
}
class AirQualityApp:
def __init__(self, root):
self.root = root
self.root.title("Python AQI 監測站 (下拉選單 & MQTT 雙控版)")
self.root.geometry("620x720")
self.root.resizable(True, True)
self.target_site = DEFAULT_SITE
self.last_update_time = 0
# UI 佈局
self.create_widgets()
# 初始化 MQTT Client
self.client_id = f"Python-AQI-{random.randint(0, 0xFFFF):04X}"
self.mqtt_client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=self.client_id)
self.mqtt_client.on_connect = self.on_mqtt_connect
self.mqtt_client.on_message = self.on_mqtt_message
# 背景線程啟動 MQTT
threading.Thread(target=self.start_mqtt, daemon=True).start()
# 啟動定時檢查更新迴圈
self.root.after(1000, self.timer_loop)
# 啟動後立即抓取預設測站資料
threading.Thread(target=self.fetch_air_data, daemon=True).start()
def create_widgets(self):
# 1. 頂部控制區 (需求一: Python 下拉式選單選擇 sitename)
control_frame = ttk.LabelFrame(self.root, text="測站選擇與控制", padding=10)
control_frame.pack(fill="x", padx=10, pady=5)
ttk.Label(control_frame, text="選擇觀測站:").grid(row=0, column=0, sticky="w", padx=2)
self.dropdown_options = [f"{site} ({county})" for site, county in ALL_SITES_DATA]
self.site_combo = ttk.Combobox(control_frame, values=self.dropdown_options, width=22, state="readonly")
# 預設定位至基隆
default_index = 0
for idx, (site, _) in enumerate(ALL_SITES_DATA):
if site == DEFAULT_SITE:
default_index = idx
break
self.site_combo.current(default_index)
self.site_combo.grid(row=0, column=1, padx=5)
self.site_combo.bind("<<ComboboxSelected>>", self.on_combo_selected)
btn_update = ttk.Button(control_frame, text="強制更新", command=self.on_btn_manual_update)
btn_update.grid(row=0, column=2, padx=5)
# 2. 模擬 20x4 LCD 螢幕區 (深綠底、亮綠字)
lcd_frame = tk.LabelFrame(self.root, text="模擬 20x4 LCD 螢幕", bg="#1b3022", fg="white", padx=10, pady=5)
lcd_frame.pack(fill="x", padx=10, pady=5)
self.lcd_lines = []
for i in range(4):
lbl = tk.Label(
lcd_frame,
text="",
font=("Courier New", 14, "bold"),
bg="#1b3022",
fg="#33ff33",
anchor="w"
)
lbl.pack(fill="x", pady=1)
self.lcd_lines.append(lbl)
self.update_lcd_display("AQI: ---", "PM2.5: --- ug/m3", "Connecting...", "Time: --:--:--")
# 3. 分頁展示全資料
notebook = ttk.Notebook(self.root)
notebook.pack(fill="both", expand=True, padx=10, pady=5)
# 頁籤 1: 表格視圖
tab_table = ttk.Frame(notebook)
notebook.add(tab_table, text="所有數據明細")
columns = ("key", "name", "val")
self.tree = ttk.Treeview(tab_table, columns=columns, show="headings")
self.tree.heading("key", text="英文欄位")
self.tree.heading("name", text="中文說明")
self.tree.heading("val", text="數據內容")
self.tree.column("key", width=120)
self.tree.column("name", width=180)
self.tree.column("val", width=220)
scrollbar = ttk.Scrollbar(tab_table, orient="vertical", command=self.tree.yview)
self.tree.configure(yscrollcommand=scrollbar.set)
self.tree.pack(side="left", fill="both", expand=True)
scrollbar.pack(side="right", fill="y")
# 頁籤 2: RAW JSON 視圖
tab_json = ttk.Frame(notebook)
notebook.add(tab_json, text="RAW JSON")
self.json_text = tk.Text(tab_json, font=("Consolas", 10), wrap="word")
json_scroll = ttk.Scrollbar(tab_json, orient="vertical", command=self.json_text.yview)
self.json_text.configure(yscrollcommand=json_scroll.set)
self.json_text.pack(side="left", fill="both", expand=True)
json_scroll.pack(side="right", fill="y")
# 4. 警報 LED 狀態與通訊狀態列
status_frame = ttk.Frame(self.root, padding=10)
status_frame.pack(fill="x", padx=10, pady=2)
ttk.Label(status_frame, text="警報 LED 狀態:").pack(side="left")
self.led_canvas = tk.Canvas(status_frame, width=20, height=20, highlightthickness=0)
self.led_canvas.pack(side="left", padx=5)
self.led_light = self.led_canvas.create_oval(2, 2, 18, 18, fill="gray", outline="black")
self.lbl_mqtt_status = ttk.Label(status_frame, text="MQTT: 連線中...", foreground="orange")
self.lbl_mqtt_status.pack(side="right")
def update_lcd_display(self, line0, line1, line2, line3):
lines = [line0, line1, line2, line3]
for i in range(4):
text_str = lines[i][:20]
self.lcd_lines[i].config(text=text_str.ljust(20))
def set_led_status(self, state):
color_map = {
"bad": "#FF0000", # 亮紅色
"ok": "#00FF00", # 亮綠色
"nodata": "#888888" # 灰色
}
fill_color = color_map.get(state, "#888888")
self.led_canvas.itemconfig(self.led_light, fill=fill_color)
# --- MQTT 處理邏輯 (需求二: 從 alex9ufo/sitename 接收 sitename) ---
def start_mqtt(self):
try:
self.mqtt_client.connect(MQTT_BROKER, MQTT_PORT, 60)
self.mqtt_client.loop_forever()
except Exception as e:
print(f"[MQTT Error] {e}")
def on_mqtt_connect(self, client, userdata, flags, rc, properties=None):
print(f"[MQTT] 已成功連線至 Broker: {MQTT_BROKER}")
self.root.after(0, lambda: self.lbl_mqtt_status.config(text="MQTT: 已連線", foreground="green"))
client.subscribe(SUB_TOPIC)
print(f"[MQTT Sub] 已成功訂閱主題: {SUB_TOPIC}")
def on_mqtt_message(self, client, userdata, msg):
try:
# 取得從 Node-RED 傳來的測站名稱
rx_site = msg.payload.decode('utf-8').strip()
print(f"\n[MQTT Rx] 收到 Topic: '{msg.topic}' | Payload 測站: '{rx_site}'")
if rx_site and rx_site != self.target_site:
self.target_site = rx_site
# 自動尋找並同步切換 UI 下拉選單的顯示選項
for idx, (site, _) in enumerate(ALL_SITES_DATA):
if site == self.target_site:
self.root.after(0, lambda i=idx: self.site_combo.current(i))
break
print(f"[系統連動] 測站已切換為 '{self.target_site}'，準備發送 API 請求...")
threading.Thread(target=self.fetch_air_data, daemon=True).start()
except Exception as e:
print(f"[MQTT Parse Error] {e}")
# --- API 與資料處理邏輯 ---
def fetch_air_data(self):
# 將選取的 target_site 進行 URL 轉碼並替換至 API 網址
encoded_site = quote(self.target_site)
url = f"https://data.moenv.gov.tw/api/v2/aqx_p_432?limit=1&api_key={API_KEY}&filters=sitename,eq,{encoded_site}"
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"
}
print(f"\n[HTTP GET] 請求網址: {url}")
try:
resp = requests.get(url, headers=headers, verify=False, timeout=10)
if resp.status_code == 200:
data = resp.json()
# 相容性解析：處理 List 或 Dict 回傳格式
records = []
if isinstance(data, list):
records = data
elif isinstance(data, dict):
records = data.get("records", [])
if len(records) > 0:
print(f"[API 成功] 成功抓取到 '{self.target_site}' 的真實空氣品質資料！")
self.process_data(records[0])
else:
print(f"[警告] API 回傳成功但找不到測站 '{self.target_site}' 資料，顯示預設 --- 數據...")
self.load_fallback_data()
else:
print(f"[HTTP 錯誤] 代碼: {resp.status_code}，訊息: {resp.text}")
self.load_fallback_data()
except Exception as e:
print(f"[HTTP 請求失敗] 錯誤原因: {e}")
self.load_fallback_data()
self.last_update_time = time.time()
def process_data(self, record):
sitename = record.get("sitename", self.target_site)
county = record.get("county") or SITE_COUNTY_MAP.get(sitename, "未知")
record["county"] = county
# 判斷資料是否有效
has_aqi = bool(record.get("aqi") and record.get("aqi") != "---")
has_pm25 = bool(record.get("pm2.5") and record.get("pm2.5") != "---")
aqi_val = int(record.get("aqi")) if has_aqi else 0
pm25_val = float(record.get("pm2.5")) if has_pm25 else 0.0
status = record.get("status", "No Data")
pubtime = record.get("publishtime", "--:--:--")
time_str = pubtime[11:] if len(pubtime) >= 19 else pubtime
# 計算警報燈號
is_bad = (has_aqi and aqi_val > 50) or (has_pm25 and pm25_val > 15.4)
is_valid = has_aqi or has_pm25
if not is_valid:
led_state = "nodata"
elif is_bad:
led_state = "bad" # 紅燈
else:
led_state = "ok" # 綠燈
# 1. 控制台 (Console) 列印
print("================== [所有資料內容] ==================")
for k, v in record.items():
cn_name = FIELD_NAMES.get(k, k)
print(f"{k:<12} ({cn_name:<16}): {v if v != '' else '---'}")
print("===================================================")
# 2. UI - LCD 與 LED 更新
eng_status = self.get_english_status(status) if is_valid else "No Data"
line0 = f"AQI: {aqi_val if has_aqi else '---':<3} ({eng_status})"
line1 = f"PM2.5: {f'{pm25_val:.1f}' if has_pm25 else '---'} ug/m3"
line2 = "Air Bad!" if is_bad else ("Air OK" if is_valid else "No Data!")
line3 = f"Time: {time_str}"
self.root.after(0, lambda: self.update_lcd_display(line0, line1, line2, line3))
self.root.after(0, lambda: self.set_led_status(led_state))
# 3. UI - 更新 Treeview 表格
def update_tree():
for item in self.tree.get_children():
self.tree.delete(item)
for k, v in record.items():
cn_name = FIELD_NAMES.get(k, k)
self.tree.insert("", "end", values=(k, cn_name, v if (v != "" and v is not None) else "---"))
self.root.after(0, update_tree)
# 4. UI - 更新 RAW JSON 頁籤
def update_json_view():
self.json_text.delete("1.0", tk.END)
self.json_text.insert(tk.END, json.dumps(record, indent=4, ensure_ascii=False))
self.root.after(0, update_json_view)
# 5. 發行完整 JSON 數據至 MQTT (Topic: alex9ufo/weather)
if self.mqtt_client.is_connected():
payload_str = json.dumps(record, ensure_ascii=False)
self.mqtt_client.publish(PUB_TOPIC, payload_str)
print(f"[MQTT Tx] Published all data to topic '{PUB_TOPIC}' successfully.")
def load_fallback_data(self):
correct_county = SITE_COUNTY_MAP.get(self.target_site, "未知")
empty_record = {
"sitename": self.target_site,
"county": correct_county,
"aqi": "---",
"pollutant": "---",
"status": "---",
"so2": "---",
"co": "---",
"o3": "---",
"o3_8hr": "---",
"pm10": "---",
"pm2.5": "---",
"no2": "---",
"nox": "---",
"no": "---",
"wind_speed": "---",
"wind_direc": "---",
"publishtime": "--:--:--",
"co_8hr": "---",
"pm2.5_avg": "---",
"pm10_avg": "---",
"so2_avg": "---",
"longitude": "---",
"latitude": "---",
"siteid": "---"
}
self.process_data(empty_record)
def get_english_status(self, status):
status_map = {
"良好": "Good",
"普通": "Moderate",
"對敏感族群不健康": "Unhealthy*",
"對所有族群不健康": "Unhealthy",
"非常不健康": "V.Unhealthy",
"危害": "Hazardous"
}
return status_map.get(status, status)
# --- UI 事件處理 (需求一: 下拉選單選擇處點擊觸發) ---
def on_combo_selected(self, event):
selected_text = self.site_combo.get()
new_site = selected_text.split(" ")[0] # 解析提取測站名稱 (如從 "基隆 (基隆市)" 提取 "基隆")
if new_site and new_site != self.target_site:
self.target_site = new_site
print(f"\n[Dropdown Selected] 使用者從下拉選單切換至測站: {self.target_site}")
threading.Thread(target=self.fetch_air_data, daemon=True).start()
def on_btn_manual_update(self):
print("\n[Manual Trigger] 使用者點擊強制更新按鈕！")
threading.Thread(target=self.fetch_air_data, daemon=True).start()
def timer_loop(self):
if time.time() - self.last_update_time >= UPDATE_INTERVAL and self.last_update_time > 0:
print("\n[Auto Trigger] 定時 15 分鐘已到，自動抓取最新資料...")
threading.Thread(target=self.fetch_air_data, daemon=True).start()
self.root.after(1000, self.timer_loop)
if __name__ == "__main__":
root = tk.Tk()
app = AirQualityApp(root)
root.mainloop()
這支 Python 程式是一個基於 Tkinter 圖形化介面 (GUI) 與 MQTT 雙向通訊機制 的桌面空氣品質監測應用程式。它能夠向環境部 API 抓取真實空氣品質數據，並將資料模擬為硬體（LCD 螢幕與警報 LED），同時透過 MQTT 與 Node-RED 或 ESP32 開發板進行同步連動。
以下為您詳細說明該程式的核心架構與各項功能：
一、核心架構與元件說明
1. 多執行緒與連線架構 (Multi-threading)
背景 MQTT 執行緒：將
mqtt_client.loop_forever()放進獨立的threading.Thread中執行，避免 MQTT 的網路連線阻塞 Tkinter 的主 UI 畫面。背景 API 請求：發送 HTTP GET 請求向環境部抓取資料時，同樣啟動背景 Thread (
threading.Thread(target=self.fetch_air_data))，確保 UI 在抓取網路資料時不會出現「沒有回應」或卡頓現象。
2. 全台測站字典與對照表
ALL_SITES_DATA：內建全台 84 個空氣品質觀測站（如基隆、汐止、大里等）與對應縣市的清單。
SITE_COUNTY_MAP：建立快速查詢字典，當 API 回傳缺漏縣市資訊時可自動進行補全。
FIELD_NAMES：提供 24 個標準 API 欄位的英中文名稱對照表（例如：
pm2.5$\rightarrow$細懸浮微粒 (PM2.5 ug/m3)），方便在介面上呈現中文說明。
二、介面 UI 功能模組說明
1. 測站選擇與控制區 (頂部)
ttk.Combobox下拉選單：列出全台 84 個觀測站測站名稱 (縣市)。當使用者點擊切換測站時，會觸發on_combo_selected()，即時將target_site切換為新測站並向 API 發送請求。強制更新按鈕：允許使用者忽略 15 分鐘計時，立即手動發送 HTTP 請求刷新當前測站數據。
2. 模擬 20x4 LCD 螢幕區 (中間深綠區塊)
使用深綠背景 (
#1b3022) 與亮綠字體 (#33ff33)，擬真等寬字型 (Courier New)，完美還原物聯網實體 20x4 LCD 螢幕的顯示效果：第 0 行：AQI 指標與英文狀態（如：
AQI: 54 (Moderate)）。第 1 行：細懸浮微粒濃度（如：
PM2.5: 10.0 ug/m3）。第 2 行：空氣品質判定狀態（如：
Air OK或Air Bad!）。第 3 行：發布時間（如：
Time: 16:00:00）。
3. 分頁數據展示區 (Notebook)
頁籤一「所有數據明細」 (Treeview)：以表格方式列出 API 回傳的全部 24 個欄位，包含英文欄位名、中文說明與數據內容。
頁籤二「RAW JSON」 (Text)：將 API 原始回傳的字典/物件格式化為帶縮排的 JSON 字串，方便開發與除錯。
4. 警報 LED 燈與狀態列 (底部)
tk.Canvas警報 LED 燈：當 $\text{AQI} > 50$ 或 $\text{PM2.5} > 15.4$ 時，LED 轉為紅色 (
#FF0000)（空氣不佳警報）。空氣品質正常時，顯示為綠色 (
#00FF00)。無資料或斷線時，顯示為灰色 (
#888888)。
MQTT 連線狀態標籤：即時顯示目前與 HiveMQ Broker 的連線狀況（連線中/已連線）。
三、MQTT 雙向控制與 API 抓取邏輯
1. 接收外部控制命令（訂閱 alex9ufo/sitename）
當 Node-RED 或其他裝置發布測站名稱（例如：
汐止）至alex9ufo/sitename時：on_mqtt_message()被觸發並解析出汐止。自動將 Python UI 的下拉選單位置同步選取至
汐止 (新北市)。背景自動調用
fetch_air_data()向 API 獲取新測站數據。
2. 動態網址抓取與相容性解析 (fetch_air_data)
動態 URL 轉碼：使用
urllib.parse.quote()將中文測站名稱轉碼，組裝成專屬請求網址：Pythonurl = f"https://data.moenv.gov.tw/api/v2/aqx_p_432?limit=1&api_key={API_KEY}&filters=sitename,eq,{encoded_site}"相容性資料解析：判斷 API 回傳格式為
list陣列或帶有records的dict字典，避免 JSON 結構變更導致程式崩潰。
3. 數據廣播反饋（發行 alex9ufo/weather）
資料成功獲取並解析後，
process_data()會將該測站包含所有 24 個欄位的完整record物件轉為 JSON 字串，發布至 MQTT 主題 alex9ufo/weather。此舉能讓 Node-RED Dashboard 上的 24 欄位 HTML 明細表格立刻收到最新的完整氣象數據。
4. 自動定時輪詢 (timer_loop)
利用
self.root.after(1000, self.timer_loop)每秒檢查時間差。當距離上次更新超過 15 分鐘 (UPDATE_INTERVAL = 900秒) 時，自動觸發 API 抓取，保持數據時效性。
MQTT Box（或類似的 MQTT 測試工具如 MQTTX、MQTT.fx）是物聯網開發中非常重要的測試與除錯工具。
在您目前的系統架構中，Python 程式同時扮演了 Subscriber（訂閱者） 與 Publisher（發行者） 的雙重角色。透過 MQTT Box，您可以輕鬆模擬 Node-RED 或 ESP32 的行為，進行獨立除錯與測試。
一、Python 程式與 MQTT Box 的互動關係
| 測試方向 | MQTT Box 扮演的角色 | Python 程式的反應 | 觀察重點與目的 |
| 測試 1：控制 Python | Publisher (發行者) 發送站名至 | Subscriber (訂閱者) 收到站名後，自動切換下拉選單並向 API 抓資料 | 驗證 Python 程式能否正確接收外部 MQTT 指令，並連動 UI 與 HTTP API 請求。 |
| 測試 2：監聽 Python | Subscriber (訂閱者) 訂閱 | Publisher (發行者) 抓完資料後，發布完整 24 欄位 JSON | 驗證 Python 抓取 API 後，是否有把完整的 JSON 資料正確打包並廣播出去。 |
二、MQTT Box 設定步驟
步驟 1：建立 MQTT Box 連線設定 (MQTT Client)
在 MQTT Box 中點選 Create MQTT Client，輸入以下資訊：
MQTT Client Name:
Test-MQTTBoxMQTT Protocol:
mqtt / tcpBroker Host:
broker.hivemq.comBroker Port:
1883設定完成後點選 Save，並確認狀態顯示為 Connected（綠色）。
三、實作互動測試
情況 A：用 MQTT Box 發送指令給 Python（模擬 Node-RED 切換站名）
在 MQTT Box 的 Publish 區塊進行設定：
Topic to publish:
alex9ufo/sitenamePayload:
汐止（或士林、大里等任意全台測站）
點選 Publish 按鈕。
🔍 Python 端預期反應：
Console 控制台會印出：
[MQTT Rx] 收到 Topic: 'alex9ufo/sitename' | Payload 測站: '汐止'。Python GUI 畫面的下拉選單會自動切換至
汐止 (新北市)。背景會發送 HTTP 請求向環境部抓取汐止的空氣品質資料，並更新 Python 畫面的 LCD 螢幕與 LED 燈號。
情況 B：用 MQTT Box 接收 Python 發出的氣象資料（模擬 Node-RED 接收表格數據）
在 MQTT Box 的 Subscribe 區塊進行設定：
Topic to subscribe:
alex9ufo/weatherQoS:
0或1
點選 Subscribe 按鈕。
此時回到 Python GUI 畫面，隨便切換一個測站（或點擊「強制更新」按鈕）。
🔍 MQTT Box 預期反應：
MQTT Box 的 Subscribe 區塊會立刻收到一包由 Python 發出的 JSON Payload，內容包含 24 個完整欄位：
{
"sitename": "汐止",
"county": "新北市",
"aqi": "45",
"pollutant": "",
"status": "良好",
"so2": "1.2",
"co": "0.21",
"pm2.5": "9",
"publishtime": "2026/10/02 09:00:00",
...
}
四、效益與除錯價值
使用 MQTT Box 與 Python 程式互動有三大好處：
單元測試 (Unit Testing)：不需要同時開著 ESP32 模擬器或 Node-RED，就能單獨驗證 Python 程式的 MQTT 功能是否正常。
格式檢驗：可以直接檢查 Python 發出的 JSON 格式是否有語法錯誤或欄位缺漏。
網路除錯：確認 HiveMQ Broker 是否運作正常、訊息有無延遲或丟包狀況。
沒有留言:
張貼留言