從韌體跨足軟體領域的學習歷程-2

python sql mssql 氣象公開資料

Meerkat 2024-03-03 19:07:35 ‧ 836 瀏覽

分享至

在「從韌體跨足軟體領域的學習歷程-1」練習了基本的Python與SQL互動，為了更加熟悉Python各種語法，決定繼續做更進一步的練習。

本次目標:從全台氣象觀測站取得氣象資訊

資料來源:中央氣象署-開放資料平臺
登入後先取得「授權碼」
有了授權碼後就能到這裡測試各類API
經過一番研究後發現「中央氣象署-開放資料平台」將全台氣象觀測站分為「自動觀測站」與「人工觀測站」
填寫授權碼並設定回傳前100筆資料，測試一下能取到什麼資料
- 自動觀測站
- 人工觀測站
確認資料符合需求後，開始建立資料表，沿用上篇的資料庫「NewsHandle」，創建資料表

SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
CREATE TABLE [dbo].[Weather](
	[wid] [nvarchar](max) NOT NULL,
	[name] [nvarchar](max) NOT NULL,
	[country] [nvarchar](max) NOT NULL,
	[town] [nvarchar](max) NOT NULL,
	[y] [decimal](9, 6) NOT NULL,
	[x] [decimal](9, 6) NOT NULL,
	[obstime] [datetime2](0) NOT NULL,
	[weather] [nvarchar](max) NOT NULL,
	[rainnow] [decimal](4, 1) NOT NULL,
	[wd] [decimal](4, 1) NOT NULL,
	[ws] [decimal](4, 1) NOT NULL,
	[t] [decimal](4, 1) NOT NULL,
	[rh] [int] NOT NULL,
	[ap] [decimal](5, 1) NOT NULL,
	[dailyhigh] [decimal](4, 1) NOT NULL,
	[dailylow] [decimal](4, 1) NOT NULL,
	[insertDT] [datetime2](0) NOT NULL
) ON [PRIMARY] TEXTIMAGE_ON [PRIMARY]
GO

為了方便後續Python將資料存入資料表，寫一個Store Procedure供Python呼叫

SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
CREATE procedure [dbo].[sp_InsertWeather](
    @wid       nvarchar(max) = null,
    @name      nvarchar(max) = null,
    @country   nvarchar(max) = null,
    @town      nvarchar(max) = null,
    @x         decimal(9, 6) = null,
    @y         decimal(9, 6) = null,
    @obstime   datetime2(0)  = null,
    @weather   nvarchar(max) = null,
    @rainnow   decimal(4, 1) = null,
    @wd        decimal(4, 1) = null,
    @ws        decimal(4, 1) = null,
    @t         decimal(4, 1) = null,
    @rh        int           = null,
    @ap        decimal(5, 1) = null,
    @dailyhigh decimal(4, 1) = null,
    @dailylow  decimal(4, 1) = null,
    @insertDT  datetime2(0)  = null
)
as
begin
    declare @now datetime2(0) = GETDATE(); 
    
    insert into Weather(wid, name, country, town, x, y, obstime, weather, rainnow, wd, ws, t, rh, ap, dailyhigh, dailylow, insertDT)
    values(@wid, @name, @country, @town, @x, @y,  @obstime, @weather, @rainnow, @wd, @ws, @t, @rh, @ap, @dailyhigh, @dailylow, @now);
end
GO

資料表與Procedure後開始實作Python

from datetime import datetime
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import requests
import time
import json
import pyodbc

# 出錯時發送信件至信箱
def send_email(subject, body):
    sender_email = "chu@gmail.com"
    receiver_email = "chu@gmail.com"
    app_password = ""

    message = MIMEMultipart()
    message["From"] = sender_email
    message["To"] = receiver_email
    message["Subject"] = subject
    body_text = MIMEText(body, "plain")
    message.attach(body_text)

    try:
        with smtplib.SMTP("smtp.gmail.com", 587) as server:
            server.starttls()
            server.login(sender_email, app_password)
            server.sendmail(sender_email, receiver_email, message.as_string())
        print("電子郵件通知發送成功。")
    except Exception as e:
        print(f"發送電子郵件通知時出錯：{str(e)}")

# 呼叫API
def call_api(open_api):
    try:
        packet = None
        while packet is None:
            try:
                response = requests.get(open_api)
                if response.status_code == 200:
                    data = response.json()
                    packet = json.loads(json.dumps(data, ensure_ascii=False))
                    return packet
                else:
                    print(f"請求發生錯誤，HTTP狀態碼：{response.status_code}")
                    time.sleep(3)
            except requests.exceptions.RequestException as e:
                print(f"請求發生錯誤：{str(e)}")
                time.sleep(3)
    except Exception as e:
        error_subject = "Python腳本中的錯誤"
        error_message = f"腳本中發生錯誤：\n\n{str(e)}"
        send_email(error_subject, error_message)

# 資料處理
def proc_weather_data(data_list):
    packet = []
    for station in data_list:
        stationName = station["StationName"]  # 測站名稱
        stationId = station["StationId"]  # 測站ID
        stationCountyName = station["GeoInfo"]["CountyName"]  # 測站所屬縣市
        stationTownName = station["GeoInfo"]["TownName"]  # 測站所屬鄉鎮區
        stationLat = station["GeoInfo"]["Coordinates"][1]["StationLatitude"]  # 緯度
        stationLng = station["GeoInfo"]["Coordinates"][1]["StationLongitude"]  # 經度
        obsTime = station["ObsTime"]["DateTime"]  # 觀測時間
        weather = station["WeatherElement"]["Weather"]  # 天氣現象
        rainNow = station["WeatherElement"]["Now"]["Precipitation"]  # 現在降雨量
        windDirection = station["WeatherElement"]["WindDirection"]  # 風向
        windSpeed = station["WeatherElement"]["WindSpeed"]  # 風速
        airTemperature = station["WeatherElement"]["AirTemperature"]  # 氣溫
        relativeHumidity = station["WeatherElement"]["RelativeHumidity"]  # 相對濕度
        airPressure = station["WeatherElement"]["AirPressure"]  # 大氣壓力
        dailyHigh = station["WeatherElement"]["DailyExtreme"]["DailyHigh"][
            "TemperatureInfo"
        ][
            "AirTemperature"
        ]  # 最高溫
        dailyLow = station["WeatherElement"]["DailyExtreme"]["DailyLow"][
            "TemperatureInfo"
        ][
            "AirTemperature"
        ]  # 最低溫

        packet.append(
            {
                "WID": stationId,
                "Name": stationName,
                "County": stationCountyName,
                "Town": stationTownName,
                "Lat": stationLat,
                "Lng": stationLng,
                "ObsTime": obsTime,
                "Weather": weather,
                "RainNow": rainNow,
                "WD": windDirection,
                "WS": windSpeed,
                "T": airTemperature,
                "RH": relativeHumidity,
                "AP": airPressure,
                "DailyHigh": dailyHigh,
                "DailyLow": dailyLow,
            }
        )
    return packet

# 呼叫Procedure將資料存入資料庫
def save_to_NewsHandle(data_list, server_config_path="server.json"):
    def handle_error(function_name, error):
        error_subject = f"Error in {function_name} function"
        error_message = (
            f"An error occurred in the {function_name} function:\n\n{str(error)}"
        )
        send_email(error_subject, error_message)
        print(f"An error occurred in {function_name} function: {str(error)}")

    try:
        with open(server_config_path, "r") as config_file:
            server_config = json.load(config_file)

        connection_string = (
            f"DRIVER={{{server_config['drv']}}};"
            f"SERVER={server_config['srv']};"
            f"DATABASE={server_config['db']};"
            f"UID={server_config['uid']};"
            f"PWD={server_config['pwd']}"
        )
        connection = pyodbc.connect(connection_string)
        cursor = connection.cursor()

        query = """
            exec NewsHandle.dbo.sp_InsertWeather
                @wid = ?, 
                @Name = ?, 
                @country = ?, 
                @town = ?, 
                @x = ?, 
                @y = ?, 
                @obstime = ?, 
                @weather = ?, 
                @rainnow = ?, 
                @wd = ?, 
                @ws = ?, 
                @t = ?, 
                @rh = ?, 
                @ap = ?, 
                @dailyhigh = ?, 
                @dailylow = ?;
        """
        new_order = [
            "WID",
            "Name",
            "County",
            "Town",
            "Lat",
            "Lng",
            "ObsTime",
            "Weather",
            "RainNow",
            "WD",
            "WS",
            "T",
            "RH",
            "AP",
            "DailyHigh",
            "DailyLow"
        ]
        transformed_data_list = [
            tuple(data[field] for field in new_order) for data in data_list
        ]
        cursor.executemany(query, transformed_data_list)
        connection.commit()
    except Exception as e:
        handle_error("exec_procedure", e)
    finally:
        if cursor:
            cursor.close()
        if connection:
            connection.close()

# 主程式
if __name__ == "__main__":
    print(f'開始更新天氣資訊 -> {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}')

    weather = []
    api_key = ""
    api_name = ["O-A0001-001", "O-A0003-001"]
    for name in api_name:
        stations = call_api(
            f"https://opendata.cwa.gov.tw/api/v1/rest/datastore/{name}?Authorization={api_key}&format=JSON"
        )["records"]["Station"]
        weather.extend(proc_weather_data(stations))

    save_to_NewsHandle(weather)

    print(f'天氣資訊更新完成 -> {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}')
    print("-" * 50)
    print()

成果

心得
有了之前的經驗後，此次練習下來花費的時間明顯少了許多而熟練度也提升不少，其中對於Python與資料庫之間各種型態的轉換也越發熟悉了。這次之所以選擇天氣資訊做練習，除了中央氣象署的氣象資料開放平台中有提供許多資料可以免費取得，另一方面是因為未來我想要練習寫前端，而氣象資訊在前端比較好做應用練習，希望接下來的練習可以越來越上手~

從韌體跨足軟體領域的學習歷程-1