雑コラBOTを作ろう⑥: Face APIで顔検出をしてみる

投稿日 2018年8月8日
著者 azure-recipe-user
カテゴリー Azure Functions
カテゴリー Cognitive Services

Michaelです。

「雑コラBOT」の作成の過程からAzure Functionsの設定方法を紹介する「雑コラBotを作ろう」の6回目です。
今回からはいよいよ「雑コラ」を作る仕組みを作成していきます。まずは、「雑コラ」を作るうえで重要な顔の位置を特定するために、Face APIを追加して顔の検出をしてみます。

構成

画像処理用のPython関数にFace APIをつなぎ、顔の検出をできるようにします。

Face APIの作成

Face APIを登録して、APIキーを取得します。
Azureポータルの新規作成画面で「Face API」を検索してリソースの作成を開始します。

リソース名、場所を設定します。
東日本リージョンでもFace APIが使えるようになっているので、雑コラBotの構成と同じ「東日本」で作成します。

リソースの作成が完了したらFace APIのリソース画面を開き、APIキーとエンドポイントをコピーして控えておきます。

スクリプト

スクリプトは以下のように作成しました。

# coding: utf-8

# ライブラリの読み込み
import os, json
import imghdr
import urllib.parse
import urllib.request

import hashlib


# バインド設定の読み込み
with open(os.path.dirname(__file__) + "/function.json", "r") as f:
    bindings = json.load(f)
    bindings = {bind["name"]: bind for bind in bindings["bindings"]}


# 1. Face APIキー情報
face_api_key = "{API Key}"
face_api_endpoint = "https://japaneast.api.cognitive.microsoft.com/face/v1.0"


# 2. Face APIへのリクエスト
def detect_face_from_data(api_key, body, endpoint_or_location="japaneast", returnFaceId=True, returnFaceLandmarks=False, returnFaceAttributes=[]):
    
    # URL生成
    params = urllib.parse.urlencode({
        "returnFaceId": returnFaceId,
        "returnFaceLandmarks": returnFaceLandmarks,
    })
    if returnFaceAttributes != []:
        params["returnFaceAttributes"] = ",".join(returnFaceAttributes)

    if "http" in endpoint_or_location:
        url = "{}/detect?{}".format(endpoint_or_location, params)
    else:
        url = "https://{}.api.cognitive.microsoft.com/face/v1.0/detect?{}".format(endpoint_or_location, params)

    # リクエストヘッダ
    headers = {
        "Ocp-Apim-Subscription-Key": api_key,
        "Content-Type": "application/octet-stream"
    }

    # HTTPリクエスト
    req = urllib.request.Request(url, data=body, headers=headers)
    try:
        with urllib.request.urlopen(req) as res:
            res_json = json.loads(res.read())

    except urllib.error.HTTPError as err:
        print(err.code)
        print(err.read())
        res_json = []

    except urllib.error.URLError as err:
        print(err.reason)
        res_json = []
    
    return res_json


print("Function Start. ID:" + os.environ["INVOCATIONID"])

def handler():

    # メッセージ読み込み
    with open(os.environ["inputMessage"], "r") as f:
        input_msg = json.load(f)

    # コンテンツのダウンロード
    print("Downloading input image")
    dl_url = input_msg["img_url"]
    req = urllib.request.Request(dl_url)
    with urllib.request.urlopen(req) as res:
        img_bytes = res.read()


    # 実行履歴の取得
    output_path = bindings["outputBlob"]["path"].format(blob_path=input_msg["blob_path"]).split("/")
    container = output_path[0]
    blob_path = "/".join(output_path[1:])
    img_name = output_path[-1]
    img_md5 = hashlib.md5(img_bytes).hexdigest()

    
    # 画像フォーマット判定
    print("Checking input image format")
    img_format = imghdr.what(None, h=img_bytes)
    
    if img_format in ["jpeg", "png", "bmp", "gif"]:
        
        if img_format == "jpeg":
            img_format = "jpg"


        # 3. 顔検出結果の取得
        print("Detecting Faces")
        detected_faces = detect_face_from_data(face_api_key, img_bytes, location=face_api_endpoint)

        if len(detected_faces):
            result_message = "%s face(s) datected." % len(detected_faces)
        else:
            result_message = "No faces detected."

        print(result_message)


        # 画像の出力
        with open(os.environ["outputBlob"], "wb") as fw:
            fw.write(img_bytes)


        # 出力メッセージの設定
        result = {
            "status": "ok",
            #"message": "Image is uplaoded.",
            "message": result_message,
            "container": container,
            "blob_path": blob_path,
            "filename": img_name,
            "content_md5": img_md5,
            "input": input_msg
        }


    else:
        # 出力メッセージの設定
        result = {
            "status": "error",
            "filename": img_name,
            "content_md5": img_md5,
            "input": input_msg,
        }

        if img_format is None:
            result["message"] = "Input file is not image."
        else:
            result["message"] = "Unsupported image format."


    # キューメッセージの出力
    print("Output Queue Message")
    with open(os.environ["outputQueueItem"], "w") as fw:
        json.dump(result, fw)


    # 実行履歴の書き込み
    print("Insert Log to Table")
    insert_entity = {key: val for key, val in result.items()}
    insert_entity.update({
        "PartitionKey": "results",
        "RowKey": os.environ["INVOCATIONID"]
    })

    with open(os.environ["outputTable"], "w") as fw:
        json.dump(insert_entity, fw)
        
    return


if __name__ == "__main__":
    handler()

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

# coding: utf-8

# ライブラリの読み込み

import os, json

import imghdr

import urllib.parse

import urllib.request

import hashlib

# バインド設定の読み込み

with open(os.path.dirname(__file__) + "/function.json", "r") as f:

bindings = json.load(f)

bindings = {bind["name"]: bind for bind in bindings["bindings"]}

# 1. Face APIキー情報

face_api_key = "{API Key}"

face_api_endpoint = "https://japaneast.api.cognitive.microsoft.com/face/v1.0"

# 2. Face APIへのリクエスト

def detect_face_from_data(api_key, body, endpoint_or_location="japaneast", returnFaceId=True, returnFaceLandmarks=False, returnFaceAttributes=[]):

# URL生成

params = urllib.parse.urlencode({

"returnFaceId": returnFaceId,

"returnFaceLandmarks": returnFaceLandmarks,

})

if returnFaceAttributes != []:

params["returnFaceAttributes"] = ",".join(returnFaceAttributes)

if "http" in endpoint_or_location:

url = "{}/detect?{}".format(endpoint_or_location, params)

else:

url = "https://{}.api.cognitive.microsoft.com/face/v1.0/detect?{}".format(endpoint_or_location, params)

# リクエストヘッダ

headers = {

"Ocp-Apim-Subscription-Key": api_key,

"Content-Type": "application/octet-stream"

}

# HTTPリクエスト

req = urllib.request.Request(url, data=body, headers=headers)

try:

with urllib.request.urlopen(req) as res:

res_json = json.loads(res.read())

except urllib.error.HTTPError as err:

print(err.code)

print(err.read())

res_json = []

except urllib.error.URLError as err:

print(err.reason)

res_json = []

return res_json

print("Function Start. ID:" + os.environ["INVOCATIONID"])

def handler():

# メッセージ読み込み

with open(os.environ["inputMessage"], "r") as f:

input_msg = json.load(f)

# コンテンツのダウンロード

print("Downloading input image")

dl_url = input_msg["img_url"]

req = urllib.request.Request(dl_url)

with urllib.request.urlopen(req) as res:

img_bytes = res.read()

# 実行履歴の取得

output_path = bindings["outputBlob"]["path"].format(blob_path=input_msg["blob_path"]).split("/")

container = output_path[0]

blob_path = "/".join(output_path[1:])

img_name = output_path[-1]

img_md5 = hashlib.md5(img_bytes).hexdigest()

# 画像フォーマット判定

print("Checking input image format")

img_format = imghdr.what(None, h=img_bytes)

if img_format in ["jpeg", "png", "bmp", "gif"]:

if img_format == "jpeg":

img_format = "jpg"

# 3. 顔検出結果の取得

print("Detecting Faces")

detected_faces = detect_face_from_data(face_api_key, img_bytes, location=face_api_endpoint)

if len(detected_faces):

result_message = "%s face(s) datected." % len(detected_faces)

else:

result_message = "No faces detected."

print(result_message)

# 画像の出力

with open(os.environ["outputBlob"], "wb") as fw:

fw.write(img_bytes)

# 出力メッセージの設定

result = {

"status": "ok",

#"message": "Image is uplaoded.",

"message": result_message,

"container": container,

"blob_path": blob_path,

"filename": img_name,

"content_md5": img_md5,

"input": input_msg

}

else:

# 出力メッセージの設定

result = {

"status": "error",

"filename": img_name,

"content_md5": img_md5,

"input": input_msg,

}

if img_format is None:

result["message"] = "Input file is not image."

else:

result["message"] = "Unsupported image format."

# キューメッセージの出力

print("Output Queue Message")

with open(os.environ["outputQueueItem"], "w") as fw:

json.dump(result, fw)

# 実行履歴の書き込み

print("Insert Log to Table")

insert_entity = {key: val for key, val in result.items()}

insert_entity.update({

"PartitionKey": "results",

"RowKey": os.environ["INVOCATIONID"]

})

with open(os.environ["outputTable"], "w") as fw:

json.dump(insert_entity, fw)

return

if __name__ == "__main__":

handler()

スクリプトの説明

Face APIキー情報
Face API作成時に取得したAPIキーとエンドポイントを設定します。
Face APIへのリクエスト
Face APIにリクエストし、画像内の顔を検出します。
今回は、関数でダウンロードした画像の顔検出を行うため、リクエストヘッダの「Content-Type」に「application/octet-stream」を指定し、関数が保持するバイナリデータを直接送信して顔検出を行います。
顔検出結果の取得
Face APIで顔を検出し、検出結果を取得します。
Face APIの顔検出は個々の顔の情報を配列で返すため、配列の要素数を取得してメッセージとしてSlackに出力し、顔検出できていることを確認します。

実行結果

Slackのチャンネルに画像のURLを投稿すると顔の検出数がBotのメッセージで返されました。

同様に画像を5人の子供の写真に変更すると、顔の検出数は5となり、正しく顔を検出できていることが確認できました。

まとめ

Face APIを追加して顔検出ができるようになりました。あとは、検出した顔の情報をもとにコラ画像を作成すれば雑コラBotの完成です。
次回は、今回のコードに画像合成の処理を追加して雑コラBotを完成させます。

この記事を書いた人

azure-recipe-user

記事一覧

雑コラBOTを作ろう⑥: Face APIで顔検出をしてみる

構成

Face APIの作成

スクリプト

スクリプトの説明

実行結果

まとめ

この記事を書いた人

azure-recipe-user

Cognitive Services: Translator Text API v3を試す

Azure Functions:キューメッセージをトリガーとしたPython関数を作成してみる

Azure Functions: FaceAPIとOpenCVで顔検出をしてみた。② －Face APIの利用－

Cognitive Services: Face APIについて