Azure Face API にデータをバッチで登録する

最終更新日

はじめに

Azure Face API ではAPIが提供されているが、事前にデータを準備しないとAPIを読んでも意味がない、今回はこの事前データを準備するためのバッチを作成したので以下にソースコードを示す。

概要

  1. グループの削除・作成する(リセット)
  2. Person を作成する
  3. Person に Face を追加する
  4. トレーニングを開始して終了まで待つ

使用方法

  1. KEYを設定してください。
    FACE_KEY = ‘XXXXXX’
  2. GROUP_IDを指定してください。
    GROUP_ID = ‘xxxxxx’
  3. images 以下のフォルダに名前フォルダと画像ファイルを格納してください。
説明
main.pyメインプログラム
imagesここのフォルダ以下が登録対象になる
├ 名前1
│ └ イメージ1
├ 名前2
│ ├ イメージ2
│ ├ イメージ3
│ └ イメージ4
└ 名前3
  └ イメージ5

※イメージは1人分の顔が映っている物を用意してください。

ソースコード

main.py
import os
import glob
import time
import urllib3
from time import sleep
import cognitive_face as CF

FACE_KEY = 'xxxxx'
FACE_URL = 'https://japaneast.api.cognitive.microsoft.com/face/v1.0'
GROUP_ID = 'xxxxx'

# init face api
urllib3.disable_warnings()
CF.Key.set(FACE_KEY)
CF.BaseUrl.set(FACE_URL)

def apiwait():
    sleep(5)

########################################################################
# テストコード
if __name__ == '__main__':

    # person の定義
    person = list()
    for d in os.listdir('./images/'):
        person.append({'user': d, 'personId': None})
    print('init : person cnt=' + str(len(person)))

    # image フォルダのファイルを列挙
    images = list()
    for p in person:
        dir = p['user']
        for fpath in glob.glob('./images/' + dir + "/*"):
            images.append({'user': dir, 'personId': None, 'url': fpath})
    # print(images)
    print('init : image cnt=' + str(len(images)))

    # group の再作成
    print('delete group...')
    try:
        result = CF.person_group.delete(GROUP_ID)
        print('delete group...OK')
    except CF.CognitiveFaceException as ex:
        print('delete group...SKIP')
    print('create group...')
    result = CF.person_group.create(GROUP_ID)
    print('create group...OK')

    # すべての person情報を作成
    for idx, p in enumerate(person):
        result = apiwait() or CF.person.create(GROUP_ID, p["user"])
        p['personId'] = result['personId']
        for i in images:
            if i['user'] == p['user']:
                i['personId'] = p['personId']
        print(str(idx) + '/' + str(len(person)) + ' person add', p)

    # faceの登録
    for idx, i in enumerate(images):
        try:
            print(str(idx) + '/' + str(len(images)) + ' face add...', i)
            result = apiwait() or CF.person.add_face(i['url'], GROUP_ID, i['personId'])
            # print(str(idx) + '/' + str(len(images)) + ' face add...OK')
        except CF.CognitiveFaceException as ex:
            print(str(idx) + '/' + str(len(images)) + ' face add...Exception\n' + "{0}".format(ex))

    # traningの開始
    result = apiwait() or CF.person_group.train(GROUP_ID)
    # print(result)
    print('training...')
    while True:
        result = apiwait() or CF.person_group.get_status(GROUP_ID)
        if result['status'] == 'succeeded':
            break
        time.sleep(5)
    print('training...OK')