在前一篇測試中我們使用 HTTP GET 方法, 利用 urequests.get() 呼叫 ThingSpeak API 將氣象資料傳送到 ThingSpeak 通道資料表中儲存, 本篇則是要改用 POST 方法, 順便測試原 xtools 函式庫中的 webhook.post() 函式, 它原本是呼叫 xrequests.post(), 但我將其改為 urequests.post(), 如果沒問題就可以拿掉 xrequests.py 了.
本系列之前的文章參考 :
本篇測試使用修訂過的 xtools 函式庫, 關於 xtools 參考 :
更多 MicroPython 文章參考 :
以下實驗是在燒錄 MicroPython v1.23 版韌體的 ESP32-WROOM 開發板上進行. 為了不干擾前一篇實驗的運作, 這次改用 Gmail 申請的 ThingSpeak 帳號來測試, 建立了一個 '高雄的天氣2' 通道, 同樣有 field1 (溫度), field2 (濕度), 與 field3 (氣壓) 三個欄位 :
點選 API key 頁籤將 WRITE API key 記下來備用 (存入 config.py).
5. 使用 HTTP POST 方法呼叫 ThingSpeak API :
先匯入 config 與 xtools 模組呼叫 xtools.coneect_wifi() 連上網路 :
>>> import xtools
>>> import config
>>> ip=xtools.connect_wifi(config.SSID, config.PASSWORD)
network config: ('192.168.50.9', '255.255.255.0', '192.168.50.1', '192.168.50.1')
'192.168.50.9'
我已經將 ThingSpeak 的通道寫入 API key 儲存於 config.py 的 THINGSPEAK_API_KEY 字串變數中, 只要將其取出即可 :
>>> thingspeak_api_key=config.THINGSPEAK_API_KEY
接著製作 POST 方法要傳送的資料字典, 將 API key 放入 api_key 鍵, 欄位資料分別放入 filed1~filed8 鍵 :
>>> data={
'api_key': thingspeak_api_key,
'field1': 27,
'field2': 87,
'field3': 980
}
在 CPython 中可以在呼叫 requests.post() 時將字典直接傳給 data 參數, 但在 MicroPython 不能這麼做, 因為 MicroPython 的 urequests 模組要求傳給 post() 函式的 data 參數必須是字串或 bytes 類型, 故此處須先呼叫 xtools 的 urlencode() 函式先將 data 字典轉成 URL 字串, 然後呼叫 encode() 以 utf-8 格式把 URL 字串編碼為 bytes 類型資料 :
>>> payload=xtools.urlencode(data).encode('utf-8')
這樣就可以匯入 urequests 呼叫其 post() 方法將編碼後的 payload 傳給 data 參數, ThingSpeak 的 POST API 網址如下 :
# https://api.thingspeak.com/update
>>> import urequests
>>> url='https://api.thingspeak.com/update'
>>> r=urequests.post(url, data=payload)
>>> r.text
'1'
傳回數值表示請求成功, 1 表示這是向 ThingSpeak 傳送的第一則訊息, 免費帳戶一年內有 300 萬則訊息額度可用. 這時 Dashbord 會在欄位的折線圖上顯示這個數據 :
改用 xtools 的 webhook_post(url, value) 也是可以的, 只要將編碼過的 payload 傳給它的第二參數 value 即可 :
>>> r=xtools.webhook_post(url, payload)
invoking webhook
Webhook invoked
>>> r.text
'2'
傳回 2 表示第二則訊息也傳成功了, 這時 Dashboard 上會顯示兩個相同值的點 :
注意, xtools 中的 webhook_post() 原本是呼叫原作者自訂 xrequests 模組的 post(), 我已將其改為 MicroPython 內建的 urequests,post() 且傳回回應值 r :
def webhook_post(url, value):
print("invoking webhook")
r = urequests.post(url, data=value)
if r is not None and r.status_code == 200:
print("Webhook invoked")
else:
print("Webhook failed")
show_error()
return r
以上測試確認 webhook_post() 改用 urequests.post() 沒有問題, 應此在新版 xtools 函式庫中我已移除 xrequests.py 這個檔案, 且 xtools.py 中取消匯入 xrequests.py.
我將上一篇的 weather_app.py 改成如下的 POST 版 :
# weather_app.py
import xtools
import config
import time
import urequests
import ujson
def get_weather_now(country, city, api_key):
url=f'https://api.openweathermap.org/data/2.5/weather?q={city},{country}&units=metric&lang=zh_tw&appid={api_key}'
try:
res=urequests.get(url)
data=ujson.loads(res.text)
if data['cod']==200: # 注意是數值
ret={'geocode': data['id'],
'description': data['weather'][0]['description'],
'temperature': data['main']['temp'],
'min_temperature': data['main']['temp_min'],
'max_temperature': data['main']['temp_max'],
'pressure': data['main']['pressure'],
'humidity': data['main']['humidity']}
return ret
else:
return None
except Exception as e:
return None
def main():
thingspeak_api_key=config.THINGSPEAK_WRITE_API_KEY
weather_api_key=config.WEATHER_API_KEY
city='Kaohsiung'
country='TW'
while True:
r=get_weather_now(country, city, weather_api_key)
if r != None:
temperature=r['temperature']
humidity=r['humidity']
pressure=r['pressure']
url=f'https://api.thingspeak.com/update'
data={
'api_key': thingspeak_api_key,
'field1': temperature,
'field2': humidity,
'field3': pressure
}
payload=xtools.urlencode(data).encode('utf-8')
r=urequests.post(url, data=payload)
print(r.text)
time.sleep(15)
else:
print('Fail to get weather data.')
if __name__ == '__main__':
main()
修改之處為藍色粗體部分, 將程式上傳後按 Reset 鈕重啟系統, 就可看到新帳號上的 '高雄的天氣2' 通道每 15 秒更新的氣象資料統計圖 :
開啟 Sharing 為 "Share chnnel view with everyone" :
不過此通道只是測試 POST 用, 跑一個下午後就會關掉了.
沒有留言 :
張貼留言