ABEJA Platform Advent Calendar 2018の11日目です。

概要

当初はABEJA Platformにアノテーション済みデータを投げる方法に関する記事を書く予定だったのですが、そもそもPlatformの一連の流れを書いている記事がもう少し欲しいよねって思い、急遽予定を変更し、Object Detectionをターゲットに、データ集めから学習、推論までを一気にやってみる事にしました。

ぎょりさんが9目に酒類をオススメするAIを作った際に一連の流れを書いてくれているので、本記事はもう少し詳細な部分まで踏み込みます。

なお、本記事のサンプルコードはここに置きました。

準備

CLIとSDKのインストール

ABEJA Platformを利用するために、ブラウザから扱う事もできますがCLIを使ってターミナルから直接指示を出す事もできます。慣れてくると、そちらの方が容易になってきます。また、操作説明のために、画面のキャプチャを使うのもちょっと大変ですしね。まず、上記ABEJA CLIと、プラットフォームの各種機能を扱うためのSDKのインストールを行います。

$ curl -s https://packagecloud.io/install/repositories/abeja/platform-public/script.python.sh | bash
$ pip install abejacli
$ pip install abeja-sdk

初期設定

CLIのインストールが完了したら、まずはクレデンシャルを設定します。abeja configureコマンドを叩くことで、クレデンシャルを設定できますので、自身のユーザー名・アクセストークンを入力してください。

$ abeja configure
abeja-platform-user  : {User-ID}
personal-access-token: {Token}
organization-name    : {Organization-ID or Organization-Name}

注意点
- user名にはuser-は入れないこと
  - ○ 1234567890123
  - × user-1234567890123

データの収集

ABEJA Platformは、カメラなどのエッジデバイスからプラットフォームにデータを安全にアップロードするためのデータソースという仕組みを持ちます。通常、データをアップロードするためには、ユーザの認証情報が必要となるのですが、エッジデバイスにその認証情報を渡すのは危険です。そこで、データをアップロードする入口専用の認証情報を作成し、その認証情報を用いてデータをアップロードする事で、安全なアップロードが可能となります。本節では、認証情報の作成からアップロード方法までについて解説します。

カメラとABEJA Platformの接続

まずは、データを入れる場所となるDatalakeのチャンネルを作成します。DatalakeのチャンネルはCLIを使い以下のように作成します。ここで、{CHANNEL_NAME}には任意のチャンネル名、{DESCRIPTION}にはチャンネルの説明文を入れます。

$ abeja datalake create-channel -n {CHANNEL_NAME} -d {DESCRIPTION}

上記を実行すると、チャンネルが作成され以下のようなjsonが返りますので、そのchannel_idを控えておいてください。

{
    "channel": {
        "archived": false,
        "channel_id": "XXXXXXXXXXX",
        "created_at": "2018-12-10T09:27:11Z",
        "description": "{DESCRIPTION}",
        "display_name": "{CHANNEL_NAME}",
        "name": "camera_upload_test",
        "storage_type": "datalake",
        "updated_at": "2018-12-10T09:27:11Z"
    },
    "created_at": "2017-09-12T10:11:46Z",
    "organization_id": "XXXXXXX",
    "organization_name": "XXXXX",
    "updated_at": "2017-09-12T10:11:46Z"
}

続いて、上記に述べた認証情報を作成します。以下のようにcurlを用いてAPIを直接呼びます。{ORGANIZATION ID}、{USER_ID}、{ACCESS_TOKEN}は、ブラウザからコンソールを開き、右上のユーザ名をクリックすると確認できます。{DATASOURCE NAME}には、任意のデータソース名を入れます。

$ curl -X POST https://api.abeja.io/organizations/{ORGANIZATION ID}/datasources  \
	-u user-{USER_ID}:{ACCESS_TOKEN} \
	-H 'Content-Type:application/json' \
	-d '{"display_name": "{DATASOURCE NAME}"}'

上記を実行すると、以下のようなjsonが返ってきます。作成されたdatasource_idとsecretを控えておいてください。

{
  "updated_at": "2017-09-12T10:11:46Z",
  "organization_name": "XXXXXXXXXXXXXXXXXXXX",
  "organization_id": "{ORGANIZATION ID}",
  "datasource": {
    "updated_at": "2018-12-10T09:31:56Z",
    "secret": "XXXXXXXXXXXXXXXXXXXXXXXXXX",
    "display_name": "{DATASOURCE NAME}",
    "datasource_id": "1626744485075",
    "created_at": "2018-12-10T09:31:56Z"
  },
  "created_at": "2017-09-12T10:11:46Z"
}

最後に、はじめに作成したDatalakeのチャンネルと、Data sourceを紐付け、Data sourceの認証情報でDatalakeにデータをアップロードできるようにします。

$ curl -X PUT https://api.abeja.io/organizations/{ORGANIZATION ID}/channels/{CHANNEL ID}/datasources/{DATASOURCE} \
	-u user-{USER_ID}:{ACCESS_TOKEN}

データのアップロード

続いて、デバイスのカメラで撮影した画像を、上記で作成した認証情報を用いてDatalakeにアップロードしていきます。カメラはインターネットに繋がっていればなんでも良いです。アップロードのためには最初にインストールしたSDKを利用します。以下のようにclientを作り、post_channel_file_uploadでアップロードします。

from abeja.datalake import APIClient

credential = {
    'user_id': DATASOURCE_ID,
    'personal_access_token': DATASOURCE_TOKEN
}

client = APIClient(credential=credential)

def upload(img):
    img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)

    img = Image.fromarray(img)
    img_byte = io.BytesIO()
    img.save(img_byte, format='jpeg')
    data = img_byte.getvalue()
    content_type = 'image/jpeg'
    filename = datetime.now().strftime("%Y%m%d_%H%M%S") + '.jpg'
    metadata = {
        'x-abeja-meta-filename': filename
    }
    client.post_channel_file_upload(CHANNEL_ID, data, content_type, metadata=metadata)

撮影のところは、OpenCVを使って適宜撮影してください。画像を半分のサイズにして、Enterを押したらアップロードすることにします。

if __name__ == '__main__':
    cap = cv2.VideoCapture(0)

    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) // 2
    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) // 2

    while True:
        ret, frame = cap.read()
        img = cv2.resize(frame, (width, height))
        cv2.imshow('image', img)
        key = cv2.waitKey(1)
        if key == ord('q'):
            break
        elif key == 13:
            upload(img)

アノテーション

プロジェクトの作成

アノテーション作業の前に、タスクの種類、スキーマなどの設定によってプロジェクトを作成します。まずはじめにプロジェクトを作成しましょう。プラットフォームのコンソールの左にあるAnnotationを選ぶと、アノテーションツールのマネジメント画面に飛びます。以下のように、プロジェクトを作成しましょう。

今回は、Detectionを選びます。

コップを認識する事にしてみましょう。とりあえずクラス数は１で。

ひとまず記事向けにさっとやるということで、目標データ数は100とします。

設定を確認して、プロジェクトを作成します。

この後、プロジェクトのSyncを押して、作業者にデータを送ります。

アノテーションの実施

作業者は、プロジェクトの項目から作業を選んで、アノテーションを始めます。

Object Detectionタスクでは以下のようにアノテーションを進めていきます。

データセットの作成方法なども記事化しておきたいので、ここではアノテーション結果と元のデータを紐づけるという学習済みデータ作成プロセスを、手動でやってみる事にしましょう。といことで、終わったらアノテーション結果をダウンロードします。しばらくすると、アノテーション結果へのリンクが送られてくるのでダウンロードしましょう。

学習済みデータの作成

さて、作成したアノテーション結果を使い、Platformの学習に用いるには、少しややこしいプロセスを踏みます。具体的には

学習データのスキーマを定義
ダウンロードしたファイルから(1)Datalake情報と、(2)アノテーション結果、を読み込む
学習データのアノテーションフォーマットに変換してアップロード
という変換手順が必要です。順に確認していきます。

生データにアノテーション情報が紐づいたものを、ABEJA Platfromではデータセット、と呼びます。学習ではデータセットを読み込み学習を行います。データセットを作るには、スキーマの定義と、生データにアノテーション情報を付与してアップロード、という２つの作業が必要となります。まず、スキーマを定義しましょう。スキーマには、学習データにどのようなラベルが入っているかを入れておきます。今回は、ラベルはCup一種類として、以下のように定義します。

{
  "categories": [
    {
      "category_id": 0,
      "name": "0",
      "labels": [
        {
          "label": "Cup",
          "label_id": 0
        }
      ]
    }
  ]
}

ダウンロードしたデータには様々な情報が入っていますが、アノテーション情報はinformationの中に入っており、2018年12月現在Object detectionの場合は、下記のようなフォーマットです。多分、そのうち変わるので、適宜読み替えてください。rectがバウンディングボックス、classesがラベル情報となります。

{
  "information": [
    {
      "rect": [
        398.3673469387755,
        163.9183673469388,
        534.8571428571429,
        326.53061224489795
      ],
      "color": "rgb(249, 123, 106)",
      "classes": [
        {
          "id": 1,
          "name": "Cup",
          "category_id": 0
        }
      ]
    }
  ]
}

学習データに入れるアノテーション情報は以下のように定義することとします。フォーマットは特に決まっておらず、学習コード側で、このフォーマットから読み出せるように作ります。読み出すコードは後述します。

{
  "detection": [
    {
      "category_id": 0,
      "label": "Cup",
      "rect": {
        "ymin": 163.9183673469388,
        "ymax": 326.53061224489795,
        "xmax": 534.8571428571429,
        "xmin": 398.3673469387755
      },
      "label_id": 0
    }
  ]
}

なおアップロードするデータのうち、生データの情報として以下のようなdictを作成します。CHANNEL IDとFILE IDはアノテーションデータのmetadataの項目に入っているので、その値を利用します。

[
  {
    "data_uri": "datalake://{CHANNEL ID}/{FILE ID}",
    "data_type": "image/jpeg"
  }
]

これらをまとめると、スキーマを定義し、アノテーション情報を読み込み、アノテーションデータをアップロードする処理となりますが、コードを全て載せると長くなるので、ここではポイントのみ紹介します。スキーマを定義してDatasetの箱を作るのは下記となります。ここのprops変数は上記で定義したものと同一のものになります。以下を実行するとdatasetname変数に入れた名前のDatasetが作成されます。ラベルのIDは、アノテーション時に1からスタートした場合など、Deep Learningのインデックスとして扱いづらい場合があるので、0から切り詰めて設定し直します。label_to_idがラベル名からインデクスに変換するdictです。APIClientにクレデンシャルを設定し、create_datasetで作成したスキーマを引数として、データセットを作成します。

credential = {
    'user_id': ABEJA_PLATFORM_USER_ID,
    'personal_access_token': ABEJA_PLATFORM_TOKEN
}

labels =[]
label_to_id = {}
for count, cat in enumerate(categories):
    label_to_id[cat['name']] = count
    labels.append({
        'label': cat['name'],
        'label_id': count
    })
 
category = {
        'labels': labels,
        'category_id': 0,
        'name': datasetname
    }
    props = {'categories': [category]}

    api_client = APIClient(credential)
    dataset = api_client.create_dataset(organization_id, datasetname, 'custom', props)

また、アノテーション結果を読み取り、アップロードするフォーマットに変換する部分は以下になります。前半は、アノテーション結果を、attributeの形式に変換しています。後半はdata_sourceを作成し、最後にcreate_dataset_itemでデータを追加します。

info = []
for rect in d['information']:
    label = rect['classes'][0]['name']
    label_id = label_to_id[label]
    bbox = rect['rect']
    info.append({
        'category_id': 0,
        'label': label,
        'label_id': int(label_id),
        'rect': {
            'xmin': bbox[0],
            'ymin': bbox[1],
            'xmax': bbox[2],
            'ymax': bbox[3],
        }
    })

filename = d['task']['metadata'][0]['information']['filename']
file_id = d['task']['metadata'][0]['source']

content_type = 'image/jpeg'
data_uri = 'datalake://{}/{}'.format(channel_id, file_id)
source_data = [{'data_uri': data_uri, 'data_type': content_type}]
attributes = {'detection': info}

api_client.create_dataset_item(organization_id, dataset_id, source_data, attributes)

上記を実行すると、Datasetに指定したデータセットが作成されていることを確認できます。

学習の実行

さて、学習です。学習はSSDモデルを利用することとしましょう。また、データ量が少ないので、finetuningをする事とします。pretrainedモデルとして、PascalVOC2007と2012から学習済みのモデルを利用し、重みをコピーします。この辺、意外と面倒（追記もっと簡単にコピーする方法あるので、追記する）。

def copy_ssd(model, premodel):
    _ = premodel(np.zeros((1, 3, 300, 300), np.float32))
    _ = model(np.zeros((1, 3, 300, 300), np.float32))

    extractor_src = premodel.__dict__['extractor']
    multibox_src = premodel.__dict__['multibox']
    extractor_dst = model.__dict__['extractor']
    multibox_dst = model.__dict__['multibox']

    layers = extractor_src.__dict__['_children']
    for l in layers:
        if l == 'norm4':
            if extractor_dst[l].scale.shape == extractor_src[l].scale.shape:
                extractor_dst[l].copyparams(extractor_src[l])
        else:
            if extractor_dst[l].W.shape == extractor_src[l].W.shape:
                extractor_dst[l].copyparams(extractor_src[l])

    for c_src, c_dst in zip(multibox_src['conf'], multibox_dst['conf']):
        if c_dst.W.data.shape == c_src.W.data.shape:
            c_dst.copyparams(c_src)

    for c_src, c_dst in zip(multibox_src['loc'], multibox_dst['loc']):
        if c_dst.W.data.shape == c_src.W.data.shape:
            c_dst.copyparams(c_src)

また、取り敢えず、何も考えずにconv1_1あたりからnorm4あたりまでをがっつりフリーズさせます。

def fix_ssd(train_chain):
    names = ['conv1_1', 'conv1_2', 'conv2_1', 'conv2_2',
        'conv3_1', 'conv3_2', 'conv3_3',
        'conv4_1', 'conv4_2', 'conv4_3',
        'conv5_1', 'conv5_2', 'conv5_3',
        'conv6', 'conv7',
        'norm4']
    d = train_chain.model.extractor.__dict__
    for name in train_chain.model.extractor._children:
        if name in names:
            layer = d[name]
            layer.disable_update()

Platformにおける学習コードの詳細については、別の機会に譲りまして、ここではデータ周りのコードだけ、軽く紹介します。Chainerでは、データのバッチを出力するDatasetクラスを用意して、trainerに与える必要があります。ABEJA Platformと連携する際は、このDatasetクラス内で、プラットフォームからデータを読み込む仕組みにします。ここでは、DetectionDatasetFromAPIクラスというものを用意する事とします。

    train_data = DetectionDatasetFromAPI(train_data_raw)
    train_data = TransformDataset(train_data, Transform(model.coder, model.insize, model.mean))

DetectionDatasetFromAPIは長いので、いくつかに分けて紹介します。まず画像データを持ってくるところは以下のようになります。PlatformのDatasetAPIを用いてアイテムのリストを取得し、_get_imageメソッドで、アイテムを取得します。画像データはio.BytesIO形式で得られるので、PILを用いて開きます。

from abeja.datasets import Client

def load_dataset_from_api(self, dataset_id):
    client = Client()
    dataset = client.get_dataset(dataset_id)
    dataset_list = dataset.dataset_items.list(prefetch=True)
    return list(dataset_list)
    
def _get_image(self, i):
    item = self.dataset_list[i]
    file_content = item.source_data[0].get_content()
    file_like_object = io.BytesIO(file_content)
    img = Image.open(file_like_object)
    img = np.asarray(img, dtype=np.float32)
    img = img.transpose((2, 0, 1))
    return img

また、アノテーションデータを持ってくるところだけをピックアップすると以下のようになります。先ほどのアイテムから、各データのフォーマットは上記attributesで設定した通りになっていますので、それを読み込みます。

item = self.dataset_list[i]
annotations = item.attributes['detection']
bbox = []
label = []
difficult = []
for annotation in annotations:
    d = annotation['difficult'] if 'difficult' in annotation else False

    if not self.use_difficult and d:
        continue

    rect = annotation['rect']
    box = rect['xmin'], rect['ymin'], rect['xmax'], rect['ymax']
    bbox.append(box)
    label.append(annotation['label_id'])
    difficult.append(d)

bbox = np.stack(bbox).astype(np.float32)
label = np.stack(label).astype(np.int32)
difficult = np.array(difficult, dtype=np.bool)

さて、色々やってたら時間もあまり無くなってきたので、学習が終わらないし、チューニングの時間もない！今度時間があったら記事更新します・・・。

推論の実行

さて学習はそこそこに推論です。推論は二日目に実施したので、ほぼそのままですね。さて実行してみましょう。

まずは、モデルを作り、

import numpy as np
import chainer
from chainercv.datasets import voc_bbox_label_names
from chainercv.links import SSD300

ABEJA_TRAINING_RESULT_DIR = os.environ.get('ABEJA_TRAINING_RESULT_DIR', '.')

pretrained_model = os.path.join(ABEJA_TRAINING_RESULT_DIR, 'model_epoch_2000')
model = SSD300(n_fg_class=1, pretrained_model=pretrained_model)

def handler(_itr, ctx):
    for img in _itr:
        img = img.transpose(2, 0, 1)
        bboxes, labels, scores = model.predict([img])
        bbox, label, score = bboxes[0], labels[0], scores[0]

        result = []
        for b, lbl, s in zip(bbox, label, score):
            r = {'box': b.tolist(),
                 'label': voc_bbox_label_names[lbl],
                 'score': float(s)}
            result.append(r)
        yield result

ちょっとだけできた（しかし本当はこれ以外では結構認識できていない、データは大事）。

まとめ

本記事では、プラットフォームを使い、データソースを用いたカメラでの安全なデータ収集・アノテーション作業と結果の取り込み、学習データの作成、学習（SSDのfinetuning）、推論と盛り盛りな内容になってしまいました。結構説明飛ばしまくってるので、後ほど追加していきます。