氣象機器人 (2) - 目前氣象資訊
延續「氣象機器人 (1) - 雷達回波與地震資訊」文章，這篇教學會讓 LINE 氣象機器人串接目前氣象資訊，只要使用 LINE 預設的功能提供地址，就能回傳該地址的即時天氣資訊。
取得地址資訊
延續「氣象機器人 (1) - 雷達回波與地震資訊」的範例，在 if 判斷式裡面加上 message type 為 location 的判斷，如果是 location，就取出 address 的內容，並使用 replace 將「台」換成「臺」( 因為氣象局都是用「臺」，但 LINE 的 location 都用「台」 )
if 'message' in json_data['events'][0]:
if json_data['events'][0]['message']['type'] == 'location': # 如果 message 的類型是地圖 location
address = json_data['events'][0]['message']['address'].replace('台','臺') # 取出地址資訊，並將「台」換成「臺」
print(address)
if json_data['events'][0]['message']['type'] == 'text':
text = json_data['events'][0]['message']['text']
if text == '雷達回波圖' or text == '雷達回波':
reply_image(f'https://cwbopendata.s3.ap-northeast-1.amazonaws.com/MSC/O-A0058-003.png?{time.time_ns()}', reply_token, access_token)
elif text == '地震資訊' or text == '地震':
msg = earth_quake()
push_message(msg[0], user_id, access_token)
reply_image(msg[1], reply_token, access_token)
else:
reply_message(text, reply_token, access_token)
完成後重新執行，設定新的 ngrok 網址為 Webhook，驗證成功後 ( 有時因為快取的緣故，需要重新產生兩次才會正常運作 )，LINE 裡面傳送地址資訊，Colab 裡就會看到出現地址的內容。
爬取目前氣象資訊
參考「爬取現在天氣」教學，從氣象資料開放平臺裡，找到「自動氣象站-氣象觀測資料」和「局屬氣象站-現在天氣觀測報告」兩個資料集，複製 JSON 網址 ( 「自動氣象站-氣象觀測資料」包含鄉鎮區域的資訊，「局屬氣象站-現在天氣觀測報告」則是以縣市為主的天氣資訊 )。
資料連結 ( 需要登入才看得到 JSON 連結 )
建立 current_weather 函式，函式包含一個 address 參數，負責承接 LINE 收到的地址資訊，程式相關重點如下：
- 定義爬取資料的函式 get_data，篩選出 JSON 資料裡的縣市名稱、鄉鎮行政區、氣溫、相對濕度和累積雨量。
- 由於可能會遇到找不到行政區的狀況，所以會額外獨立縣市名稱作為其中一個 key。
- 如果是以縣市名稱為 key，則內容是採用該縣市裡鄉鎮行政區的平均數值 ( 例如新北市有十個區，每個區有各自的溫度，新北市的溫度就是這十個區的平均 )。
- 計算串列裡的平均數使用標準函式庫的 statistics.mean 方法。
- 定義 check_data 函式，處理小於 0 的數值 ( 因有時數據會出現 -99 之類的數值，所以將小於 0 的數值回傳為 False )。
- 定義 msg_content 函式，產生回傳的訊息。
# 目前天氣函式
def current_weather(address):
city_list, area_list, area_list2 = {}, {}, {} # 定義好待會要用的變數
msg = '找不到氣象資訊。' # 預設回傳訊息
# 定義取得資料的函式
def get_data(url):
w_data = requests.get(url) # 爬取目前天氣網址的資料
w_data_json = w_data.json() # json 格式化訊息內容
location = w_data_json['cwbopendata']['location'] # 取出對應地點的內容
for i in location:
name = i['locationName'] # 測站地點
city = i['parameter'][0]['parameterValue'] # 縣市名稱
area = i['parameter'][2]['parameterValue'] # 鄉鎮行政區
temp = check_data(i['weatherElement'][3]['elementValue']['value']) # 氣溫
humd = check_data(round(float(i['weatherElement'][4]['elementValue']['value'] )*100 ,1)) # 相對濕度
r24 = check_data(i['weatherElement'][6]['elementValue']['value']) # 累積雨量
if area not in area_list:
area_list[area] = {'temp':temp, 'humd':humd, 'r24':r24} # 以鄉鎮區域為 key，儲存需要的資訊
if city not in city_list:
city_list[city] = {'temp':[], 'humd':[], 'r24':[]} # 以主要縣市名稱為 key，準備紀錄裡面所有鄉鎮的數值
city_list[city]['temp'].append(temp) # 記錄主要縣市裡鄉鎮區域的溫度 ( 串列格式 )
city_list[city]['humd'].append(humd) # 記錄主要縣市裡鄉鎮區域的濕度 ( 串列格式 )
city_list[city]['r24'].append(r24) # 記錄主要縣市裡鄉鎮區域的雨量 ( 串列格式 )
# 定義如果數值小於 0，回傳 False 的函式
def check_data(e):
return False if float(e)<0 else float(e)
# 定義產生回傳訊息的函式
def msg_content(loc, msg):
a = msg
for i in loc:
if i in address: # 如果地址裡存在 key 的名稱
temp = f"氣溫 {loc[i]['temp']} 度，" if loc[i]['temp'] != False else ''
humd = f"相對濕度 {loc[i]['humd']}%，" if loc[i]['humd'] != False else ''
r24 = f"累積雨量 {loc[i]['r24']}mm" if loc[i]['r24'] != False else ''
description = f'{temp}{humd}{r24}'.strip('，')
a = f'{description}。' # 取出 key 的內容作為回傳訊息使用
break
return a
try:
# 因為目前天氣有兩組網址，兩組都爬取
code = '你的氣象資料授權碼'
get_data(f'https://opendata.cwb.gov.tw/fileapi/v1/opendataapi/O-A0001-001?Authorization={code}&downloadType=WEB&format=JSON')
get_data(f'https://opendata.cwb.gov.tw/fileapi/v1/opendataapi/O-A0003-001?Authorization={code}&downloadType=WEB&format=JSON')
for i in city_list:
if i not in area_list2: # 將主要縣市裡的數值平均後，以主要縣市名稱為 key，再度儲存一次，如果找不到鄉鎮區域，就使用平均數值
area_list2[i] = {'temp':round(statistics.mean(city_list[i]['temp']),1),
'humd':round(statistics.mean(city_list[i]['humd']),1),
'r24':round(statistics.mean(city_list[i]['r24']),1)
}
msg = msg_content(area_list2, msg) # 將訊息改為「大縣市」
msg = msg_content(area_list, msg) # 將訊息改為「鄉鎮區域」
return msg # 回傳 msg
except:
return msg # 如果取資料有發生錯誤，直接回傳 msg
完成後，在最上方 import statistics 函式庫，並修改 if 判斷式裡的內容。
from flask_ngrok import run_with_ngrok
from flask import Flask, request
from linebot import LineBotApi, WebhookHandler
from linebot.models import MessageEvent, TextMessage, TextSendMessage
import requests, json, time, statistics # import statistics 函式庫
app = Flask(__name__)
access_token = '你的 LINE Channel access token'
channel_secret = '你的 LINE Channel secret'
@app.route("/", methods=['POST'])
def linebot():
body = request.get_data(as_text=True)
try:
line_bot_api = LineBotApi(access_token)
handler = WebhookHandler(channel_secret)
signature = request.headers['X-Line-Signature']
handler.handle(body, signature)
json_data = json.loads(body)
reply_token = json_data['events'][0]['replyToken']
user_id = json_data['events'][0]['source']['userId']
print(json_data)
if 'message' in json_data['events'][0]:
if json_data['events'][0]['message']['type'] == 'location':
address = json_data['events'][0]['message']['address'].replace('台','臺')
# 回覆爬取到的相關氣象資訊
reply_message(f'{address}\n\n{current_weather(address)}', reply_token, access_token)
print(address)
if json_data['events'][0]['message']['type'] == 'text':
text = json_data['events'][0]['message']['text']
if text == '雷達回波圖' or text == '雷達回波':
reply_image(f'https://cwbopendata.s3.ap-northeast-1.amazonaws.com/MSC/O-A0058-003.png?{time.time_ns()}', reply_token, access_token)
elif text == '地震資訊' or text == '地震': # 如果是地震相關的文字
msg = earth_quake() # 爬取地震資訊
push_message(msg[0], user_id, access_token) # 傳送地震資訊 ( 用 push 方法，因為 reply 只能用一次 )
reply_image(msg[1], reply_token, access_token) # 傳送地震圖片 ( 用 reply 方法 )
else:
reply_message(text, reply_token, access_token) # 如果是一般文字，直接回覆同樣的文字
except:
print('error')
return 'OK'
if __name__ == "__main__":
run_with_ngrok(app)
app.run()
'''
下方接續其他函式
'''
完成後重新執行，設定新的 ngrok 網址為 Webhook，驗證成功後 ( 有時因為快取的緣故，需要重新產生兩次才會正常運作 )，LINE 裡面傳送地址資訊，Colab 裡就會看到該地址的即時天氣資訊。
小結
目前的氣象機器人已經可以串接地震資訊、雷達回波和即時天氣狀況，接下來的幾篇會繼續介紹如何串接天氣預報和空氣品質資訊。
繼續閱讀：
意見回饋
如果有任何建議或問題，可傳送「意見表單」給我，謝謝～