お知らせ機能付きエッジAIカメラとは
今回紹介する「お知らせ機能付きエッジAIカメラ」はこれ。
動画はこれ
https://youtu.be/PtSoUgCXeSI
監視カメラからプッシュ通知が欲しい
夜間にもオペレーターなしで無人で連続稼働してくれるロボットを業務で扱っています。夜間無人ロボットは便利ですが時々予期せぬトラブルで夜間に停止してしまい、時間と材料費をロスします。単純に設定ミスで止まることもありますが、微妙な取り上げる部品の形状誤差や位置ずれなどで止まることがあり注意しても避けられません。
せめて、夜間の稼働状況を、自分のスマートフォンや自宅のPCから見たいと考えて、obnizで遠隔監視カメラを作成し、qiitaに書きました。このobnizを使った遠隔監視カメラは非常に役立っています。自宅からも、無人ロボットの制御PCの画面を見て、ロボットが止まっていないか確認できます。十分便利なのですが、制御PCの画面にエラーメッセージが出たらプッシュ通知が欲しいと思うようになってきました。これが今回の課題です。
↑何とかしたい。↓
BrownieのMaker Faire Taipei 2019 Kitに出会った。
その時に、ミクミンP/Kazuhiro Sasao @ksasao さんのBrownieのMaker Faire Taipei 2019 Kitに出会いました。ミクミンPさんのデモを見せていただき感動しました。
Brownie Learnは、k-NN(k-Nearest Neighbor)を利用したオフライン転移学習というのを使っているそうです。QRコードを使って1枚の画像で学習できてしまうことがスゴイと思います。私の課題解決の用途にぴったりです。
このBrownieについてはミクミンP/Kazuhiro Sasao @ksasao さんの書かれたQiita記事「3000円の液晶付きAIカメラでオフライン転移学習する #M5StickV」、AIカメラ M5StickV 向けアプリ Brownieおよびスライドに詳しく書かれています。
WifiとLINE Notifyへの送信は、M5StickCを使おう！
M5StickVにはWifi機能はありません。このため、BrownieのMaker Faire Taipei 2019 Kitは、QRコードに書かれた文字列情報を、シリアル接続でノートPCに送り、ノートPCにインストールしたBrownie Moniterというソフトウェアによりアプリケーションを起動してIFTTTに送ったりSlackに送ったり音を鳴らしたりできます。この方法で実験室の監視カメラに使うためには専用のノートPCが1台必要になりコストが高くなるので、ノートPCではなくM5StickCを使ってWifi接続してLINE NOTIFYで知らせてほしいと考え、M5StickCとUART接続することを試みました。
ミクミンPさんもM5StickCでIFTTT経由でLINE NOTIFYに通知を実装されています。ミクミンPさんのTwitterはこれ
（小さな違いですが、このQiita記事に掲載した私の作ったM5StickC用のプログラムは、IFTTT経由でなく、直接LINE NOTIFYに送るプログラムです。）
M5StickV用プログラムにUART通信部分を付け加えた
ミクミンP/Kazuhiro SasaoさんのBrownieのMaker Faire Taipei 2019 Kitの中のboot.pyを書き換えました。
M5StickVとM5StickCとを通信させるためのプログラムは、ano研 @nnn112358 さんの書かれたWi-FiがないM5StickVを、M5StickCと繋ぎLINEに投稿してみるまでの手順を参考にさせていただきました。
書き換えたboot.pyは以下の通りです。
コメントアウト部分（#UART to StickCの10行 と #200(original) was modified to 40の1行）が、ミクミンP/Kazuhiro Sasao @ksasao さんのoriginalのboot.pyから変更した箇所です。
import brownie as br
import KPU as kpu
import sensor
import lcd
import time
import uos
import gc
from Maix import GPIO #UART to StickC
from fpioa_manager import fm, board_info #UART to StickC
from machine import UART #UART to StickC
fm.register(35, fm.fpioa.UART2_TX, force=True) #UART to StickC
fm.register(34, fm.fpioa.UART2_RX, force=True) #UART to StickC
uart_Port = UART(UART.UART2, 115200,8,0,0, timeout=1000, read_buf_len= 4096) #UART to StickC
def get_feature(task):
count = 20
for i in range(3):
for j in range(count):
img = sensor.snapshot()
if j < (count>>1):
img.draw_rectangle(1,46,222,132,color=br.get_color(255,0,0),thickness=3)
lcd.display(img)
time.sleep(1.0)
feature = kpu.forward(task,img)
return feature
def get_nearest(feature_list,feature):
nearest = 10000
name = ''
for n,vec in feature_list:
dist = 0
for i in range(0,768,3):
dist = (dist
+ (feature[i]-vec[i])**2
+ (feature[i+1]-vec[i+1])**2
+ (feature[i+2]-vec[i+2])**2)
if dist < nearest:
nearest = dist
name = n
return name,nearest
def get_unit_vector(vec):
sum_vec = 0
unit_vec = []
for i in range(768):
sum_vec = sum_vec + vec[i] * vec[i]
sum_vec = math.sqrt(sum_vec)
for i in range(768):
unit_vec.append(vec[i]/sum_vec)
return unit_vec
def get_angle(unit_vec1,unit_vec2):
sum_vec = 0
for i in range(768):
sum_vec = sum_vec + unit_vec1[i] * unit_vec2[i]
return math.acos(sum_vec)*180.0/math.pi
def get_dist(a,b,p):
u = 0
l = 0
for i in range(768):
u = u + (p[i]-a[i])
l = l + (b[i]-a[i])
return u / l
def load(filename):
feature_list=[]
feature_0=[]
feature_100=[]
try:
with open(filename, 'rt') as f:
li = f.readline()
while li:
li = li.strip().split(',')
n = str(li[0])
vec = [float(v) for v in li[1:]]
if n == '0':
feature_0.append([n,get_unit_vec(vec)])
elif n == '100':
feature_100.append([n,get_unit_vec(vec)])
else:
feature_list.append([n,vec])
li = f.readline()
except:
print("no data.")
return feature_list,feature_0,feature_100
def save(filename,feature_list):
gc.collect()
output = ""
try:
with open(filename, 'wt') as f:
for n,vec in feature_list:
f.write(n)
for v in vec:
vec_str=",{0:.5f}".format(v)
f.write(vec_str)
f.write('\n')
except:
print("write error.")
#
# main
#
br.show_logo()
br.exit_check()
br.initialize_camera()
feature_file = "/sd/features.csv"
feature_list,feature_0,feature_100 = load(feature_file)
task = kpu.load("/sd/model/mbnet751_feature.kmodel")
print('[info]: Started.')
info=kpu.netinfo(task)
#a=kpu.set_layers(task,29)
old_name=''
marker_0_100=0
clock = time.clock()
try:
while(True):
img = sensor.snapshot()
# QR Code check
res = img.find_qrcodes()
if len(res) > 0:
name = res[0].payload()
if name=="*reset":
feature_list = []
feature_0 = []
feature_100 = []
save(feature_file, feature_list)
br.play_sound("/sd/reset.wav")
else:
br.play_sound("/sd/camera.wav")
feature = get_feature(task)
feature_list.append([name,feature[:]])
if name=='0':
feature_0.append([name,feature[:]])
if name=='100':
feature_100.append([name,feature[:]])
save(feature_file, feature_list)
br.play_sound("/sd/set.wav")
gc.collect()
# print(gc.mem_free())
kpu.fmap_free(feature)
print("[QR]: " + name)
continue
# inference
fmap = kpu.forward(task, img)
plist=fmap[:]
clock.tick()
if len(feature_0)>0 and len(feature_100)>0:
p = plist
f0 = feature_0[0]
f100 = feature_100[0]
dist = 100.0 * get_dist(f0[1],f100[1],p)
dist_str = "%.1f"%(dist)
print("[DISTANCE]: " + dist_str)
img.draw_string(2,47, dist_str,scale=3)
lcd.display(img)
continue
name,dist = get_nearest(feature_list,plist)
#print(clock.fps())
if dist < 40 and name != "exclude": #200(original) was modified to 40
img.draw_rectangle(1,46,222,132,color=br.get_color(0,255,0),thickness=3)
img.draw_string(2,47 +30, "%s"%(name),scale=3)
if old_name != name:
namestring = str(name) + "\n" #UART to StickC
uart_Port.write(namestring) #UART to StickC
lcd.display(img)
br.play_sound("/sd/voice/"+name+".wav")
old_name = name
else:
old_name = ''
# output
img.draw_string(2,47, "%.2f "%(dist),scale=3)
lcd.display(img)
kpu.fmap_free(fmap)
except KeyboardInterrupt:
kpu.deinit(task)
sys.exit()
uart_Port.deinit() #UART to StickC
del uart_Port #UART to StickC
M5StickC用にUART通信で数値を受けて、LINE Notifyに送るようにした
M5stickCには以下のプログラムを書いて、インストールしました。M5StickVに対して数字の1,2,3の名前で学習させることを前提に書いています。YOUR_WIFI_SSIDとYOUR_WIFI_PASSWORDとYOUR_LINE_NOTIFY_TOKENは、自身のものに書き換えてください。LINEに送るメッセージも自身がわかるメッセージに書き換えてください。一つの特徴として、M5stickCのディスプレイに認識した画像（数字）の履歴が表示されるようにしました。
#include <M5StickC.h>
#include <WiFi.h>
#include <ssl_client.h>
#include <WiFiClientSecure.h>
const char* ssid = "YOUR_WIFI_SSID";
const char* passwd = "YOUR_WIFI_PASSWORD";
const char* token = "YOUR_LINE_NOTIFY_TOKEN";
const char* host = "notify-api.line.me";
HardwareSerial serial_ext(2);
void setup_wifi() ;
void setup() {
M5.begin();
setup_wifi();
send("LabCam was connected");
M5.Lcd.setRotation(1);
M5.Lcd.setCursor(2, 4, 2);
M5.Lcd.println("Wifi_Connected");
serial_ext.begin(115200, SERIAL_8N1, 32, 33);
}
void loop() {
M5.update();
delay(500);
if (serial_ext.available()>0) {
String str = serial_ext.readStringUntil('\n');
int data = str.toInt();
Serial.print("data:");
Serial.write(data);
if (data == 1) {
M5.Lcd.print('1');
send("画像1を認識した。");
} else if (data == 2) {
M5.Lcd.print('2');
send("画像2を認識した。");
} else if (data == 3) {
M5.Lcd.print('3');
send("画像3を認識した。");
}
}
vTaskDelay(2000 / portTICK_RATE_MS);
}
/* Wifiに接続する */
void setup_wifi() {
// We start by connecting to a WiFi network
Serial.println();
Serial.print("Connecting to ");
Serial.println(ssid);
WiFi.begin(ssid, passwd);
while (WiFi.status() != WL_CONNECTED) {
delay(500);
Serial.print(".");
}
Serial.println("");
Serial.println("WiFi connected");
Serial.println("IP address: ");
Serial.println(WiFi.localIP());
}
/* LINE Notifyに送る */
void send(String message) {
WiFiClientSecure client;
Serial.println("Try");
//LineのAPIサーバに接続
if (!client.connect(host, 443)) {
Serial.println("Connection failed");
return;
}
Serial.println("Connected");
//リクエストを送信
String query = String("message=") + message;
String request = String("") +
"POST /api/notify HTTP/1.1\r\n" +
"Host: " + host + "\r\n" +
"Authorization: Bearer " + token + "\r\n" +
"Content-Length: " + String(query.length()) + "\r\n" +
"Content-Type: application/x-www-form-urlencoded\r\n\r\n" +
query + "\r\n";
client.print(request);
//受信終了まで待つ
while (client.connected()) {
String line = client.readStringUntil('\n');
Serial.println(line);
if (line == "\r") {
break;
}
}
String line = client.readStringUntil('\n');
Serial.println(line);
}
githubにも上記プログラム入れています。ご利用ください。
https://github.com/MikiHiroshi/LabCam
ケースを作成
GROVEの4ピンケーブル(10cm)でM5StickVとm5StickCを接続して、ケースに入れて、作ってみました。M5StickVとm5StickCのケースへの固定はネオジウム磁石を使って「外したいときは外せて固定したいときは固定できる」方法を工夫して固定しました。ネオジウム磁石を使っていますので、M5StickVとm5StickCが誤動作しないか心配でしたが、ネオジウム磁石の影響と考えられる誤動作はありません。結構、磁力が強い状況でも、M5StickVとm5StickCは正常に動きます！！
監視カメラとして実用的に使うための課題
実験室で監視カメラとして動かしてみました。監視対象は、ロボットを制御しているノートPCの画面です。エラーメッセージが出たらLINEで知らせてくれます。実際に使ってみると課題が見えてきました。
（課題１）夜に実験室の電灯を消すと正常動作しません。部屋が明るい状態で覚えさせたこともあり、部屋が真っ暗になると、異なる画像に判断します。手っ取り早い解決策は、部屋の電灯をつけたままにして帰ることです。または、実験室の電灯を消した状態で画像を学習させる方法もあります。
（課題２）認識判定感度をゆるく(例：boot.py内のdistを < 200 と設定)すると、私の学習させる画像においては誤認識し、LINE NOTIFYに連続的に通知が来すぎます。これは、boot.py内のif dist < 40 and name != "exclude": のところです。逆に、認識判定感度を厳しく(例：boot.py内のdistについて < 20と設定)すると、同じ画像を何度も認識判断して、これも同じく、LINE NOTIFYに連続的に通知が来て迷惑で耐えられません。認識判定感度の指標distはゆるくても厳しくても（数字で言えば高くても低くても）実際には使えず、認識する画像に応じて適切に設定する必要があることがわかりました。私の学習させる画像の場合はboot.py内のdist < 40の記述でうまくいきました。この数値がうまく設定できると実用に耐えるお知らせ機能付きエッジAIカメラとして使えます。厳しくする設定するときはまず、<40 と設定して試すところから始めます。デモとして見せる時など、はじめからゆるく（認識を甘く、幅広く）設定した方がいいだろうなと考えるときは <100 と設定してください。
この監視カメラを広く使っていただきたいという思いでクラウドファンディングを始めたところ、目標額を達成し、プロジェクトがSUCCESSしました！！
https://camp-fire.jp/projects/view/211754
クラウドファンディングに限らず、今後も使いたいという方がいらっしゃったら、広める活動をしようと考えています。
快適な実験ライフを！！
記事の2か月後
この記事を書いた2か月後に、パトライトも同時に動いてお知らせする機能を追加しました。LINEだけでは気づかない場合のために。youtube動画はこれ。
https://www.youtube.com/watch?v=orj0XCE_fjM
Qiita記事はこれ
M5StickVとM5StickCでお知らせ機能付きエッジAIカメラにパトランプでお知らせする機能を追加した
https://qiita.com/MikH/items/275dd18d8c1b6039c0bc