2025年8月4日 星期一

opendata 空氣品質指標(AQI) ---Python TKinter

 opendata 空氣品質指標(AQI) ---Python TKinter

空氣品質指標(AQI)

每小時提供各測站之空氣品質指標(AQI),原始資料版本公告於空氣品質監測網https://airtw.moenv.gov.tw


https://data.moenv.gov.tw/dataset/detail/AQX_P_432


https://data.moenv.gov.tw/api/v2/aqx_p_432?api_key=9b651a1b-0732-418e-b4e9-e784417cadef&limit=1000&sort=ImportDate%20desc&format=JSON



<<Python TKinter程式>>

import tkinter as tk

from tkinter import ttk

from tkinter import scrolledtext

import requests

import json

import threading


# --- API URL ---

API_URL = "https://data.moenv.gov.tw/api/v2/aqx_p_432?api_key=9b651a1b-0732-418e-b4e9-e784417cadef&limit=1000&sort=ImportDate%20desc&format=JSON"


class AQIApp:

    def __init__(self, master):

        self.master = master

        master.title("台灣空氣品質監測站 (AQI)")


        self.label = tk.Label(master, text="選擇測站名稱:")

        self.label.pack()

        

        self.sitename_combobox = ttk.Combobox(master, state="readonly")

        self.sitename_combobox.pack()


        self.fetch_single_button = tk.Button(master, text="查詢選定測站", command=self.start_fetch_single, state=tk.DISABLED)

        self.fetch_single_button.pack()


        self.fetch_all_button = tk.Button(master, text="查詢所有測站", command=self.start_fetch_all, state=tk.DISABLED)

        self.fetch_all_button.pack()


        self.status_label = tk.Label(master, text="")

        self.status_label.pack()


        self.result_text = scrolledtext.ScrolledText(master, width=80, height=20)

        self.result_text.pack()


        self.update_status("正在載入測站清單...", "blue")

        threading.Thread(target=self.load_sitenames).start()

    

    def load_sitenames(self):

        try:

            response = requests.get(API_URL, timeout=10)

            response.raise_for_status()

            data = response.json()

            records = data.get('records', [])

            

            sitenames = sorted([record.get('sitename') for record in records if record.get('sitename')])

            

            self.sitename_combobox['values'] = sitenames

            

            if sitenames:

                self.sitename_combobox.current(0)

                self.fetch_single_button['state'] = tk.NORMAL

            

            self.fetch_all_button['state'] = tk.NORMAL

            self.update_status("測站清單載入完成!", "green")

            

        except requests.exceptions.RequestException as e:

            self.update_status(f"載入測站清單失敗:網路錯誤或API不穩定。({e})", "red")

        except json.JSONDecodeError as e:

            self.update_status(f"載入測站清單失敗:JSON 解析錯誤。({e})", "red")


    def start_fetch_single(self):

        target_sitename = self.sitename_combobox.get()

        if not target_sitename:

            self.update_status("錯誤:請從下拉選單中選擇一個測站。", "red")

            return

        

        self.update_status(f"正在查詢 '{target_sitename}' 的資料...", "blue")

        self.result_text.delete("1.0", tk.END)

        threading.Thread(target=self.fetch_data_single, args=(target_sitename,)).start()


    def start_fetch_all(self):

        self.update_status("正在查詢所有測站資料...", "blue")

        self.result_text.delete("1.0", tk.END)

        threading.Thread(target=self.fetch_data_all).start()


    # --- 已修正為您指定的輸出欄位 ---

    def fetch_data_single(self, target_sitename):

        try:

            response = requests.get(API_URL, timeout=10)

            response.raise_for_status()

            data = response.json()

            records = data.get('records', [])

            

            found = False

            for record in records:

                if target_sitename == record.get('sitename', ''):

                    found = True

                    # 依據使用者指定的欄位清單進行輸出

                    result_str = f"測站名稱: {record.get('sitename')}\n"

                    result_str += f"縣市: {record.get('county')}\n"

                    result_str += f"AQI: {record.get('aqi')}\n"

                    result_str += f"主要污染物: {record.get('pollutant')}\n"

                    result_str += f"狀態: {record.get('status')}\n"

                    result_str += f"二氧化硫: {record.get('so2')} ppb\n"

                    result_str += f"一氧化碳: {record.get('co')} ppm\n"

                    result_str += f"臭氧: {record.get('o3')} ppb\n"

                    result_str += f"臭氧8小時移動平均: {record.get('o3_8hr')} ppb\n"

                    result_str += f"懸浮微粒(PM10): {record.get('pm10')} μg/m³\n"

                    result_str += f"細懸浮微粒(PM2.5): {record.get('pm2.5')} μg/m³\n"

                    result_str += f"二氧化氮: {record.get('no2')} ppb\n"

                    result_str += f"氮氧化物: {record.get('nox')} ppb\n"

                    result_str += f"一氧化氮: {record.get('no')} ppb\n"

                    result_str += f"資料發布時間: {record.get('publishtime')}\n"

                    self.result_text.insert(tk.END, result_str)

                    break


            if found:

                self.update_status(f"成功查詢 '{target_sitename}' 資料。", "green")

            else:

                self.update_status(f"錯誤:找不到測站 '{target_sitename}'。", "red")


        except requests.exceptions.RequestException as e:

            self.update_status(f"HTTP 請求錯誤: {e}", "red")

        except json.JSONDecodeError as e:

            self.update_status(f"JSON 解析錯誤: {e}", "red")


    # --- 已修正為您指定的輸出欄位 ---

    def fetch_data_all(self):

        try:

            response = requests.get(API_URL, timeout=10)

            response.raise_for_status()

            data = response.json()

            records = data.get('records', [])


            if not records:

                self.update_status("錯誤:API 回應中無資料。", "red")

                return

            

            for record in records:

                # 依據使用者指定的欄位清單進行輸出

                result_str = f"測站名稱: {record.get('sitename')}\n"

                result_str += f"縣市: {record.get('county')}\n"

                result_str += f"AQI: {record.get('aqi')}\n"

                result_str += f"主要污染物: {record.get('pollutant')}\n"

                result_str += f"狀態: {record.get('status')}\n"

                result_str += f"二氧化硫: {record.get('so2')} ppb\n"

                result_str += f"一氧化碳: {record.get('co')} ppm\n"

                result_str += f"臭氧: {record.get('o3')} ppb\n"

                result_str += f"臭氧8小時移動平均: {record.get('o3_8hr')} ppb\n"

                result_str += f"懸浮微粒(PM10): {record.get('pm10')} μg/m³\n"

                result_str += f"細懸浮微粒(PM2.5): {record.get('pm2.5')} μg/m³\n"

                result_str += f"二氧化氮: {record.get('no2')} ppb\n"

                result_str += f"氮氧化物: {record.get('nox')} ppb\n"

                result_str += f"一氧化氮: {record.get('no')} ppb\n"

                result_str += f"資料發布時間: {record.get('publishtime')}\n"

                result_str += "-------------------------\n"

                self.result_text.insert(tk.END, result_str)


            self.update_status(f"成功查詢所有 {len(records)} 筆資料。", "green")


        except requests.exceptions.RequestException as e:

            self.update_status(f"HTTP 請求錯誤: {e}", "red")

        except json.JSONDecodeError as e:

            self.update_status(f"JSON 解析錯誤: {e}", "red")


    def update_status(self, message, color):

        self.status_label.config(text=message, fg=color)


if __name__ == "__main__":

    root = tk.Tk()

    app = AQIApp(root)

    root.mainloop()



{
    "fields": [
        {
            "id": "sitename",
            "type": "text",
            "info": {
                "label": "測站名稱"
            }
        },
        {
            "id": "county",
            "type": "text",
            "info": {
                "label": "縣市"
            }
        },
        {
            "id": "aqi",
            "type": "text",
            "info": {
                "label": "空氣品質指標"
            }
        },
        {
            "id": "pollutant",
            "type": "text",
            "info": {
                "label": "空氣污染指標物"
            }
        },
        {
            "id": "status",
            "type": "text",
            "info": {
                "label": "狀態"
            }
        },
        {
            "id": "so2",
            "type": "text",
            "info": {
                "label": "二氧化硫(ppb)"
            }
        },
        {
            "id": "co",
            "type": "text",
            "info": {
                "label": "一氧化碳(ppm)"
            }
        },
        {
            "id": "o3",
            "type": "text",
            "info": {
                "label": "臭氧(ppb)"
            }
        },
        {
            "id": "o3_8hr",
            "type": "text",
            "info": {
                "label": "臭氧8小時移動平均(ppb)"
            }
        },
        {
            "id": "pm10",
            "type": "text",
            "info": {
                "label": "懸浮微粒(μg\/m3)"
            }
        },
        {
            "id": "pm2.5",
            "type": "text",
            "info": {
                "label": "細懸浮微粒(μg\/m3)"
            }
        },
        {
            "id": "no2",
            "type": "text",
            "info": {
                "label": "二氧化氮(ppb)"
            }
        },
        {
            "id": "nox",
            "type": "text",
            "info": {
                "label": "氮氧化物(ppb)"
            }
        },
        {
            "id": "no",
            "type": "text",
            "info": {
                "label": "一氧化氮(ppb)"
            }
        },
        {
            "id": "wind_speed",
            "type": "text",
            "info": {
                "label": "風速(m\/sec)"
            }
        },
        {
            "id": "wind_direc",
            "type": "text",
            "info": {
                "label": "風向(degrees)"
            }
        },
        {
            "id": "publishtime",
            "type": "text",
            "info": {
                "label": "資料發布時間"
            }
        },
        {
            "id": "co_8hr",
            "type": "text",
            "info": {
                "label": "一氧化碳8小時移動平均(ppm)"
            }
        },
        {
            "id": "pm2.5_avg",
            "type": "text",
            "info": {
                "label": "細懸浮微粒移動平均值(μg\/m3)"
            }
        },
        {
            "id": "pm10_avg",
            "type": "text",
            "info": {
                "label": "懸浮微粒移動平均值(μg\/m3)"
            }
        },
        {
            "id": "so2_avg",
            "type": "text",
            "info": {
                "label": "二氧化硫移動平均值(ppb)"
            }
        },
        {
            "id": "longitude",
            "type": "text",
            "info": {
                "label": "經度"
            }
        },
        {
            "id": "latitude",
            "type": "text",
            "info": {
                "label": "緯度"
            }
        },
        {
            "id": "siteid",
            "type": "text",
            "info": {
                "label": "測站編號"
            }
        }
    ],
    "resource_id": "8d2f907f-bbb4-4fdf-8f08-8eabae15da45",
    "__extras": {
        "api_key": "9b651a1b-0732-418e-b4e9-e784417cadef"
    },
    "include_total": true,
    "total": "86",
    "resource_format": "object",
    "limit": "1000",
    "offset": "0",
    "_links": {
        "start": "\/api\/v2\/aqx_p_432?api_key=9b651a1b-0732-418e-b4e9-e784417cadef&limit=1000&format=JSON&sort=ImportDate desc",
        "next": "\/api\/v2\/aqx_p_432?api_key=9b651a1b-0732-418e-b4e9-e784417cadef&limit=1000&format=JSON&sort=ImportDate desc&offset=1000"
    },
    "records": [
        {
            "sitename": "基隆",
            "county": "基隆市",
            "aqi": "27",
            "pollutant": "",
            "status": "良好",
            "so2": "0.2",
            "co": "0.16",
            "o3": "30",
            "o3_8hr": "26",
            "pm10": "14",
            "pm2.5": "3",
            "no2": "2",
            "nox": "3.7",
            "no": "0.9",
            "wind_speed": "1",
            "wind_direc": "293",
            "publishtime": "2025\/08\/05 14:00:00",
            "co_8hr": "0.1",
            "pm2.5_avg": "4.5",
            "pm10_avg": "16",
            "so2_avg": "0",
            "longitude": "121.760056",
            "latitude": "25.129167",
            "siteid": "1"
        },
        {
            "sitename": "汐止",
            "county": "新北市",
            "aqi": "42",
            "pollutant": "",
            "status": "良好",
            "so2": "0.7",
            "co": "0.17",
            "o3": "18",
            "o3_8hr": "22",
            "pm10": "22",
            "pm2.5": "11",
            "no2": "13",
            "nox": "16",
            "no": "2.9",
            "wind_speed": "1",
            "wind_direc": "20",
            "publishtime": "2025\/08\/05 14:00:00",
            "co_8hr": "0.2",
            "pm2.5_avg": "10.5",
            "pm10_avg": "22",
            "so2_avg": "0",
            "longitude": "121.64081",
            "latitude": "25.06624",
            "siteid": "2"
        },

沒有留言:

張貼留言

Telegram +ESP32自動發報機

  Telegram   +ESP32自動發報機 這套系統是一個典型的 IoT(物聯網)架構 ,結合了遠端配置(Python)、通訊中介(MQTT)與硬體執行(ESP32)。 以下我為您拆解這兩支程式的核心運作原理: 一、 系統架構流程 Python 端 (控制台) :使用者輸入...