2026年10月1日 星期四

台中公車雙向即時資訊系統 (TaichungBusLiveQuery)- Node-red 程式

 台中公車雙向即時資訊系統 (TaichungBusLiveQuery)  Node-red 程式






  • Python Tkinter 是單機桌面 UI 程式,主動發起 API 請求並渲染視窗。

  • Node-RED 是基於事件驅動的流程(Flow-based)系統,適合在背景執行、定時抓取資料,並透過 Dashboard (UI 介面) 提供網頁端網頁視窗供使用者選擇與檢視。

Node-RED 實現架構說明

在 Node-RED 中,我們可以使用 node-red-dashboard(或新版的 node-red-dashboard-2)來建構網頁 UI。

整體流程分為 3 大核心區塊:

  1. 取得 Access Token:啟動時及每小時候自動向 TDX 取得 Access Token。

  2. 載入業者與路線選單:系統啟動時載入路線資料,並依「業者 $\rightarrow$ 路線」構建動態下拉選單。

  3. 30 秒定時輪詢與資料渲染:當使用者選定路線後,啟動 30 秒定時輪詢,並依去程/返程將結果透過 HTML 樣式(紅/綠/黑文字)呈現於介面上。

Node-RED Flow JSON 程式碼

請複製下方 JSON 程式碼,並在 Node-RED 菜單中選擇 Import(匯入) 貼上即可:


[{"id":"inject_init","type":"inject","z":"ece6aff5c6943929","name":"啟動初始化","props":[],"repeat":"3600","crontab":"","once":true,"onceDelay":0.1,"topic":"","x":170,"y":160,"wires":[["func_get_token"]]},{"id":"func_get_token","type":"function","z":"ece6aff5c6943929","name":"準備 Auth 請求","func":"msg.url = 'https://tdx.transportdata.tw/auth/realms/TDXConnect/protocol/openid-connect/token';\nmsg.headers = {'content-type': 'application/x-www-form-urlencoded'};\nmsg.payload = {\n    'grant_type': 'client_credentials',\n    'client_id': 'alex9ufo-d431e6e81-d1850-4d96',\n    'client_secret': '925521bb-86014-40610-b38c-8bead1e8ae704'\n};\nreturn msg;","outputs":1,"timeout":"","noerr":0,"initialize":"","finalize":"","libs":[],"x":370,"y":160,"wires":[["http_auth"]]},{"id":"http_auth","type":"http request","z":"ece6aff5c6943929","name":"取得 Token","method":"POST","ret":"obj","paytoqs":"ignore","url":"","tls":"","persist":false,"proxy":"","insecureHTTPParser":false,"authType":"","senderr":false,"headers":[],"x":550,"y":160,"wires":[["func_save_token"]]},{"id":"func_save_token","type":"function","z":"ece6aff5c6943929","name":"儲存 Token 並請求路線資料","func":"global.set('tdx_token', msg.payload.access_token);\n\nmsg.url = 'https://tdx.transportdata.tw/api/basic/v2/Bus/Route/City/Taichung?$format=JSON';\nmsg.headers = {'Authorization': 'Bearer ' + msg.payload.access_token};\nmsg.payload = {};\nreturn msg;","outputs":1,"noerr":0,"initialize":"","finalize":"","libs":[],"x":790,"y":160,"wires":[["http_routes"]]},{"id":"http_routes","type":"http request","z":"ece6aff5c6943929","name":"取得公車路線資料","method":"GET","ret":"obj","paytoqs":"ignore","url":"","tls":"","persist":false,"proxy":"","insecureHTTPParser":false,"authType":"","senderr":false,"headers":[],"x":1050,"y":160,"wires":[["func_parse_operators"]]},{"id":"func_parse_operators","type":"function","z":"ece6aff5c6943929","name":"解析業者與路線資料","func":"let routes = msg.payload;\nlet operators_map = {};\n\nroutes.forEach(r => {\n    let routeName = r.RouteName ? r.RouteName.Zh_tw : '未知路線';\n    let routeUID = r.RouteUID;\n    let operators = r.Operators || [];\n    \n    operators.forEach(op => {\n        let opName = op.OperatorName ? op.OperatorName.Zh_tw : '其他業者';\n        if (!operators_map[opName]) {\n            operators_map[opName] = {};\n        }\n        if (!operators_map[opName][routeName]) {\n            operators_map[opName][routeName] = [];\n        }\n        operators_map[opName][routeName].push(routeUID);\n    });\n});\n\nglobal.set('operators_map', operators_map);\n\nlet opList = Object.keys(operators_map).sort();\nmsg.options = opList;\nreturn [msg, null];","outputs":1,"noerr":0,"initialize":"","finalize":"","libs":[],"x":380,"y":220,"wires":[["ui_select_operator"]]},{"id":"ui_select_operator","type":"ui_dropdown","z":"ece6aff5c6943929","name":"1. 選擇業者","label":"1. 選擇公車業者","tooltip":"","place":"請選擇業者","group":"taichung_bus_group_select","order":1,"width":4,"height":1,"passthru":false,"multiple":false,"options":[],"payload":"","topic":"operator_selected","topicType":"str","className":"","x":590,"y":220,"wires":[["func_update_routes_dropdown"]]},{"id":"func_update_routes_dropdown","type":"function","z":"ece6aff5c6943929","name":"更新路線下拉選單","func":"let selectedOp = msg.payload;\nflow.set('selected_op', selectedOp);\nlet operators_map = global.get('operators_map') || {};\n\nif (operators_map[selectedOp]) {\n    let routes = Object.keys(operators_map[selectedOp]).sort();\n    msg.options = routes;\n} else {\n    msg.options = [];\n}\nreturn msg;","outputs":1,"noerr":0,"initialize":"","finalize":"","libs":[],"x":770,"y":220,"wires":[["ui_select_route"]]},{"id":"ui_select_route","type":"ui_dropdown","z":"ece6aff5c6943929","name":"2. 選擇路線","label":"2. 選擇公車路線","tooltip":"","place":"請選擇路線","group":"taichung_bus_group_select","order":4,"width":4,"height":1,"passthru":false,"multiple":false,"options":[],"payload":"","topic":"route_selected","topicType":"str","className":"","x":990,"y":220,"wires":[["func_trigger_fetch_eta"]]},{"id":"func_trigger_fetch_eta","type":"function","z":"ece6aff5c6943929","name":"記錄選擇路線並觸發抓取","func":"flow.set('selected_route', msg.payload);\nreturn { payload: 'TRIGGER' };","outputs":1,"noerr":0,"initialize":"","finalize":"","libs":[],"x":390,"y":320,"wires":[["func_prepare_eta_request"]]},{"id":"inject_30s_loop","type":"inject","z":"ece6aff5c6943929","name":"每 30 秒定時觸發","props":[],"repeat":"30","crontab":"","once":false,"onceDelay":0.1,"topic":"","x":390,"y":380,"wires":[["func_prepare_eta_request"]]},{"id":"func_prepare_eta_request","type":"function","z":"ece6aff5c6943929","name":"準備 ETA 請求","func":"let token = global.get('tdx_token');\nlet selectedRoute = flow.get('selected_route');\n\nif (!token || !selectedRoute) {\n    return null;\n}\n\nmsg.url = 'https://tdx.transportdata.tw/api/basic/v2/Bus/EstimatedTimeOfArrival/City/Taichung?$format=JSON';\nmsg.headers = {'Authorization': 'Bearer ' + token};\nreturn msg;","outputs":1,"noerr":0,"initialize":"","finalize":"","libs":[],"x":600,"y":320,"wires":[["http_eta"]]},{"id":"http_eta","type":"http request","z":"ece6aff5c6943929","name":"取得 ETA 到站資料","method":"GET","ret":"obj","paytoqs":"ignore","url":"","tls":"","persist":false,"proxy":"","insecureHTTPParser":false,"authType":"","senderr":false,"headers":[],"x":790,"y":320,"wires":[["func_render_eta_tables"]]},{"id":"func_render_eta_tables","type":"function","z":"ece6aff5c6943929","name":"解析 ETA 並生成 HTML 表格","func":"let etaData = msg.payload || [];\nlet selectedOp = flow.get('selected_op');\nlet selectedRoute = flow.get('selected_route');\nlet operators_map = global.get('operators_map') || {};\n\nif (!selectedOp || !selectedRoute) return null;\n\nlet targetUIDs = (operators_map[selectedOp] && operators_map[selectedOp][selectedRoute]) ? operators_map[selectedOp][selectedRoute] : [];\n\nlet matchedEtas = etaData.filter(e => targetUIDs.includes(e.RouteUID));\n\nlet outwardList = matchedEtas.filter(e => e.Direction === 0).sort((a,b) => (a.StopSequence || 0) - (b.StopSequence || 0));\nlet returnList = matchedEtas.filter(e => e.Direction === 1).sort((a,b) => (a.StopSequence || 0) - (b.StopSequence || 0));\n\nconst ETA_STATUS_MAP = { 0: \"正常預估\", 1: \"尚未發車\", 2: \"交管不停靠\", 3: \"末班車已過\", 4: \"今日未營運\" };\n\nfunction generateHTMLTable(list, isOutward) {\n    let defaultColor = isOutward ? 'darkgreen' : 'black';\n    let html = '<table style=\"width:100%; border-collapse: collapse;\">';\n    html += '<tr style=\"border-bottom: 2px solid #ddd;\"><th style=\"text-align:center;\">順序</th><th style=\"text-align:left;\">站牌名稱</th><th style=\"text-align:center;\">預估到站</th></tr>';\n    \n    list.forEach((eta, idx) => {\n        let stopSeq = eta.StopSequence || (idx + 1);\n        let stopName = eta.StopName ? eta.StopName.Zh_tw : '未知站牌';\n        let status = eta.EstimateTime;\n        let statusStr = '';\n        let textColor = defaultColor;\n        let isBold = false;\n        \n        if (status !== undefined && status !== null) {\n            let minutes = Math.floor(status / 60);\n            if (minutes <= 10) {\n                textColor = 'red';\n                isBold = true;\n            }\n            statusStr = minutes < 1 ? '即將進站' : `${minutes} 分鐘`;\n        } else {\n            let code = eta.EstimateStatus !== undefined ? eta.EstimateStatus : 1;\n            statusStr = ETA_STATUS_MAP[code] || '尚未開出';\n        }\n        \n        let fontStyle = `color: ${textColor}; ${isBold ? 'font-weight: bold;' : ''}`;\n        html += `<tr style=\"border-bottom: 1px solid #eee; ${fontStyle}\">`;\n        html += `<td style=\"text-align:center; padding: 6px;\">${stopSeq}</td>`;\n        html += `<td style=\"text-align:left; padding: 6px;\">${stopName}</td>`;\n        html += `<td style=\"text-align:center; padding: 6px;\">${statusStr}</td>`;\n        html += '</tr>';\n    });\n    html += '</table>';\n    return html;\n}\n\nlet outwardHTML = generateHTMLTable(outwardList, true);\nlet returnHTML = generateHTMLTable(returnList, false);\n\nreturn [\n    { payload: outwardHTML },\n    { payload: returnHTML }\n];","outputs":2,"noerr":0,"initialize":"","finalize":"","libs":[],"x":1020,"y":320,"wires":[["ui_outward_template"],["ui_return_template"]]},{"id":"ui_outward_template","type":"ui_template","z":"ece6aff5c6943929","group":"taichung_bus_group_outward","name":"往程表格渲染","order":1,"width":8,"height":10,"format":"<div ng-bind-html=\"msg.payload\"></div>","storeOutMessages":true,"fwdInMessages":true,"resendOnRefresh":true,"templateScope":"local","className":"","x":1240,"y":280,"wires":[[]]},{"id":"ui_return_template","type":"ui_template","z":"ece6aff5c6943929","group":"taichung_bus_group_return","name":"返程表格渲染","order":1,"width":8,"height":10,"format":"<div ng-bind-html=\"msg.payload\"></div>","storeOutMessages":true,"fwdInMessages":true,"resendOnRefresh":true,"templateScope":"local","className":"","x":1240,"y":340,"wires":[[]]},{"id":"taichung_bus_group_select","type":"ui_group","name":"路線選擇","tab":"taichung_bus_tab","order":1,"disp":true,"width":5},{"id":"taichung_bus_group_outward","type":"ui_group","name":"往程 (去程)","tab":"taichung_bus_tab","order":2,"disp":true,"width":8},{"id":"taichung_bus_group_return","type":"ui_group","name":"返程","tab":"taichung_bus_tab","order":3,"disp":true,"width":8},{"id":"taichung_bus_tab","type":"ui_tab","name":"台中公車即時通","icon":"bus","order":1},{"id":"ce7c20be6a357e08","type":"global-config","env":[],"modules":{"node-red-dashboard":"3.6.6"}}]


 Node-RED 流程中,已經包含了「30 秒定期更新」的機制。

它是透過流程中的 inject(注入節點) 來實現的。

🔍 它是如何在 Node-RED 中運作的?

在匯入的流程中,有一個名為 每 30 秒定時觸發 的節點:

[ 啟動/選擇路線 ] ───┐
                    ├───> [ 準備 ETA 請求 ] ───> [ 抓取 TDX API ] ───> [ 更新畫面 HTML ]
[ 每 30 秒觸發 ] ───┘
  1. 第一次觸發:當您在網頁下拉選單選擇了公車號碼時,流程會立即發起第一次查詢並顯示結果。

  2. 後續自動循環:每 30 秒定時觸發 節點會每隔 30 秒自動發送一個訊號給 準備 ETA 請求 節點,重新向 TDX 抓取最新的公車預估到站資料並更新網頁畫面。

⚙️ 如何在 Node-RED 檢查或修改更新時間?

如果您想確認設定或調整時間(例如改為 15 秒或 60 秒):

  1. 在 Node-RED 工作區雙擊開啟 每 30 秒定時觸發 節點。

  2. 查看 Repeat(重複) 屬性:

    • 模式設定為 at interval(固定間隔)。

    • 時間設定為 every 30 seconds(每 30 秒)。

  3. 修改完畢後點擊右上角的 Deploy(部署) 即可生效。

台中公車雙向即時資訊系統 (TaichungBusLiveQuery)

台中公車雙向即時資訊系統 (TaichungBusLiveQuery)

這份程式碼是一個基於 Python Tkinter 所開發的桌面應用程式,專門用來實時查詢臺中市公車預估到站時間。

它整合了交通部 TDX 運輸資料流通服務平臺的開放 API,並具備兩層式動態下拉選單、去程/返程分頁顯示、30 秒背景自動定時更新以及到站時間顏色提示等功能。

一、 主要功能亮點

  1. 二階聯動選單(業者 $\rightarrow$ 路線):

    • 第一層:先選擇公車業者(如:台中客運、統聯客運等)。

    • 第二層:根據所選業者,動態載入該業者營運的公車路線號碼。

  2. 去程與返程分頁呈現:

    • 利用 ttk.Notebook 標籤頁將路線分為「往程 (去程)」與「返程」兩個獨立表格,避免資料混雜。

  3. 顏色識別系統:

    • 去程基本字體:綠色 (darkgreen)。

    • 返程基本字體:黑色 (black)。

    • 進站預警($\le 10$ 分鐘):無論去程或返程,只要預估到站時間小於等於 10 分鐘(包含「即將進站」),該列會粗體紅色 (red) 顯示。

  4. 30 秒自動背景更新機制:

    • 選定路線後,系統每秒進行倒數(於介面上顯示 🔄 N 秒後自動更新)。

    • 倒數歸零時自動向 TDX 抓取最新到站時間,無需手動刷網頁,且更新過程 UI 介面不會凍結或卡頓。

二、 程式碼結構與運作原理

程式結構主要分為 API 資料處理 與 UI 介面邏輯 兩大模組:

┌──────────────────────────────────────┐
│                      主程式 (Tkinter)                                                   │
├─────────────── ┬──────────────────────┤
│       1. 認證與抓取           │         2. 介面互動                             │
│  • get_access_token()    │  • combo_operator (選業者)             │
│  • fetch_routes()             │  • combo_route (選路線)                   │
│  • fetch_eta_data()         │  • Treeview (去/返程表格)                │
├──────────────────────────┴───────────┤
│                     3. 自動更新機制                                                   │
│  • start_timer() / update_timer_tick() / stop_timer()                  │
└──────────────────────────────── ─────┘

1. API 驗證與資料請求模組

  • get_access_token():

    • 使用 OAuth2 client_credentials 模式,將您提供的 CLIENT_ID 及 CLIENT_SECRET 發送至 TDX 驗證伺服器 (AUTH_URL),取得臨時存取金鑰(Token)。

  • fetch_routes():

    • 向 API 請求臺中市所有公車路線清單,包含路線名稱與營運業者資訊。此動作僅在程式啟動時執行 1 次。

  • fetch_eta_data():

    • 向 API /v2/Bus/EstimatedTimeOfArrival/City/Taichung 請求全臺中市公車的即時預估到站資料。這部分會每 30 秒重複調用。

2. UI 互動與邏輯處理模組 (TaichungBusApp 類別)

  • init_data()(初始化):

    • 呼叫 API 取得路線資料後,整理成雜湊表結構:{ 業者名稱: { 路線名稱: [RouteUIDs] } }。

    • 填入第一層「業者」下拉選單。

  • on_operator_selected()(選擇業者時):

    • 當使用者選定業者後,根據業者名稱從資料結構中找出對應的路線,並更新第二層「路線」選單。同時暫停計時器並清空表格。

  • on_route_selected()(選擇路線時):

    • 選定路線後,立即調用 refresh_eta_and_update_ui() 抓取最新到站時間並畫在表格上,隨後調用 start_timer() 啟動 30 秒自動倒數。

  • refresh_eta_and_update_ui() 與 populate_tree()(更新與繪製):

    1. 根據目前選定的路線 UID,在所有到站資料中進行過濾。

    2. 依 Direction == 0(去程)與 Direction == 1(返程)將資料分流,並按 StopSequence(站牌順序)進行排序。

    3. 依到站時間進行邏輯判斷:

      • EstimateTime 換算為分鐘。

      • 若 $\le 10$ 分鐘,套用 red_text 標籤(紅色)。

      • 若 $> 10$ 分鐘,去程套用 green_text,返程套用 black_text。

      • 若未發車或異常,則查表 ETA_STATUS_MAP 顯示文字(如「尚未發車」、「末班車已過」)。

3. 自動更新機制 (after 非阻塞定時器)

程式沒有使用傳統多執行緒(Threading),而是使用 Tkinter 的內建排程 self.root.after(1000, self.update_timer_tick):

  1. 每秒執行一次 update_timer_tick(),將倒數秒數減 1,並更新 UI 狀態標籤(如:🔄 29 秒後自動更新)。

  2. 當 countdown == 0 時,觸發 API 重新讀取,刷新介面並將秒數重置為 30,形成循環。

  3. 當使用者切換業者或路線時,會觸發 stop_timer() 取消原本的排程,防止背景計時器重複或發送無效請求。

  4. TDX運輸資料流通服務平臺






import requests

import json

import tkinter as tk

from tkinter import ttk, messagebox


# ================= 1. TDX API 設定 =================

#CLIENT_ID = "YOUR_CLIENT_ID_HERE"      # 請填入您的 Client ID

#CLIENT_SECRET = "YOUR_CLIENT_SECRET_HERE"  # 請填入您的 Client Secret

# ================= 1. TDX API 設定 =================

CLIENT_ID = "alex9ufo-d413e6e281-d8e560-e4d296"      # 請填入您的 Client ID

CLIENT_SECRET = "9251521bb-81604-432060-b38c-8b2eade28ae704"  # 請填入您的 Client Secret


# API 網址設定

AUTH_URL = "https://tdx.transportdata.tw/auth/realms/TDXConnect/protocol/openid-connect/token"

ROUTE_API_URL = "https://tdx.transportdata.tw/api/basic/v2/Bus/Route/City/Taichung"

ETA_API_URL = "https://tdx.transportdata.tw/api/basic/v2/Bus/EstimatedTimeOfArrival/City/Taichung"


REFRESH_INTERVAL = 30  # 設定自動更新週期為 30 秒


# 狀態碼對照表

ETA_STATUS_MAP = {

    0: "正常預估",

    1: "尚未發車",

    2: "交管不停靠",

    3: "末班車已過",

    4: "今日未營運"

}


# ================= 2. API 請求函式 =================

def get_access_token():

    """取得 TDX OAuth2 Access Token"""

    auth_data = {

        'grant_type': 'client_credentials',

        'client_id': CLIENT_ID,

        'client_secret': CLIENT_SECRET

    }

    try:

        response = requests.post(AUTH_URL, data=auth_data, timeout=10)

        response.raise_for_status()

        return response.json().get('access_token')

    except Exception as e:

        messagebox.showerror("認證失敗", f"無法取得 Access Token,請檢查 Key 設定。\n{e}")

        return None


def fetch_routes(token):

    """取得公車路線與業者資訊"""

    headers = {"Authorization": f"Bearer {token}"}

    res = requests.get(f"{ROUTE_API_URL}?$format=JSON", headers=headers, timeout=10)

    return res.json() if res.status_code == 200 else []


def fetch_eta_data(token):

    """單獨取得預估到站資料 (批次更新)"""

    headers = {"Authorization": f"Bearer {token}"}

    res = requests.get(f"{ETA_API_URL}?$format=JSON", headers=headers, timeout=10)

    return res.json() if res.status_code == 200 else []


# ================= 3. Tkinter UI 邏輯 =================

class TaichungBusApp:

    def __init__(self, root):

        self.root = root

        self.root.title("臺中市公車預估到站查詢(30秒自動更新+顏色顯示)")

        self.root.geometry("680x580")


        # 資料與狀態儲存

        self.token = None

        self.operators_map = {}  # {業者名稱: {路線名稱: [RouteUID, ...]}}

        self.eta_data = []       # API 傳回的所有 ETA 資料

        

        self.countdown = REFRESH_INTERVAL  # 倒數計時器

        self.timer_id = None               # Tkinter after 計時器 ID


        # 畫面元件佈局

        self.setup_ui()

        

        # 初始化載入路線與業者資料

        self.init_data()


    def setup_ui(self):

        # 選單與狀態區域

        frame_select = ttk.Frame(self.root, padding=10)

        frame_select.pack(fill=tk.X)


        # 第一層選單:業者

        ttk.Label(frame_select, text="1. 選擇公車業者:").grid(row=0, column=0, sticky=tk.W, pady=5)

        self.combo_operator = ttk.Combobox(frame_select, state="readonly", width=28)

        self.combo_operator.grid(row=0, column=1, pady=5, padx=5)

        self.combo_operator.bind("<<ComboboxSelected>>", self.on_operator_selected)


        # 第二層選單:公車路線

        ttk.Label(frame_select, text="2. 選擇公車路線:").grid(row=1, column=0, sticky=tk.W, pady=5)

        self.combo_route = ttk.Combobox(frame_select, state="readonly", width=28)

        self.combo_route.grid(row=1, column=1, pady=5, padx=5)

        self.combo_route.bind("<<ComboboxSelected>>", self.on_route_selected)


        # 立即更新按鈕

        btn_refresh = ttk.Button(frame_select, text="手動立即更新", command=self.manual_refresh)

        btn_refresh.grid(row=0, column=2, padx=10)


        # 自動更新狀態提示標籤

        self.lbl_status = ttk.Label(frame_select, text="請選擇業者與路線", foreground="gray")

        self.lbl_status.grid(row=1, column=2, padx=10, sticky=tk.W)


        # 使用 Notebook (頁籤) 將「去程」與「返程」分開

        self.notebook = ttk.Notebook(self.root)

        self.notebook.pack(fill=tk.BOTH, expand=True, padx=10, pady=5)


        # 建立「去程」分頁與表格

        self.tab_outward = ttk.Frame(self.notebook)

        self.notebook.add(self.tab_outward, text=" 往程 (去程) ")

        self.tree_outward = self.create_treeview(self.tab_outward)


        # 建立「返程」分頁與表格

        self.tab_return = ttk.Frame(self.notebook)

        self.notebook.add(self.tab_return, text=" 返程 ")

        self.tree_return = self.create_treeview(self.tab_return)


        # 廣域設定 Treeview 樣式 tags 顏色

        # 去程預設:綠色,返程預設:黑色,10分鐘內:紅色

        for tree in (self.tree_outward, self.tree_return):

            tree.tag_configure("green_text", foreground="darkgreen")

            tree.tag_configure("black_text", foreground="black")

            tree.tag_configure("red_text", foreground="red", font=('TkDefaultFont', 9, 'bold'))


    def create_treeview(self, parent):

        """共用的表格建立邏輯"""

        frame = ttk.Frame(parent, padding=5)

        frame.pack(fill=tk.BOTH, expand=True)


        columns = ("stop_seq", "stop_name", "eta_status")

        tree = ttk.Treeview(frame, columns=columns, show="headings")

        

        tree.heading("stop_seq", text="順序")

        tree.heading("stop_name", text="站牌名稱")

        tree.heading("eta_status", text="預估到站狀態")

        

        tree.column("stop_seq", width=60, anchor=tk.CENTER)

        tree.column("stop_name", width=300, anchor=tk.W)

        tree.column("eta_status", width=180, anchor=tk.CENTER)


        scrollbar = ttk.Scrollbar(frame, orient=tk.VERTICAL, command=tree.yview)

        tree.configure(yscroll=scrollbar.set)

        

        tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)

        scrollbar.pack(side=tk.RIGHT, fill=tk.Y)

        return tree


    def init_data(self):

        """取得 Token 並載入路線與業者資料"""

        self.token = get_access_token()

        if not self.token:

            return


        routes = fetch_routes(self.token)

        self.operators_map.clear()


        # 解析路線資料,整理為 { 業者: { 路線名: [RouteUIDs] } }

        for r in routes:

            operators = r.get("Operators", [])

            route_name = r.get("RouteName", {}).get("Zh_tw", "未知路線")

            route_uid = r.get("RouteUID")


            for op in operators:

                op_name = op.get("OperatorName", {}).get("Zh_tw", "其他業者")

                if op_name not in self.operators_map:

                    self.operators_map[op_name] = {}

                

                if route_name not in self.operators_map[op_name]:

                    self.operators_map[op_name][route_name] = []

                self.operators_map[op_name][route_name].append(route_uid)


        # 更新第一層選單(業者)

        operators_list = sorted(list(self.operators_map.keys()))

        self.combo_operator["values"] = operators_list

        self.combo_route["values"] = []

        self.combo_route.set("")

        self.combo_operator.set("")

        self.clear_trees()


    def clear_trees(self):

        """清空去程與返程表格"""

        for item in self.tree_outward.get_children():

            self.tree_outward.delete(item)

        for item in self.tree_return.get_children():

            self.tree_return.delete(item)


    def on_operator_selected(self, event):

        """選取業者時,更新路線選單,並停止定時器"""

        self.stop_timer()

        self.lbl_status.config(text="請選擇路線", foreground="gray")

        

        selected_op = self.combo_operator.get()

        if selected_op in self.operators_map:

            routes_list = sorted(list(self.operators_map[selected_op].keys()))

            self.combo_route["values"] = routes_list

            self.combo_route.set("")

            self.clear_trees()


    def on_route_selected(self, event):

        """選取路線時,觸發首次更新並啟動 30 秒自動更新"""

        self.stop_timer()

        self.refresh_eta_and_update_ui()

        self.start_timer()


    def refresh_eta_and_update_ui(self):

        """核心向 API 抓取 ETA 並刷新表格畫面的邏輯"""

        selected_op = self.combo_operator.get()

        selected_route = self.combo_route.get()


        if not selected_op or not selected_route or not self.token:

            return


        target_uids = self.operators_map.get(selected_op, {}).get(selected_route, [])

        

        # 重新向 API 請求最新預估到站資料

        self.eta_data = fetch_eta_data(self.token)

        self.clear_trees()


        # 篩選選擇路線的資料

        matched_etas = [eta for eta in self.eta_data if eta.get("RouteUID") in target_uids]


        # 分別整理去程 (Direction=0) 與 返程 (Direction=1)

        outward_list = [e for e in matched_etas if e.get("Direction") == 0]

        return_list = [e for e in matched_etas if e.get("Direction") == 1]


        # 依站牌順序 (StopSequence) 排序

        outward_list.sort(key=lambda x: x.get("StopSequence", 0))

        return_list.sort(key=lambda x: x.get("StopSequence", 0))


        # 填入 UI (去程傳入預設標籤 green_text, 返程傳入 black_text)

        self.populate_tree(self.tree_outward, outward_list, default_tag="green_text")

        self.populate_tree(self.tree_return, return_list, default_tag="black_text")


    def populate_tree(self, tree, eta_list, default_tag):

        """將 ETA 資料寫入 Treeview 並依條件判斷字型顏色"""

        for idx, eta in enumerate(eta_list, start=1):

            stop_seq = eta.get("StopSequence", idx)

            stop_name = eta.get("StopName", {}).get("Zh_tw", "未知站牌")

            status = eta.get("EstimateTime")

            

            tag_to_use = default_tag

            

            if status is not None:

                minutes = status // 60

                

                # 判定:若小於等於 10 分鐘(包含即將進站),使用紅色文字

                if minutes <= 10:

                    tag_to_use = "red_text"


                status_str = "即將進站" if minutes < 1 else f"{minutes} 分鐘"

            else:

                code = eta.get("EstimateStatus", 1)

                status_str = ETA_STATUS_MAP.get(code, "尚未開出")


            tree.insert("", tk.END, values=(stop_seq, stop_name, status_str), tags=(tag_to_use,))


    # ================= 4. 自動定時器機制 =================

    def start_timer(self):

        """啟動 1 秒執行一次的倒數計時器"""

        self.countdown = REFRESH_INTERVAL

        self.update_timer_tick()


    def stop_timer(self):

        """停止計時器"""

        if self.timer_id:

            self.root.after_cancel(self.timer_id)

            self.timer_id = None


    def update_timer_tick(self):

        """每秒更新一次倒數文字,歸零時觸發 API 重新整理"""

        self.lbl_status.config(text=f"🔄 {self.countdown} 秒後自動更新", foreground="blue")

        

        if self.countdown == 0:

            self.refresh_eta_and_update_ui()

            self.countdown = REFRESH_INTERVAL  # 重設倒數時間

        else:

            self.countdown -= 1


        # 排程 1000 毫秒 (1秒) 後再次觸發

        self.timer_id = self.root.after(1000, self.update_timer_tick)


    def manual_refresh(self):

        """手動立即更新按鈕事件"""

        if self.combo_route.get():

            self.stop_timer()

            self.refresh_eta_and_update_ui()

            self.start_timer()


# ================= 5. 主程式啟動 =================

if __name__ == "__main__":

    root = tk.Tk()

    app = TaichungBusApp(root)

    root.mainloop()


全台空氣品質智慧雙向監控系統 (Python & Node-RED IoT Solution)

全台空氣品質智慧雙向監控系統 (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 時:

    1. on_mqtt_message() 被觸發並解析出 汐止。

    2. 自動將 Python UI 的下拉選單位置同步選取至 汐止 (新北市)。

    3. 背景自動調用 fetch_air_data() 向 API 獲取新測站數據。

2. 動態網址抓取與相容性解析 (fetch_air_data)

  • 動態 URL 轉碼:使用 urllib.parse.quote() 將中文測站名稱轉碼,組裝成專屬請求網址:

    Python
    url = 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 (發行者)


發送站名至 alex9ufo/sitename

Subscriber (訂閱者)


收到站名後,自動切換下拉選單並向 API 抓資料

驗證 Python 程式能否正確接收外部 MQTT 指令,並連動 UI 與 HTTP API 請求。
測試 2:監聽 Python

Subscriber (訂閱者)


訂閱 alex9ufo/weather

Publisher (發行者)


抓完資料後,發布完整 24 欄位 JSON

驗證 Python 抓取 API 後,是否有把完整的 JSON 資料正確打包並廣播出去。

二、MQTT Box 設定步驟

步驟 1:建立 MQTT Box 連線設定 (MQTT Client)

在 MQTT Box 中點選 Create MQTT Client,輸入以下資訊:

  • MQTT Client Name: Test-MQTTBox

  • MQTT Protocol: mqtt / tcp

  • Broker Host: broker.hivemq.com

  • Broker Port: 1883

  • 設定完成後點選 Save,並確認狀態顯示為 Connected(綠色)。

三、實作互動測試

情況 A:用 MQTT Box 發送指令給 Python(模擬 Node-RED 切換站名)

  1. 在 MQTT Box 的 Publish 區塊進行設定:

    • Topic to publish: alex9ufo/sitename

    • Payload: 汐止(或 士林、大里 等任意全台測站)

  2. 點選 Publish 按鈕。

🔍 Python 端預期反應:

  • Console 控制台會印出:[MQTT Rx] 收到 Topic: 'alex9ufo/sitename' | Payload 測站: '汐止'。

  • Python GUI 畫面的下拉選單會自動切換至 汐止 (新北市)。

  • 背景會發送 HTTP 請求向環境部抓取汐止的空氣品質資料,並更新 Python 畫面的 LCD 螢幕與 LED 燈號。

情況 B:用 MQTT Box 接收 Python 發出的氣象資料(模擬 Node-RED 接收表格數據)

  1. 在 MQTT Box 的 Subscribe 區塊進行設定:

    • Topic to subscribe: alex9ufo/weather

    • QoS: 0 或 1

  2. 點選 Subscribe 按鈕。

  3. 此時回到 Python GUI 畫面,隨便切換一個測站(或點擊「強制更新」按鈕)。

🔍 MQTT Box 預期反應:

  • MQTT Box 的 Subscribe 區塊會立刻收到一包由 Python 發出的 JSON Payload,內容包含 24 個完整欄位:

JSON
{
    "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 程式互動有三大好處:

  1. 單元測試 (Unit Testing):不需要同時開著 ESP32 模擬器或 Node-RED,就能單獨驗證 Python 程式的 MQTT 功能是否正常。

  2. 格式檢驗:可以直接檢查 Python 發出的 JSON 格式是否有語法錯誤或欄位缺漏。

  3. 網路除錯:確認 HiveMQ Broker 是否運作正常、訊息有無延遲或丟包狀況。

台中公車雙向即時資訊系統 (TaichungBusLiveQuery)- Node-red 程式

  台中公車雙向即時資訊系統 ( TaichungBusLiveQuery )   Node-red 程式 Python Tkinter 是單機桌面 UI 程式,主動發起 API 請求並渲染視窗。 Node-RED 是基於事件驅動的流程(Flow-based)系統,適合在背...