スクリプトは以下のように作成しました。
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キー情報
faceapikey = "{API Key}"
faceapiendpoint = "https://japaneast.api.cognitive.microsoft.com/face/v1.0"
2. Face APIへのリクエスト
def detectfacefromdata(apikey, body, endpointorlocation="japaneast", returnFaceId=True, returnFaceLandmarks=False, returnFaceAttributes=[]):
URL生成
params = urllib.parse.urlencode({
"returnFaceId": returnFaceId,
"returnFaceLandmarks": returnFaceLandmarks,
})
if returnFaceAttributes != []:
params["returnFaceAttributes"] = ",".join(returnFaceAttributes)
if "http" in endpointorlocation:
url = "{}/detect?{}".format(endpointorlocation, params)
else:
url = "https://{}.api.cognitive.microsoft.com/face/v1.0/detect?{}".format(endpointorlocation, 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")
dlurl = inputmsg["img_url"]
req = urllib.request.Request(dl_url)
with urllib.request.urlopen(req) as res:
img_bytes = res.read()
実行履歴の取得
outputpath = bindings["outputBlob"]["path"].format(blobpath=inputmsg["blobpath"]).split("/")
container = output_path[0]
blobpath = "/".join(outputpath[1:])
imgname = outputpath[-1]
imgmd5 = hashlib.md5(imgbytes).hexdigest()
画像フォーマット判定
print("Checking input image format")
imgformat = imghdr.what(None, h=imgbytes)
if img_format in ["jpeg", "png", "bmp", "gif"]:
if img_format == "jpeg":
img_format = "jpg"
3. 顔検出結果の取得
print("Detecting Faces")
detectedfaces = detectfacefromdata(faceapikey, imgbytes, location=faceapi_endpoint)
if len(detected_faces):
resultmessage = "%s face(s) datected." % len(detectedfaces)
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,
"blobpath": blobpath,
"filename": img_name,
"contentmd5": imgmd5,
"input": input_msg
}
else:
出力メッセージの設定
result = {
"status": "error",
"filename": img_name,
"contentmd5": imgmd5,
"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()
スクリプトの説明
1. Face APIキー情報
Face API作成時に取得したAPIキーとエンドポイントを設定します。
2. Face APIへのリクエスト
Face APIにリクエストし、画像内の顔を検出します。
今回は、関数でダウンロードした画像の顔検出を行うため、リクエストヘッダの「Content-Type」に「application/octet-stream」を指定し、関数が保持するバイナリデータを直接送信して顔検出を行います。
3. 顔検出結果の取得
Face APIで顔を検出し、検出結果を取得します。
Face APIの顔検出は個々の顔の情報を配列で返すため、配列の要素数を取得してメッセージとしてSlackに出力し、顔検出できていることを確認します。
実行結果
Slackのチャンネルに画像のURLを投稿すると顔の検出数がBotのメッセージで返されました。
同様に画像を5人の子供の写真に変更すると、顔の検出数は5となり、正しく顔を検出できていることが確認できました。
まとめ
Face APIを追加して顔検出ができるようになりました。あとは、検出した顔の情報をもとにコラ画像を作成すれば雑コラBotの完成です。
次回は、今回のコードに画像合成の処理を追加して雑コラBotを完成させます。
