2024年10月4日 星期五

MicroPython 學習筆記 : 串接物聯網雲端平台 ThingSpeak (二)

今天是山陀兒颱風假第四天, 昨晚暴風圈北移後雨停了, 我外出去餵河堤小咪發現外面滿目瘡痍, 路樹倒了很多, 環保局漏夜清理中, 原以為週五該上班了, 但看到這種情形難怪邁邁要放這史無前例的四天颱風假, 這次的 17 級颱風真的刷新了高雄人自賽洛瑪以來對風災的記憶. 

昨晚註冊了 ThingSpeak 新帳號後, 今天早上繼續測試 ESP32 串接其 API 的實驗, 任務是從  OpenWeatherMap 取得高雄的即時溫溼度與大氣壓資料後傳送至 ThingSpeak 儲存與顯示. 本系列之前的文章參考 :


在前一篇測試中已建立一個 ThingSpeak 通道 (即資料表) 來儲存高雄的氣象資料, 此通道資料表含有下列三個欄位 :
  • field1 : 溫度 (Temperature)
  • field2 : 濕度 (Humidity)
  • field3 : 氣壓 (Pressure)
本篇的測試目標是要將 OpenWeatherMap 取得的溫溼度與氣壓數值透過呼叫 ThingSpeak API 將它們存入通道的欄位中, 關於從 OpenWeatherMap 網站取得氣象資料的做法參考 :



3. 使用 ThingSpeak REST API 上傳資料 : 

ThingSpeak 提供 REST 與 MQTT 兩種 API 讓使用者上傳感測器資料, 官方文件參考 : 


REST API 就是透過網址 URL 向伺服器請求, URL 格式如下 :

https://api.thingspeak.com/update?api_key=<API_KEY>&field1=<V1>&field2=<V2> .....

此 API 會將欄位資料存入通道的欄位裡, 其中 API_KEY 是該通道的 Write API key, field1, field2, filed3 ... 則是該通道的 1~8 欄位, V1, V2, V3, ... 是要寫入各欄位之值.

先匯入 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'

設定 api_key 變數製作 API 網址字串, 先用一組固定資料測試看看 : 

>>> api_key='在此填入 ThingSpeak WRITE API key'     
>>> temperature=25.6    
>>> humidity=67     
>>> pressure=998     
>>> url=f'https://api.thingspeak.com/update?api_key={api_key}&field1={temperature}>&field2={humidity}&field3={pressure}'      
>>> url  
'https://api.thingspeak.com/update?api_key=<API_KEY>&field1=25.6>&field2=67&field3=998'

然後匯入 urequests, 呼叫其 get() 函式對此 API 提出 GET 請求 :

>>> r=urequests.get(url)     
>>> r.text   
'2'

傳回值 2 表示是第二次請求, 因為之前我有先用官網教學文件中的 URL 範例提出第一次請求, 直接貼到瀏覽器網址列, 網頁顯示 1 表示是第一次請求 :

https://api.thingspeak.com/update?api_key=<API_KEY>&field1=0 



也可以用 xtools.webhook_get() 呼叫 API : 

>>> r=xtools.webhook_get(url)   
invoking webhook
Webhook invoked
>>> r.text     
'3'

傳回 3 表示這是第三次請求, 這時檢視 Dashboard, 會發現此通道已有此三欄位之資料圖形 :




免費帳戶的 API 請求限制是必須間隔 15 秒以上, 接下來使用 xtools 的 random_in_range() 函式來產生虛擬的氣象資料來跑一跑看看 :

>>> import time    
>>> while True:
    temperature=xtools.random_in_range(10, 40)
    humidity=xtools.random_in_range(50, 100)
    pressure=xtools.random_in_range(600, 1200)
    url=f'https://api.thingspeak.com/update?api_key={api_key}&field1={temperature}&field2={humidity}&field3={pressure}'
    r=urequests.get(url)
    print(r.text)
    time.sleep(15)
    
4
5
6
7
...
390
391
392
393
394
395

跑了 395 筆後結果如下 :




右邊一段空白是我外出時筆電休眠所致. 

接下來要從 OpenWeatherMap 取得氣象資料, 在之前的測試中我們已經將擷取即時氣象資料的程式碼寫成如下的函式 get_weather_now() :

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

此函式若成功取得氣象資料會傳回一個字典, 失敗傳回 None, 成功的話只要用傳回值字典的 temperature, humidity, 與 pressure 鍵即可取得傳送給 ThingSpeak 所需要的溫溼度與大氣壓等欄位參數. 因此處會用到 ThingSpeak 與 OpenWeatherMap 兩個服務的 API key, 所以我將 OpenWeatherMap 的 API key 變數名稱設為 weather_api_key 以資分別 :

>>> api_key='在此填入 ThingSpeak WRITE API key'   
>>> weather_api_key='在此輸入 OpenWeatherMap 的 API keye'      
>>> city='Kaohsiung'     
>>> country='TW'    
>>> r=get_weather_now(country, city, weather_api_key)        
>>> r['temperature']   
28.91
>>> r['humidity']     
83
>>> r['pressure']    
1010  

然後改寫上面的無窮迴圈, 改為填入從 OpenWeatherMap 取回的即時氣象資料 : 

>>> import time   
>>> 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?api_key={api_key}&field1={temperature}&field2={humidity}&field3={pressure}'
        r=urequests.get(url)
        print(r.text)
        time.sleep(15)
    else:
        print('Fail to get weather data.')
        
0
515
516
517
518
...
530
531
532
533
534

結果如下 :




可見與之前的隨機氣象資料比起來, 實際的氣象數值變化不大, 幾乎是一條水平線, 時間要拉長後才看得出起伏變化, 或者把取得氣象資料與上傳 ThingSpeak 的週期加長 (例如 60 秒或 5 分鐘抓一次) 亦可, 畢竟氣象數據變化並沒有那麼快速. 

因為我在前一篇測試中有設定此 ThingSpeak 通道有公開分享, 所以在 Dashboard 上按 Public View 頁籤就可以看到公開檢視的網頁, 上方網址列 URL 如下 :


URL 後面的數值就是此通道的 Channel ID (即資料表編號). 




以上在互動環境完成測試功能正常, 就可以將程式碼寫成如下的 weather_app.py 檔案 :

# 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?api_key={thingspeak_api_key}&field1={temperature}&field2={humidity}&field3={pressure}'
            r=urequests.get(url)
            print(r.text)
            time.sleep(15)
        else:
            print('Fail to get weather data.')

if __name__ == '__main__':  
    main()  
      
主程式是放在 main() 中, 它會先從 config.py 中讀取 THINGSPEAK_WRITE_API_KEY 與 WEATHER_API_KEY 這兩個金鑰字串, 然後進入一個無窮迴圈每隔 15 秒呼叫 get_weather_now() 函式向 OpenWeatherMap 擷取高雄氣象資料, 然後呼叫 ThingSpeak API 將溫溼度與氣壓寫入通道欄位中. 

然後寫一個主程式 main.py 負責連網 : 

import xtools    
import config
import weather_app    

ip=xtools.connect_wifi(config.SSID, config.PASSWORD)
mac=xtools.get_id()
print('ip: ', ip)
print('mac: ', mac)
if not ip: # 連線失敗傳回 None
    print('無法連線 WiFi 基地台')
else:      # 連線成功傳回 ip 字串: 執行 app
    weather_app.main() # 連線成功才執行 App 程式

此主程式利用 xtools.connect_wifi() 連線 WiFi 基地台, 連網成功會傳回 IP 字串, 這時就呼叫執行上面 weather_app.py 中的 main() 函式; 連網失敗傳回 None 就印出連線失敗訊息而結束程式. 

不論是 WiFi 連網的基地台帳密或 OpenWeatherMap 與 ThingSpeak 的 API key 都寫入 config.py 裡的變數中, 再於程式中讀取出來 : 




以上程式架構原理參考 :


簡言之, MicroPython 韌體其實就是一個微作業系統, 它擁有一個檔案系統與精簡版的 Python 3 函式庫, 以及一個 Python 互動執行環境. 當開發板硬啟動時 (即按下 Reset 鈕或插上電源), 此微作業系統會先去執行根目錄下的 boot.py, 接著執行 main.py (如果有的話). 

應用程式可以直接寫在 main.py, 也可以寫在各個 app.py 程式檔中, 可於 main.py 裡切換鑰呼叫哪個 app.py 中的 main(). boot,py 的作用類似 Arduino 程式中的 setup(); 而 main() 或 app.py 裡的無窮迴圈則相當於 Arduino 裡的 loop(). 


4. 使用 ThingSpeak 的小工具 (Widget) : 

ThingSpeak 除了預設用折線圖來顯示通道中各欄位值的變化外, 還提供了三個小工具來呈現資料內容 : 
  • Gauge 量表 : 以指針表顯示欄位的即時數值
  • Numeric Display 數位顯示器 : 直接顯示即時欄位值
  • Lamp Indicator : 可設定條件, 當欄位值符合時點亮指示燈號
在 Dashboard 按 "Add widgets" 鈕新增小工具 :




在彈出視窗中點選鑰新增的小工具類型, 例如 Gauge, 按 Next : 




於彈出視窗中勾選鑰顯示的欄位 (此處選擇溫度), 填寫名稱與單位等, 其中 range 可設定量表的數值範圍以指定之顏色顯示 (此處設 35~100 表示高溫區), 按 Create 新增 : 




這時 Dashboard 就會出現一個即時的溫度欄位量表 : 



重複上面操作添加 Numeric Display 小工具來顯示欄位 3 大氣壓 (單位 hPa) : 




結果如下 : 




最後新增 Lamp Indicator, 設定當 filed2 的溼度大於 90% 時讓顯示燈亮 :




結果如下 :




因目前濕度尚未達到 90 以上門檻, 故指示燈未亮. 至此全部 Dahboard 外觀如下 :




以上測試之程式碼壓縮檔可從 GitHub 下載 (使用前要先編籍 config.py) :


張貼者： 於
標籤： , , ,

沒有留言 :

張貼留言

訂閱： 張貼留言 ( Atom )