Interface誌に、「ESP32+MicroPython」で記事を書いています。章は「新連載 逆引きMicroPythonプログラム集」です。ハイレベルな記事に囲まれて畏れ多いですが。。もし興味がありましたらご覧ください。記載内容がおかしい、よくわからない等ありましたら、コメントに書いていただいたら極力(best effortで:-)対応します。
Interface(インターフェース) 2021年 4 月号
今月号はFree RTOSと骨太な特集です。時間があったら自分もビルドしてみたい。しかし、、表紙の萌えイラストがなんとも。。萌え切らないし。。
デモ目的で、、ジェスチャ認識して、エアコンとか動いたらおもしろいかと。
ジェスチャ認識はGoogleのMedia Pipeというのがこなれているのでこれを使うと楽に動かせるのではと期待
cameraといえばopencvということで、rpiでopencvをインストール
apt-get install python3-opencv
ラズパイカメラから画像を取得して表示を繰り返すサンプルプログラム
(https://watlab-blog.com/2019/09/22/webcamera-realtime/)様より引用させていただきました
#!/usr/bin/python3
import cv2
camera = cv2.VideoCapture(0)
while True:
ret, frame = camera.read() # read frame
cv2.imshow('camera', frame) # show frame
# if enter q then exit
if cv2.waitKey(1) & 0xFF == ord('q'):
break
# release objects
camera.release()
cv2.destroyAllWindows()ラズパイで画像キャプチャできるのを確認した。
次に、MediaPipeなのだが、、
ラズパイでMediaPipe Python動かしたかったら普通には入れられず、以下の手順で入れるらしい。Experimentalの扱いです。。
https://github.com/jiuqiant/mediapipe_python_aarch64
Pythonパッケージ版のMediaPipeが超お手軽 + 簡易なMLPで指ジェスチャー推定 - Qiita
OpenCVのカメラ読み込みを高速化し、遅延時間も短くする - Qiita
事情により、Alexa -> IFTTT -> MQTTのデモを作ってみる。前提はすべて無料で！！ MQTTはずっとお試しで使っているshiftr.ioを使う。
AlexaとIFTTTをつなぐ方法はいろいろあると思いますが、IFTTTでIF部(Trigger)としてAlexa Serviceを設定して接続する。以下がおなじみのIFTTTの画面、IF部にAlexa Serviceを設定、THEN部にWebhooksからshiftr.ioへの記事投げ込みを設定する。
まず、、IFTTTでIF部(Trigger)として、Alexa Serviceを選択, Say a specific phraseを選択して、トリガーとなるキーワードを指定する。今回はエアコンと指定
THEN部でWeb APIを叩きたいので、サービスとしてWebhooksを選択すると、任意のGET/POSTが発行できる。この画面で、shiftrに対してURLでID/PWD,Topicを指定してPOSTでMessageを投げ込む
URLはID/PWD、topic、endpoint名は以下で構成される。
https://<id>:<PWD>@<instance_name>.cloud.shiftr.io/broker/<topic>
例：
ID/PWD:rai99/2423423
EndPoint:rain99.cloud.shiftr.io
Topic:room134/dev456/control
https://rain99:2423423@rain99.cloud.shiftr.io/broker/room134/dev456/control
登録が終わったら、自分のAlexaアプリで、「アレクサ、エアコン、トリガー」と発話すると、shiftr.ioにメッセージが投げ込まれる。
以下はスマフォのAlexaアプリ(Amazon Echoとかスマートスピーカが無くても音声制御ができる)
以下はShiftr.ioのコンソール画面
MQTTに記事が投げ込めたら、RPiとかでサブスクライバ（読者側）を走らせておけば遠隔制御とかが可能になる*1
IFTTTを無料枠で使う場合、作成できるルールが数種類なのと、トリガーワードに引数とか付けられなさそうな制限により、エアコンの温度設定とか運転モードとか、いろんな組み合わせは作れなさそうだ。もしまともに音声制御したかったら、Alexaのカスタムスキルを構築する必要があると思う。多分。カスタムスキルもAWS上で無料で作れるらしいのだが。。AWS上でLambdaを使えるなら、IFTTTを経由せずにAlexa->Lambda->Shiftrの流れでメッセージを投げこめばいいだろう。。
ちなみに、、IFTTTで作ったルール(アプレット)は普通に作ると個人専用になっていて、一般利用のために公開したい場合は公開手続きが必要らしいです。
ラスパイにカメラをつけてみた。Ubuntu のせいか、カメラを制御する、vcgencmd コマンドが無いようで、パッケージをインストール
$ sudo apt-get install libraspberrypi-bin
引数、get_cameraを指定して実行したが、デバイスが見つからないようだ。
$ vcgencmd get_camera VCHI initialization failed $ sudo vcgencmd get_camera supported=0 detected=0
/boot/firmware/config.txtに必要な設定を追加するらしい
$ pwd /boot/firmware $ diff config.txt_210218 config.txt 35a36,39 > > start_x=1 > gpu_mem=128 >
リブートして接続を再確認
$ sudo vcgencmd get_camera supported=1 detected=1
デバイスは見つかった。カメラ撮影コマンドで撮影してみる。
$ sudo raspistill -o /tmp/test.jpg
デバイスのPermissionの関係か、root権限でないと動作しない。撮影できたのが以下の画像
何も指定せずに撮影したが、サイズは2592x1944、JPEGフォーマットで3MB
パラメータとか調整しなくても結構きれいに撮れている。一発で動作した。すばらしい*1。
そもそも何がしたいのか？？
カメラの前でポーズしたら、エアコンが動いたり機器を制御させたい。
そのために、カメラで人物を撮影して骨格を抽出する機械学習APIを呼び出して、どんなポーズなのかを数値化させる必要がある。
当たり前だけど、少し調べた限り、ポーズ判定は計算コストがでかそうだ。だったら、、色判定で、カメラの前が青色で一杯だったらエアコンをつける・・というのでもいいのだけど。画像の4点から8点ぐらいを抽出して、RGBの比率で判断したらいい。。
■参考URL
Raspberry Pi Camera Module - Raspberry Pi Documentation
【MediaPipe】Raspberry Pi 4で環境構築し、CPU/GPUで動かしてみた（v0.7.5） | DevelopersIO
まぁ出尽くしたネタではあるが、、MQTTに投稿されたら、Node-REDでMQTTを読んでエアコンを制御するサンプルを実装
MQTTのトピックスは以下とした
<ホームID>/<デバイスID>/control
messageのPayloadはJSON形式で以下
{"control" : "pw_on"} or {"control" : "pw_off"}shiftrはID/PWDで認証するので、Node-REDでもMQTT購読設定時、ID/PWDを設定する
MQTTをウオッチして、制御が投稿されたらエアコンを制御するフローの例
状況を考慮せず、ひたすらエアコンのOn/Offを投げ続けるテストプログラム
#!/usr/bin/python3
import paho.mqtt.client as mqtt
import json
import pdb
HOST = "rainXXXX.cloud.shiftr.io"
ID ="rainXXXX"
PWD ="NhXXXXXXXX0a"
DEV_NAME = "controller00"
def on_connect(client, userdata, flags, rc):
print("Connected with result code "+str(rc))
client.subscribe("control/#")
def on_message(client, userdata, msg):
print(msg.topic+" "+str(msg.payload))
client = mqtt.Client(DEV_NAME)
client.username_pw_set(ID, password = PWD)
client.on_connect = on_connect
client.on_message = on_message
client.tls_set()
client.connect(HOST, 8883, 60)
client.loop_start()
import time
count = 0
homeID = 'home0123'
deviceID = 'ac0123'
message = {'control' : 'pw_on',} # 機器制御は、pw_on,pw_offを指定
topic = f"{homeID}/{deviceID}/control"
power_sw = False
while True:
print("wait...")
time.sleep(1)
if count % 5 == 0:
if power_sw :
message['control'] = 'pw_on'
else:
message['control'] = 'pw_off'
client.publish(topic,json.dumps(message))
print(json.dumps(message))
power_sw = not power_sw
count += 1
client.loop_stop()
client.disconnect()
補足：制御対象は、ECHONET-Liteで制御できる機器を想定
shiftrのコンソール画面（MQTTでPublishして機器制御している状態）
■追記
HTTPSで投稿する方法
shiftrではHTTPSで記事を投稿することも可能であった。mqttでは認証とかかなり苦労したのだが。。HTTPSなら一発で動く。
curl -X POST 'https://<id>:<passwd>@rainXXXXX.cloud.shiftr.io/broker/home123/dev456/control' -d 'power_on'
HTTPSで投げ込めるので、Alexa --> IFTTT --> shiftr の経路でMQTTがポストできる
■参考URL
Topicの命名法
MQTT Topic and Payload Design Notes
無料のMQTT Serverとして、mqtt.eclipseprojects.io 等があるが、認証付きのMQTT Serverを使ってみたかったので、shiftr.ioを試してみる。
アカウントを作って、無料プランでインスタンスを作れるようである。好きなホスト名(EndPointというかドメイン名というか)が選べるようなので、、適当に指定してみる。
インスタンス起動後、Clientからpub/subをしてみる。テストで使ったPythonはpaho-mqttというやつ。
paho-mqtt · PyPI
pip install paho-mqtt
使う時な以下の様にしてimportする(やり方はいろいろあると思いますが、、一例で)
import paho.mqtt.client as mqtt
動作確認したソースは以下
#!/usr/bin/python3
import paho.mqtt.client as mqtt
import pdb
HOST = "hogehoge.cloud.shiftr.io" # 作成したエンドポイント(hogehogeは仮称)
ID ="hogehoge" #エンドポイントのユニーク名がID
PWD ="XXXXXXXXXX" # PWDは作成したトークンを指定
# CONNACK を受けると呼び出される connect()
def on_connect(client, userdata, flags, rc):
print("Connected with result code "+str(rc))
client.subscribe("control/#")
# PUBLISH を受けると呼び出される message()
def on_message(client, userdata, msg):
print(msg.topic+" "+str(msg.payload))
client = mqtt.Client("dev00") # dev00は任意
client.username_pw_set(ID, password = PWD)
client.on_connect = on_connect
client.on_message = on_message
#pdb.set_trace()
client.connect(HOST, 1883, 60) # 1883指定でTCP / 8883でTLS このままだと丸見えだ
print("aft connect")
client.loop_start()
print("aft loop")
import time
count = 0
while True:
print("zzz...")
time.sleep(1)
if count % 5 == 0:
client.publish("control/dev00","stop")
count += 1 ShiftrではMQTTの利用状況がリアルタイムで表示される。一人だけでやってるとあまり面白さはないけど、複数でメッセージを共有すると視覚的効果が得られそうだ。
■経路暗号化
TLSをどう使えという説明がないので、正しいのかどうかわかりませんが、以下でTLSで接続できるのを確認
client = mqtt.Client("dev00")
client.username_pw_set(ID, password = PWD)
client.on_connect = on_connect
client.on_message = on_message
client.tls_set() # この行を追加
client.connect(HOST, 8883, 60) # ポートを8883に変更念のためパケットキャプチャして、TLSで暗号化されているのを確認
なぜMQTTか？ Node-REDでMQTTが使えるので、Node-REDとつないでみたいため。
■ご参考URL
制約はあるけどお試しで、TLS+認証が使えるMQTT Server
https://www.shiftr.io/
無料プランだと一日６Hまでとか制約があるようだ。どうやってインスタンスを止めたらいいのかわからないのだが。。
