プログラマーのalgopiaです。
昨年本会での登壇やIoTLT西東京版の主催などのご縁でこの度お声がけ頂き、IoTLT Advent Calendar 2017 (neo)の9日目を担当させていただくことになりました。

昨年の夏頃に登壇させて頂いた際、「画像をリアルタイムでクラウドに転送し続けるのは難しい」という話で纏めましたが、今回はその続きとして、リアルタイムかつ連続的に画像認識した結果をWebで可視化する手法の一つをご紹介したいと思います。

なお、本記事でのリアルタイムの定義は、認識デバイスでの物体認識を3fps程度、Webへの可視化まで2~3秒以内程度を想定してます。

プロセス

Jetson + Webカメラの撮影画像から、リアルタイムで物体認識
データ蓄積用InfluxDBの用意
認識結果をMQTTブローカーを介してInfluxDBへ転送
蓄積したDBのデータを地図上にマッピングしてWebで可視化

Jetson + Webカメラの撮影画像から、リアルタイムで物体認識

エッジ側でのリアルタイム画像認識には米国NVIDIA社のGPU一体型SoC TegraX2を搭載したJetsonTX2を使いました。
また、WebカメラにはBRIO(C1000ER)を用いました。

Faster-RCNNを用いた画像認識

Faster-RCNNはリアルタイム画像認識向けの深層学習アルゴリズムの一種です。
クラス推定ネットワーク以外にRegion Proposal Network (RPN)を有しており、物体候補領域も深層学習することにより総クラス推定数を削減することで、モデルに依存しますが認識速度を向上させることが可能となっています。

認識物体クラスと座標を文字列データにして軽量化

多数のデバイスがリアルタイムで画像を取得して高頻度でWANに転送し続けることは、2017年12月現在においては物理的にもコスト的にもあまり現実的ではないです。
そこで、デバイスで取得した認識結果をクラウド上のDBに転送するため、クラスとbboxの座標情報、スコア、検出時刻、デバイスMACアドレスのみ抽出して、MQTTでpublishすることにします。

MQTT publisher(paho)の用意

環境: JetsonTX2

MQTTクライアント(paho)のインストール

$ sudo apt-get update
$ sudo apt-get upgrade
$ sudo apt-get install python-dev python-pip
$ pip install --upgrade pip --user
$ pip install paho-mqtt --user

リアルタイム物体認識及びMQTT publish用スクリプト

必要なパッケージのインストールと改造Caffeのインストール

$ git clone --recursive https://github.com/rbgirshick/py-faster-rcnn.git

改造Caffe等のインストールをしないと実行できません。
リポジトリのREADMEを参考にインストールを進めてください。

スクリプトの作成

$ cd ~/py-faster-rcnn/tools
$ vim recognition.py

recognition.py

# -*- coding: utf-8 -*-
#!/usr/bin/env python
import _init_paths
from fast_rcnn.config import cfg
from fast_rcnn.test import im_detect
from fast_rcnn.nms_wrapper import nms
from utils.timer import Timer
import numpy as np
import scipy.io as sio
import caffe, os, sys, cv2
import argparse
from time import sleep
import paho.mqtt.client as mqtt
from datetime import datetime
import json


MAC_ADDR = '00016a7c67ef'    # MACアドレス(例)
MQTT_HOST = 'MQTTブローカーのホスト'
MQTT_PORT = 1883
MQTT_TOPIC = 'algopia/iotlt'
CLASSES = ('__background__', 'aeroplane', 'bicycle', 'bird', 'boat', 'bottle', 'bus', 'car', 'cat', 'chair',
           'cow', 'diningtable', 'dog', 'horse', 'motorbike', 'person', 'pottedplant', 'sheep', 'sofa', 'train',
           'tvmonitor', 'watch')
NETS = ('ZF', '**********.caffemodel')    # 学習済みのCaffeモデルを指定


def recognition(net):
    # カメラデバイスから画像取得
    img = cv2.VideoCapture(1)
    img.set(3, 1280)
    img.set(4, 960)
    img.set(5, 10.0)

    # 画像取得ループ
    frame = 0
    h_bbox = [0 for i in range(4)]
    client = mqtt.Client()
    client.connect(host=MQTT_HOST, port=MQTT_PORT)
    while True:
        var, im  = img.read()
        scores, boxes = im_detect(net, im)

        # 閾値
        CONF_THRESH = 0.8
    	NMS_THRESH = 0.3

        # 領域認識処理
        for cls_ind, cls in enumerate(CLASSES[1:]):
            cls_ind += 1
            cls_boxes = boxes[:, 4*cls_ind:4*(cls_ind + 1)]
            cls_scores = scores[:, cls_ind]
            dets = np.hstack((cls_boxes, cls_scores[:, np.newaxis])).astype(np.float32)
            keep = nms(dets, NMS_THRESH)
            dets = dets[keep, :]

        # 領域認識処理
        for cls_ind, cls in enumerate(CLASSES[1:]):
            cls_ind += 1
            cls_boxes = boxes[:, 4*cls_ind:4*(cls_ind + 1)]
            cls_scores = scores[:, cls_ind]
            dets = np.hstack((cls_boxes, cls_scores[:, np.newaxis])).astype(np.float32)
            keep = nms(dets, NMS_THRESH)
            dets = dets[keep, :]

            # 物体認識処理
            inds = np.where(dets[:, -1] >= CONF_THRESH)[0]
            if (len(inds) > 0):
                for i in inds:
                    bbox = dets[i, :4]
                    score = dets[i, -1]
                    cv2.rectangle(im, (bbox[0], bbox[1]), (bbox[2], bbox[3]), (255, 0, 255), 5)
                    cv2.putText(im, '{:s}'.format(cls, score), (bbox[0], int(bbox[1]+10)), cv2.FONT_HERSHEY_PLAIN, 1.0, (255, 255, 255))

                    # MQTTメッセージとしてpublish
                    date = datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ")
                    header = 0
                    h_header = format(header, '02x')
                    h_cls_ind = format(cls_ind, '02x')
                    h_bbox[0] = format(int(bbox[0]), '03x')
                    h_bbox[1] = format(int(bbox[1]), '03x')
                    h_bbox[2] = format(int(bbox[2]), '03x')
                    h_bbox[3] = format(int(bbox[3]), '03x')
                    h_score = format(int(score * 100), '02x')
                    print(header, cls_ind, cls, int(bbox[0]), int(bbox[1]), int(bbox[2]), int(bbox[3]), int(score * 100))
                    data = h_header + h_cls_ind + h_bbox[0] + h_bbox[1] + h_bbox[2] + h_bbox[3] + h_score
                    message = json.dumps({"date": date, "macAddr": MAC_ADDR, "data": data})
                    print(message)
                    client.publish(MQTT_TOPIC, message)

        # ディスプレイ表示
        im_resized = cv2.resize(im, ((im.shape[1]), (im.shape[0])))
        cv2.imshow("view", im_resized)
        key = cv2.waitKey(1)
        frame += 1

    img.release()
    cv2.destroyAllWindows()
    client.disconnect()


def parse_args():
    # パラメータ取得
    parser = argparse.ArgumentParser(description='Faster R-CNN demo')
    parser.add_argument('--gpu', dest='gpu_id', help='GPU device id to use [0]', default=0, type=int)
    parser.add_argument('--cpu', dest='cpu_mode', help='Use CPU mode (overrides --gpu)', action='store_true')
    args = parser.parse_args()
    return args


if __name__ == '__main__':
    # 変数宣言
    cfg.TEST.HAS_RPN = True
    args = parse_args()
    prototxt = os.path.join(cfg.MODELS_DIR, NETS[0], 'faster_rcnn_end2end', 'test.prototxt')
    caffemodel = os.path.join(cfg.DATA_DIR, 'faster_rcnn_models', NETS[1])

    # モデルの存在確認
    if not os.path.isfile(caffemodel):
        raise IOError(('{:s} not found.\nDid you run ./data/script/'
                       'fetch_faster_rcnn_models.sh?').format(caffemodel))

    # CPU or GPU
    if args.cpu_mode:
        caffe.set_mode_cpu()
    else:
        caffe.set_mode_gpu()
        caffe.set_device(args.gpu_id)
        cfg.GPU_ID = args.gpu_id

    # モデルのロード
    net = caffe.Net(prototxt, caffemodel, caffe.TEST)
    print '\n\nLoaded network {:s}'.format(caffemodel)

    # 認識処理開始
    recognition(net)

※ MAC_ADDR と MQTT_HOST と NETS は環境に応じて変更&指定してください。

スクリプト保存後、 data/faster_rcnn_models/ に学習済みモデルを保存してあることを確認して下記コマンドを実行し、処理を開始してください。

$ python recognition.py --gpu

出力動画はこちらになります。Jetson上で画面録画を動かすのは避けたかったので別カメラからの画像になります。

Jetson物体認識 & MQTT publish (Advent Calendar用) pic.twitter.com/17d6MGaQGD
— algopia (@algopia) 2017年12月9日

コンソールの出力がこちらになります。

...
...
(0, 15, 'person', 0, 24, 209, 713, 97)
{"date": "2017-12-07T10:19:04Z", "macAddr": "00016a7c67ef", "data": "000f0000180d12c961"}
(0, 15, 'person', 0, 12, 206, 712, 98)
{"date": "2017-12-07T10:19:05Z", "macAddr": "00016a7c67ef", "data": "000f00000c0ce2c862"}
(0, 15, 'person', 0, 16, 216, 719, 97)
{"date": "2017-12-07T10:19:05Z", "macAddr": "00016a7c67ef", "data": "000f0000100d82cf61"}
(0, 15, 'person', 0, 17, 211, 707, 97)
{"date": "2017-12-07T10:19:06Z", "macAddr": "00016a7c67ef", "data": "000f0000110d32c361"}
...
...

出力について説明しておきます。
まず、こちらは生データです。

(0, 15, 'person', 0, 24, 209, 713, 97)

左から順に、ヘッダー、クラスID、クラス名、左上x座標、左上y座標、右下x座標、右下y座標、スコアを表します。
ヘッダーは本記事では意味を持ちませんが、デバイス起動時や、通常時などに切り分けておいて、ヘッダーの値に応じてデータ列を変化させてやると、データ表現の幅が広がります。
payload長の制約が厳しい通信プロトコルでは工夫してやる必要があります。

次に、こちらがMQTTでpublishするメッセージです。

{"date": "2017-12-07T10:19:06Z", "macAddr": "00016a7c67ef", "data": "000f0000110d32c361"}

dataには先ほどの生出力の各データを16進数に変換したデータとしています。

これで画像内の物体クラスと、その座標、スコア、認識時刻を数十文字程度のテキストデータに圧縮してMQTTメッセージとしてpublishすることができました。

データ蓄積用InfluxDBの用意

Jetson+Webカメラでリアルタイム認識したデータを蓄積するためのInfluxDBを用意します。
InfluxDBは時系列データに特化したNoSQLです。

InfluxDBのセットアップ

AWS EC2インスタンスにInfluxDBをインストールして、 iotlt_db データベースと iotlt ユーザを作成します。

環境: AWS t2.micro(ubuntu16.04)

なお、説明省略のため以後本記事では全インスタンスでセキュリティグループは「全てのトラフィックを全てのIPに解放する」前提で進めますが、必ず適切に設定するようにしてください

$ wget https://dl.influxdata.com/influxdb/releases/influxdb_1.3.5_amd64.deb
$ sudo dpkg -i influxdb_1.3.5_amd64.deb
$ sudo sh -c 'echo 127.0.1.1 $(hostname) >> /etc/hosts'
$ sudo service influxdb start
$ /usr/bin/influx
> CREATE DATABASE iotlt_db
> SHOW DATABASES
name: databases
name
----
iotlt_db
> CREATE USER iotlt WITH PASSWORD 'password' WITH ALL PRIVILEGES
> SHOW USERS
user admin
---- -----
iotlt true
> exit

認識結果をMQTTブローカーを介してInfluxDBへ転送

Jetsonで検出及び加工したデータをMQTTブローカー(mosquitto)を経由して、ブリッジでsubscribeしてデータ整形を行い、InfluxDBへ挿入します。

MQTTブローカーの用意

環境: AWS t2.micro(ubuntu16.04)

$ sudo apt-get install gcc make g++ uuid-dev libssl-dev libc-ares-dev
$ cd /usr/local/src
$ sudo wget http://mosquitto.org/files/source/mosquitto-1.4.14.tar.gz
$ sudo tar zxvf mosquitto-1.4.14.tar.gz
$ cd mosquitto-1.4.14
$ sudo make && sudo make install
$ mosquitto

mosquittoを起動しておく

MQTT subscriber(paho)のインストール及びInfluxDB転送スクリプトの起動

環境: AWS t2.micro(ubuntu16.04)

$ sudo apt-get update
$ sudo apt-get upgrade
$ sudo apt-get install python-dev python-pip
$ pip install --upgrade pip --user
$ pip install boto --user
$ pip install paho-mqtt --user
$ pip install influxdb==4.1.1 --user

MQTT のsubscribe及びInfluxDBへの転送

$ vim sub.py

sub.py

# -*- coding: utf-8 -*-
import paho.mqtt.client as mqtt
from influxdb import InfluxDBClient
import json

host = "InfluxDBのホスト名"
port = '8086'
user = 'iotlt'
password = 'password'
dbname = 'iotlt_db'

# InfluxDBへの疎通
influx_client = InfluxDBClient(host, port, user, password, dbname)

# The callback for when the client receives a CONNACK response from the server.
def on_connect(client, userdata, flags, rc):
    print("Connected with result code "+str(rc))
    client.subscribe(topic='algopia/iotlt')

# The callback for when a PUBLISH message is received from the server.
def on_message(client, userdata, msg):
    # InfluxDBへのデータ投入
    json_body = json.loads(msg.payload)
    json_query = decode(json_body)
    influx_query = json.loads(json_query)
    influx_client.write_points(influx_query)

# MQTTをデコードしてInfluxDB用のデータ形式に整形
def decode(json_body):
    # データを要素に分解
    date = str(json_body['date'])
    mac_addr = str(json_body['macAddr'])
    payload = str(json_body['data'])

    # ペイロードのデコード
    header = payload[0] + payload[1]
    data = payload[2] + payload[3] + payload[4] + payload[5] + payload[6] + payload[7] + payload[8] + payload[9] + payload[10] + payload[11] + payload[12] + payload[13] + payload[14] + payload[15] + payload[16] + payload[17]

    # ヘッダーのデコード
    header_int = int(header, 16)
    header_bin = format(header_int, '08b')
    message_type = int(header_bin[0] + header_bin[1], 2)
    profile = int(header_bin[2] + header_bin[3] + header_bin[4] + header_bin[5] + header_bin[6] + header_bin[7], 2)

    # データ部のデコード
    s_bbox = [0 for i in range(4)]
    bbox = [0 for i in range(4)]
    s_cls_ind = data[0] + data[1]
    s_bbox[0] = data[2] + data[3] + data[4]
    s_bbox[1] = data[5] + data[6] + data[7]
    s_bbox[2] = data[8] + data[9] + data[10]
    s_bbox[3] = data[11] + data[12] + data[13]
    s_score = data[14] + data[15]
    cls_ind = int(s_cls_ind, 16)
    bbox[0] = int(s_bbox[0], 16)
    bbox[1] = int(s_bbox[1], 16)
    bbox[2] = int(s_bbox[2], 16)
    bbox[3] = int(s_bbox[3], 16)
    score = int(s_score, 16)

    # InfluxDBへqueryとして送るjsonを作成
    json_query =\
    [
        {
            "measurement": "iotlt_data",
            "tags": {
                "macAddr": mac_addr
            },
            "time": date,
            "fields": {
                "message_type": message_type,
                "profile": profile,
                "cls_ind": cls_ind,
                "bbox[0]": bbox[0],
                "bbox[1]": bbox[1],
                "bbox[2]": bbox[2],
                "bbox[3]": bbox[3],
                "score": score
            }
        }
    ]

    # jsonデータを書き込む
    jsonstring = json.dumps(json_query)
    return jsonstring

# MQTTへの疎通
client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message

client.connect(host="MQTTブローカーのホスト", port=1883)

client.loop_forever()

スクリプトを起動しておく

$ python sub.py

実行すると、Jetsonからpublishされたデータがsubscribeされます。

...
...
{u'date': u'2017-12-07T10:19:04Z', u'macAddr': u'00016a7c67ef', u'data': u'000f0000180d12c961'}
(15, 0, 24, 209, 713, 97)
{u'date': u'2017-12-07T10:19:05Z', u'macAddr': u'00016a7c67ef', u'data': u'000f00000c0ce2c862'}
(15, 0, 12, 206, 712, 98)
{u'date': u'2017-12-07T10:19:05Z', u'macAddr': u'00016a7c67ef', u'data': u'000f0000100d82cf61'}
(15, 0, 16, 216, 719, 97)
{u'date': u'2017-12-07T10:19:06Z', u'macAddr': u'00016a7c67ef', u'data': u'000f0000110d32c361'}
(15, 0, 17, 211, 707, 97)
{u'date': u'2017-12-07T10:19:06Z', u'macAddr': u'00016a7c67ef', u'data': u'000f00000f0d72cc61'}
(15, 0, 15, 215, 716, 97)
{u'date': u'2017-12-07T10:19:06Z', u'macAddr': u'00016a7c67ef', u'data': u'000f0000180dc2cf60'}
(15, 0, 24, 220, 719, 96)
{u'date': u'2017-12-07T10:19:07Z', u'macAddr': u'00016a7c67ef', u'data': u'000f00000d0d32b861'}
(15, 0, 13, 211, 696, 97)
{u'date': u'2017-12-07T10:19:07Z', u'macAddr': u'00016a7c67ef', u'data': u'000f0000200d12ba5e'}
(15, 0, 32, 209, 698, 94)
{u'date': u'2017-12-07T10:19:07Z', u'macAddr': u'00016a7c67ef', u'data': u'000f00002210f2cf55'}
(15, 0, 34, 271, 719, 85)
{u'date': u'2017-12-07T10:19:08Z', u'macAddr': u'00016a7c67ef', u'data': u'000f0000420e22cf54'}
(15, 0, 66, 226, 719, 84)
{u'date': u'2017-12-07T10:19:08Z', u'macAddr': u'00016a7c67ef', u'data': u'000f00003c0e12c856'}
(15, 0, 60, 225, 712, 86)
{u'date': u'2017-12-07T10:19:13Z', u'macAddr': u'00016a7c67ef', u'data': u'000f0000001cf2cf61'}
(15, 0, 0, 463, 719, 97)
{u'date': u'2017-12-07T10:19:13Z', u'macAddr': u'00016a7c67ef', u'data': u'000f0000002442c462'}
(15, 0, 0, 580, 708, 98)
{u'date': u'2017-12-07T10:19:14Z', u'macAddr': u'00016a7c67ef', u'data': u'000f0000003132c560'}
(15, 0, 0, 787, 709, 96)
{u'date': u'2017-12-07T10:19:14Z', u'macAddr': u'00016a7c67ef', u'data': u'000f0250003092cf51'}
(15, 37, 0, 777, 719, 81)
{u'date': u'2017-12-07T10:19:14Z', u'macAddr': u'00016a7c67ef', u'data': u'00080000004a32bf59'}
(8, 0, 0, 1187, 703, 89)
{u'date': u'2017-12-07T10:19:15Z', u'macAddr': u'00016a7c67ef', u'data': u'00080640024c82cf5e'}
(8, 100, 2, 1224, 719, 94)
{u'date': u'2017-12-07T10:19:16Z', u'macAddr': u'00016a7c67ef', u'data': u'000523c1422e62ba5d'}
(5, 572, 322, 742, 698, 93)
{u'date': u'2017-12-07T10:19:42Z', u'macAddr': u'00016a7c67ef', u'data': u'000f0000413862cf5c'}
(15, 0, 65, 902, 719, 92)
...
...

また、この sub.py はInfluxDBへのデータ転送も兼ねており、InfluxDBのインスタンスに接続して、DBの中身を確認すると、データが保存されていることがわかります。

$ /usr/bin/influx
> use iotlt_db
> use iotlt_db
Using database iotlt_db
> select * from iotlt_data;
name: iotlt_data
time                bbox[0] bbox[1] bbox[2] bbox[3] cls_ind macAddr      message_type profile score
----                ------- ------- ------- ------- ------- -------      ------------ ------- -----
...
...
1512641944000000000 0       24      209     713     15      00016a7c67ef 0            0       97
1512641945000000000 0       16      216     719     15      00016a7c67ef 0            0       97
1512641946000000000 0       24      220     719     15      00016a7c67ef 0            0       96
1512641947000000000 0       34      271     719     15      00016a7c67ef 0            0       85
1512641948000000000 0       60      225     712     15      00016a7c67ef 0            0       86
1512641953000000000 0       0       580     708     15      00016a7c67ef 0            0       98
1512641954000000000 0       0       1187    703     8       00016a7c67ef 0            0       89
1512641955000000000 100     2       1224    719     8       00016a7c67ef 0            0       94
1512641956000000000 572     322     742     698     5       00016a7c67ef 0            0       93
1512641982000000000 0       65      902     719     15      00016a7c67ef 0            0       92
...
...

これでリアルタイムの物体認識情報をInfluxDBへ集約できました！
Jetsonを複数台用意して、各デバイスにMACアドレスを持たせて別々の地理空間に設置してやることで、その場にいなくても様々な場所のリアルタイムな認識情報の取得・蓄積を自動化することができます（もちろん勝手に設置してはいけません）。

蓄積したDBのデータを地図上にマッピングしてWebで可視化

OpenStreetMapとZingChartで、地図上にヒートマップで可視化したり、チャートで時系列に可視化してみました。
※ Jetsonたくさんないので、画像はたくさんあるデバイスのセンサデータの描画です<(_ _)>

スクリーンショット 2017-12-06 15.23.34.png (1.9 MB)

今回の手法なら認識クラスごとにヒートマップを切り替えることもできるので、車マップや人マップ、独自クラスのマップなどに切り替えも可能ですね。

店舗経営等されてる方は、お店が多店舗展開したあかつきには、これで店舗設置の防犯カメラ+人認識を応用して混雑率などが一目瞭然になりそうです。
他にも機械学習に応用するなり、DCGANで画像生成してみるなり、と色々応用できるかと思われます。

まとめ

エッジ側にリアルタイム物体認識アルゴリズムを搭載し、データ量を圧縮することでMQTTでpub/subしてInfluxDBへ蓄積し、そのデータをWebフロントエンドjsで可視化させてみました。
説明簡略化のためInfluxDBやMQTTブローカーのクラスタ化、セキュリティなど考慮していないのでご注意ください（むしろこの上ないくらいopenにしてありますので）。