micropython on ESP32 でpush通知
TwitterやLINEで通知を試しましたが、twitterやLINEのアプリをインストールしたりアカウントを作成する手間も省きたいというモノグサ魂がムクムクと。。。
モノグサのためなら苦労も厭わないという、本末転倒な感じですが(笑)。
また、通知の痕跡が残らないようにしたいという希望も。「なお、このメッセージは自動的に消滅する」というアレみたいな感じですね。
どうせなら、PCだけでなくスマホにも、と欲求は尽きません。
で、いろいろと調べてみたところ、Web Pusht通知を使うとそんな感じの機能が実現できそうです。
ただし、すべての環境で動作するわけではなさそうです。
試した時点ではWindows10のfirefoxとChromeでは動作しましたが、Edgeでは動作しませんでした。その他のブラウザは持ってないので試してません。
MacOSは持ってないのでわかりません(貧乏なので・・・)。
スマホはAndroidのみ試しました。
iOSではPush7アプリのインストールが必要なようです。詳細はググってください。
push通知をローカルな環境だけで実現できそうにないので、push通知サービスのPush7を使うことにします。
freeプランだと5000送信/月ですが、趣味でちょっと試すくらいなら十分でしょう。
注意しないといけないのが、制限がリクエスト数ではなく、送信数だというところです。購読者が10人いると、1リクエストあたり10送信行われることになります。
例えば、10分に1回づつ通知して(そんなに通知することはないと思いますが)、1か月で6×24×30=4320回だと思うと大間違いです。さらに購読者数をかけなければいけません。
また、自分で購読要求を制御できない(そもそも他人に勝手に登録して利用してもらうサービスなので)ので、秘密の情報など流さないように注意しましょう。
まずはWebプッシュ通知サービス「Push7」の導入方法①アカウント登録編にしたがってアカウント登録～アプリケーション登録を行ってください。
アプリケーションの登録でのいっぺーちゃんなりの補足です。
- アプリケーション名(↑のヘルプページではサイト名になっている)を入力します
- 自分の好きな名前で良い
- URLを入力します。
- 購読ページの戻るボタンのリンク先です
- とりあえず自分のブログのURLとか、やほーのトップとか書いておく
- アイコンが必要なら画像をアップロードします
- カテゴリを選択します
- 「娯楽・レジャー・趣味」かな。
- アプリケーションURLを入力します
- アプリケーション名と同じにしておくと分かりやすいかな？
- 既に使用されているとダメ。
ダッシュボードでアプリケーションを開くと、「初期導入」の説明がありますが、今回はESP32からAPIで通知要求を行うので、無視して構いません。
また、APIの実行に必要な「App Number」と「API Key」は左側のメニューの「API」をクリックすると確認できます。
次は通知を受け取る側の設定(購読の設定)です。
何もしないとpush通知を受け取れません。モノグサでもこれだけは我慢してやってください。
- ブラウザ(スマホでも可)でアプリケーションURLに指定したURLにアクセス
- アプリケーションURLはアプリケーションの作成で設定した 「https://<<設定した名前>>.app.push7.jp」です。
- アプリケーションURLを忘れた場合は、以下にブラウザでアクセスすると調べられます。
- 「https://api.push7.jp/api/v1/<<app number>>/head」
- ここのdomainに表示されているURLがアプリケーションURLです。
- 「～～に通知の送信を許可して受信しますか？」と聞かれるので、「通知を許可する」をクリック
- 再度アプリケーションURLにアクセスし、「購読解除」をクリックすると購読をやめられます。
次にテストしてみましょう。
- Push7のダッシュボードに戻ります。
- 左側のメニューの一般通知(送信の下にあります)で「新規プッシュ通知」をクリック
- タイトル、URL、内容を入力し、必要なら「クリックされるまで表示」を選択し、送信をクリック
- タイトルと内容が通知ウィンドウに表示される文字列です。
- URLは通知ウィンドウをクリックしたときにジャンプするURLです。
- すべて入力しないとエラーになりますので、ダミーでも入れてください。
- 送信をクリック
- → 通知が表示されます。
「クリックされるまで表示」を選択しても、PCのブラウザ(Firefox/chromeで確認)では20秒程度で表示が消えてしまいます。
これはブラウザが表示を消しているようです。
逆にスマホだと、「クリックされるまで表示」を選択しなくても通知領域にずっと表示されているようです。
firefoxの場合、通知をクリックするまで表示させるには、以下の設定を行う必要があります(55.0.3 64bitで確認)。
ただし、どのような副作用が起こるか確認していませんので、設定は自己責任でお願いします。
- about:config で設定エディタを開く
- 「dom.webnotifications.requireinteraction.enabled」を「true」に設定
chromeでは対応する設定は見当たりませんでした。
なお、PCで通知を受け取るにはブラウザ(複数インストールされている場合は、購読のページで購読を許可したブラウザ)を起動しておく必要があります。どこか特定のページを開いておかなければならない、というようなことはありません。
いよいよESP32から通知を行います。
以下のプログラムをtiny_push7.pyという名前で保存し、いつものようにupipmでインストールしてください。
import usocket as socket import ussl as ssl class tiny_push7 : def __init__(self, app_no, api_key, icon, url, debug=False) : # パラメータチェック if type(app_no) is not str: raise ValueError("app_no must be string") if type(api_key) is not str: raise ValueError("api_key must be string") if type(icon) is not str: raise ValueError("icon must be string") if type(url) is not str: raise ValueError("url must be string") self.app_no = app_no self.api_key = api_key self.icon = icon self.url = url self.__DEBUG__ = debug def __debug_print(self, str) : if self.__DEBUG__ : print(str) def __makeRequestMessage(self, host, title, body, disappear) : # disappear の設定を文字列化 disappear_str = "true" if disappear else "false" # make request request = 'POST /api/v1/' + self.app_no + '/send HTTP/1.1\n' # bodyがbytesなので、requestもbytesに request = request.encode('utf-8') # make message body # disappear_instantlyは効かないけどとりあえず設定しておく body = '{"title":"' + title + '",' \ + '"body":"' + body + '",' \ + '"icon":"' + self.icon + '",' \ + '"url":"' + self.url + '",' \ + '"disappear_instantly":'+ disappear_str +',' \ + '"apikey":"' + self.api_key + '"}' # bodyは RFC3986 エンコードではなく、バイナリエンコード body = body.encode('utf-8') # make message header header = 'Host: ' + host['host'] + '\n' \ + 'User-Agent: micropython push7 agent v0.1\n' \ + 'Content-Type:application/json\n' \ + 'Connection: close\n' \ + 'Accept: */*\n' \ + 'Content-Length: ' + str(len(body)) + '\n' # bodyがbytesなので、headerもbytesに header = header.encode('utf-8') # request message ret = request + header + b'\n' + body + b'\n' return(ret) def __sendmessage(self, host, msg) : sock = socket.socket() addr = socket.getaddrinfo(host['host'], host['port'])[0][-1] # connect socket sock.connect(addr) try : # SSL wrap ssl_sock = ssl.wrap_socket(sock) # send data ssl_sock.write(msg) # 受信データの最初の1行 rcv_line = ssl_sock.readline() protover, status, rcv_msg = rcv_line.split(None, 2) status = status.decode('utf-8') rcv_msg = rcv_msg.decode('utf-8') self.__debug_print('%s %s %s' % (protover, status, rcv_msg)) # それ以外のレスポンスヘッダを読む while True: rcv_line = ssl_sock.readline() self.__debug_print(rcv_line) if not rcv_line: # なんらかの異常なレスポンス(ヘッダが終わる前にデータがなくなった) ssl_sock.close() raise ValueError("Unexpected EOF in HTTP headers") if rcv_line == b'\r\n': # 空行でヘッダ終了 break except Exception as e: # エラーが発生したらクローズして上位へ例外通知 ssl_sock.close() raise e # メッセージ本体を受信(とりあえず読み捨て) rcv_line = b'' while True : try : l = ssl_sock.readline() except Exception as e: # エラーが発生したらクローズして上位へ例外通知 ssl_sock.close() raise e if not l: # データがない → 終了 break # 読み込んだデータをためる rcv_line += l self.__debug_print("@@@@" + str(rcv_line)) self.__debug_print("close!!") ssl_sock.close() # 受信終了後にエラー判定 if not ((status == "200") or (status == "201")) : # status 200,201 以外はエラー raise ValueError(status +' '+rcv_msg) # ######## notify API ################################ def push(self, title, body, disappear=True) : host = { 'host': 'api.push7.jp', 'port': 443} reqMessage = self.__makeRequestMessage(host, title, body, disappear) self.__debug_print('%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%') self.__debug_print(reqMessage) self.__debug_print('%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%') self.__sendmessage(host, reqMessage)
以下解説
送信データを作成している部分です。
API仕様にしたがって送信データを作成しています。
TwitterやLINEのときは、body部分はRFC3986でエンコードしていましたが、Push7ではRFC3986ではなく、バイナリデータそのもので送信します。
よって、メッセージヘッダの Content-Length にはバイナリエンコードした後のデータ長を設定します。
(RFC3986でエンコードしたデータはASCII文字のみなので、Stringのままデータ長を取得しても大丈夫でした。ここでハマりました。。。)
def __makeRequestMessage(self, host, title, body, disappear) : ・・・・
なお、API仕様についてはPush7 APIの利用 - Push7 Docsを参照してください
データを送信する部分ですが、ほとんどこれまでの使いまわしです。
レスポンスのエラー判定方法と例外をraiseする部分をちょっと変更してあります。
def __sendmessage(self, host, msg) : ・・・・
API関数です。
パラメータtitleにタイトル、bodyに本文を設定します。
disappearにTrueを設定する(or設定を省略する)とダッシュボードから送信した時の「クリックされるまで表示」を選択していないのと同等です。
disappearにFalseを設定するとダッシュボードから送信した時の「クリックされるまで表示」を選択したのと同等です。
def push(self, title, body, disappear=True) : ・・・・
以下が通知プログラムです。
app_no = '<<ダッシュボードで取得したApp Number>>' api_key = '<<ダッシュボードで取得したAPI Key>>' icon = '<<通知に表示するアイコンのURL>>' # httpsでなければならない。512文字以下 url = '<<クリックしたときにジャンプするページのURL>>' # 512文字以下 from tiny_push7 import tiny_push7 # 初期化 tp = tiny_push7(app_no, api_key, icon, url, debug=True) title = "緊急指令" body = "このメッセージは自動的に消滅する" tp.push(title, body, True) または tp.push(title, body, False) # クリックするまで表示される
iconに設定するURLは通知に表示するアイコンのURLですが、HTTPS接続のサイトにある必要があります。
もし用意できない場合は、Push7のアプリケーションで設定したアイコンを使うと良いでしょう。
- アプリケーションで設定したアイコンのURLは以下にブラウザでアクセスすると調べられます。
- 「https://api.push7.jp/api/v1/<<app number>>/head」
- ここのiconに表示されているURLがアイコンのURLです。
urlには通知ウィンドウをクリックしたときにジャンプするページのURLを指定します。
実在しないページでも大丈夫なようですが、URLの形になっていないとエラーになります。
(ちょっと試した感じでは、 "" ⇒ NG "http://hoge" ⇒ NG "http://hoge.jp" ⇒ OK)
念のため「やほー」とか「ごーごる」のURLを入れておくのが良いと思います。
ここはHTTPS接続のサイトである必要はありません。
urlはpush API側で設定するようにしてもいいかなとも思いましたが、とりあえずコンストラクタで指定するようにしました。
あとは説明するまでもありませんね。
