ラズパイからLINE通知 - ハードウェア技術者のスキルアップ日誌

今回から見守りカメラに必要な機能を作っていきたいと思います。

masaeng.hatenablog.com

まず初めはラズパイからLINE通知をする方法です。
温度センサーをラズパイにつないで、測定した温度を定期的にLINEに通知するプログラムを作っていきます。

LINE notifyに登録

LINEから自分のアカウントにメッセージを送信するにはLINE notifyというサービスを利用します。早速登録してみましょう。
f:id:masashi_k:20200831230219p:plain

①右上の"ログイン"ボタンからログイン
②ログイン後にマイページを開く

f:id:masashi_k:20200831231745p:plain

➂アクセストークンの発行から"トークンを発行する"をクリック

f:id:masashi_k:20200831231854p:plain

④トークン名を入力し、通知を送信するトークルームを選択後、"発行する"をクリック
トークルームは「1:1でLINE Notifyから通知を受け取る」でOKです。
f:id:masashi_k:20200831232011p:plain

⑤トークンが発行されるため、控えておく

f:id:masashi_k:20200831232653p:plain

⑥自分のスマホに"LINE notify"を招待しておく

これで登録は完了です。

ラズパイから通知コマンドを送信

発行されたトークンを使って自分のアカウントにラズパイからメッセージを送信します。次のコマンドで実現できます。

$ curl https://notify-api.line.me/api/notify \
     -X POST \
     -H "Authorization: Bearer xxxxxxx" \    　# "xxxxxxx"にトークンを入力
     -F "message=Hello world!"                 # Hello world!の部分が送信されるメッセージ

送信結果

f:id:masashi_k:20200831234227p:plain
一番下にHello world!というメッセージが届いています。

次に画像を送ってみます。カメラで撮影したスクショを送信する想定です。
コマンドは以下の通り。

$ curl https://notify-api.line.me/api/notify \
     -X POST \
     -H "Authorization: Bearer xxxxxxx" \    # xxxxxxxにトークンを入力
     -F "message=test" \                     # testの部分が送信されるメッセージ
     -F "imageFile=@PATH_TO_PICTURE"         #PATH_TO_PICTUREにラズパイ上の画像のパスを入力

画像だけを送ることは出来ないようで、messageの部分には何か入れる必要があります。

送信結果

f:id:masashi_k:20200831234902p:plain

送信した"test"というメッセージとラズパイのアイコンが届いていることが確認できます。

温度センサーの測定データをLINEで通知

最後に温度センサーの測定データを通知してみます。これまではcurlコマンドを使ってきましたが、プログラムを作る上ではPython上で命令を出せた方がいいので、pythonで通知をさせる方法を調べてみました。

温度センサーとラズパイの接続

今回使用した温度センサーはBosch製のBMP280です。
　　 f:id:masashi_k:20200901234238p:plain

　メーカーサイトはこちら。データシートもここから参照できます。
　購入サイトはこちら。

インターフェースはI2CとSPIのいずれかを選択できます。ピン数が少なくて済むので、今回はI2Cを使用します。回路図は以下の通りです。

f:id:masashi_k:20200901235822p:plain

CSBpinのレベルでI2C, SPIを切り替えます。I2Cを使う場合はHigh, SPIの場合はLowにしてください。SDOはH/LでI2Cのスレーブアドレスを変更できます。モジュール内部でpull-downされているので、変更の必要がなければオープンでOKです。

Python スクリプトでLINE通知

Pythonでcurlコマンドを使いたい場合、requestというモジュールを使用します。Web APIを扱うときなどに使用するようですが、勉強不足で詳細まで理解できませんでした。

ですが、こちらのサイトを使うとcurlコマンドを自動的にpythonのコードに変換してくれます。勉強をさぼって、ありがたく使わせていただきます。
Convert cURL command syntax to Python requests, Ansible URI, browser fetch, MATLAB, Node.js, R, PHP, Strest, Go, Dart, JSON, Elixir, and Rust code

f:id:masashi_k:20200902000748p:plain

このサイトでcurlコマンドを変換すると"Hello world!"のメッセージはPython requestsでは以下のようになります。

import requests

headers = {
    'Authorization': 'Bearer xxxxxxx',　# xxxxxxxにトークンを入力
}
files = {
    'message': (None, 'Hello world!'),
}
response = requests.post('https://notify-api.line.me/api/notify', headers=headers, files=files)

温度センサーの測定データを送信

下準備ができたところで、いよいよ温度をスマホのLINEアプリに送ります。
I2Cを使用するために初期化した後、1分間隔で温度を測定し、結果を送信するプログラムです。温度算出の部分はBMP280のレジスタ値から温度を求める計算をしています。BMP280のデータシートに記載されているサンプルを実装しているので、詳細はデータシートをご参照ください。

import wiringpi
import time
import requests

#I2Cの初期化
slv_addr = 0x76wiringpi.wiringPiSetup()
i2c = wiringpi.I2C() #get I2C
Fd = i2c.setup(slv_addr) #setup I2C device
#温度測定開始
i2c.writeReg8(Fd, 0xF4 , 0x23)

try:
    while True:        #レジスタ値を読んで温度を算出(詳細はBMP280のデータシート参照)
        msb = i2c.readReg8(Fd, 0xFA) 
        lsb = i2c.readReg8(Fd, 0xFB)
        xlsb = i2c.readReg8(Fd, 0xFC)
        temp = (msb << 8)| lsb
        temp = temp << 4

        dig_t1 = i2c.readReg16(Fd,0x88)
        dig_t2 = i2c.readReg16(Fd,0x8A)
        if(dig_t2 & 0x8000):    
            dig_t2 = ((~dig_t2 & 0xFFFF) + 1)*-1
        dig_t3 = i2c.readReg16(Fd,0x8C) 
        if(dig_t3 & 0x8000):
            dig_t3 = ((~dig_t3 & 0xFFFF) + 1)*-1

        var1 = ((float)(temp)/16384.0 - (float)(dig_t1)/1024.0)*dig_t2
        var2 = (((float)(temp)/131072.0 - (float)(dig_t1)/8192.0) ** 2 )*dig_t3

        T = (var1 + var2) / 5120.0
        #print('temperture = {:.2f} [deg]'.format(T))
        #測定結果をLINEに通知                headers = { 'Authorization': 'Bearer xxxxxxx', }   #xxxxxxxにトークンを入力        files = { 'message': (None, 'Temp. = {:.2f}[deg]'.format(T)),}
        response = requests.post('https://notify-api.line.me/api/notify', headers=headers, files=files)
               #1分間隔 
        time.sleep(60)
    
except KeyboardInterrupt:
    print('!FINISH!')