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

kingyo222

6年前

Toggle

はじめに

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

概要

グループの削除・作成する（リセット）
Person を作成する
Person に Face を追加する
トレーニングを開始して終了まで待つ

使用方法

KEYを設定してください。
FACE_KEY = ‘XXXXXX’
GROUP_IDを指定してください。
GROUP_ID = ‘xxxxxx’
images 以下のフォルダに名前フォルダと画像ファイルを格納してください。

–	説明
main.py	メインプログラム
images	ここのフォルダ以下が登録対象になる
├ 名前１
│ └ イメージ１
├ 名前２
│ ├ イメージ２
│ ├ イメージ３
│ └ イメージ４
└ 名前３
└ イメージ５

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

ソースコード

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')