ども！AWS認定SAアソシエイトのまとんです。
最近、LINEボットを作るのにハマっています。
今回、赤ん坊の画像を送ると「育児してて偉い！」と返してくれる「育児応援LINEボット」を作ったので紹介します！
【使ったもの】
・AWS関連：Amazon Rekognition、Lambda、API Gateway
・LINE関連：Messaging API
・開発環境：Windows10、Python 3.7
Amazon Rekognitionを使うと、めちゃくちゃ簡単に画像解析ができました。AI楽しい！！
経緯
うちには0歳の娘がいます。
平日、僕が働いていると、嫁が娘のカワイイ写真をLINEでたくさん送ってくれます。
僕はそれを、トイレ休憩の合間なんかに見てホッコリして、「今日も育児してくれてありがとう！偉い！」と返事をするわけです。
しかし、仕事中なので、すぐに返事をすることはできません。
嫁が写真を送ってから僕が返事をするまでに、若干のタイムラグが生じてしまうのが課題でした。
そこで、娘の写真を送ったら、すぐに「育児してて偉い！」と僕の代わりに返事をしてくれるbotを作れば、嫁はタイムリーに返事が貰えて良いのでは！？と考えました。
（旦那の仕事を、クラウドにアウトソース！これが俺の""働き方改革""）
作りたいもの
作りたいものの要求仕様は下記です。
(1)LINEで写真を送ったら、すぐに返事をしてくれるLINEボット。
(2)写真が娘の写真だったら「偉い！」と褒める。関係の無い写真だったら、褒めない。
(1)については前回の記事で同等のものを実装したので、この記事では(2)について解説します。
highso.hatenablog.com
実装
< 前回の記事での実装分>
僕と嫁とbotの三人が入っているLINEのトークルームを用意します。
LINEのbotはMessagingAPIというサービスで簡単に作ることができます。
LINEのトークルームで誰かが発言すると、botはwebhookで他サービスのAPIを叩くことができます。
他サービスとして、AWSのAPI GatewayのAPIを叩くように設定します。
API Gatewayは、Lambdaを叩くように設定します。
LambdaでMessagint APIのPOSTを捌いて返事をする処理を行います。
<この記事での実装分>
Lambdaでは、まずbotが受け取った画像のデータを、MessagingAPIから受け取ります。
受け取った画像データを、Amazon Rekognitionという分析APIに投げます。
Rekognitionは、AWSによって学習済みの画像解析モデルを使って、識別だけを行ってくれるサービスです。（他にも機能はいっぱいあるかも）
Rekognitionが、画像に含まれるラベル（赤ん坊が含まれている確率90%、など）と、顔認識（推定年齢0～5歳、など）を行って、結果を返してくれます。
結果に応じて、赤ん坊だったら「偉い！」というメッセージをMessaging APIに返します。
botがメッセージを応答します。
赤ん坊の写真と判定する仕組み
Amazon Rekognitionには画像と動画を分類する様々な機能があるのですが、今回はラベル検出の「DetectLabels」と、顔検出と顔分析の「DetectFaces」の二つを使います。
これらを使って、「赤ん坊の画像化かどうか？」を判定します。
まず、DetectLabelsで、画像に含まれるラベルを検出します。
ドンピシャの「Baby」ラベルが信頼度50%以上で検出されれば、OKです。
しかし、画像によっては、うまく「Baby」が検出されず、類似の「Human」「Person」「Newborn」が検出されることもあります。
この場合には、さらにDetectFacesを使って、年齢を推定します。
DetectFacesでは、顔の画像から、「性別」や「感情」など色々な情報を推定します。
この中から「年齢レンジ」だけを使います。
年齢レンジの最大値が「4歳以下」だった場合には、赤ん坊と判定することにしました。
以上の方式をまとめると、
・画像にBaby画像があるか？
・画像に人がいる場合、年齢は4歳以下か？
で判定するようにしました。
この仕組みでうまく判定できない場合は、「赤ん坊が後ろ向きの写真」です。
顔が映っていないと、顔検出で年齢推定ができないので、上手く判定できません。
Lambdaコード
LINE bot～Lambdaまでの通信部分については、前回の記事に書いたので、そちらをご覧ください。（LINEボットライブラリをpipでローカルにインストールして、zipに固めてlambdaにアップロードする必要があります）
ソースコードはGitHubに公開しています。
github.com
以下では、Lambdaコードの部分だけを開設します。
コードは最終的に以下のようになりました。
import json
import sys
import os
import random
from datetime import datetime as dt
from datetime import timedelta
import boto3
from linebot import (
LineBotApi, WebhookHandler
)
from linebot.models import (
MessageEvent, TextMessage, TextSendMessage,
)
from linebot.exceptions import (
LineBotApiError, InvalidSignatureError
)
#(1)Amazon REKOGNITION
rekognition_label = boto3.client('rekognition')
rekognition_face = boto3.client('rekognition')
#(2)LINE Messaging API
LINE_CHANNEL_SECRET = os.getenv('LINE_CHANNEL_SECRET', None)
LINE_CHANNEL_ACCESS_TOKEN = os.getenv('LINE_CHANNEL_ACCESS_TOKEN', None)
if LINE_CHANNEL_SECRET is None:
print('Specify LINE_CHANNEL_SECRET as environment variable.')
sys.exit(1)
if LINE_CHANNEL_ACCESS_TOKEN is None:
print('Specify LINE_CHANNEL_ACCESS_TOKEN as environment variable.')
sys.exit(1)
line_bot_api = LineBotApi(LINE_CHANNEL_ACCESS_TOKEN)
handler = WebhookHandler(LINE_CHANNEL_SECRET)
#(3)Messaging API Handler
@handler.add(MessageEvent)
def message(event):
#(4)LINE message information
event_type = event.type
message_type = event.message.type
message_id = event.message.id
#msg = 'event_type: ' + event_type + ', user_id: ' + user_id + ', message_type: ' + message_type + ', message_id:' + message_id
if message_type == 'image':
#(5)Get image data from LINE
message_content = line_bot_api.get_message_content(message_id)
content = bytes()
for chunk in message_content.iter_content():
content += chunk
#(6)Recognize image by Rekognition
labels = detect_labels(content)
print(labels)
msg_label = ''
baby_flag = False
human_flag = False
for label in labels['Labels']:
msg_label += label['Name'] + ':' + str(round(label['Confidence'],1)) + '% '
if label['Name'] == 'Baby':
baby_flag = True
print('Here is a baby.')
if label['Name'] == 'Baby' or label['Name'] == 'Person' or label['Name'] == 'Human' or label['Name'] == 'Newborn':
human_flag = True
print('Here is a Human.')
msg_label = msg_label[:-1]
#(7)Detect face in image by Rekognition
if human_flag == True: #When a human is detected
FaceDetails = detect_faces(content)
print(FaceDetails)
if len(FaceDetails['FaceDetails']) == 0: #When a face is not detected
human_flag = False
else:
age_min = FaceDetails['FaceDetails'][0]['AgeRange']['Low']
age_max = FaceDetails['FaceDetails'][0]['AgeRange']['High']
msg_face = 'Age:' + str(age_min) + '-' + str(age_max) + 'years old'
if age_max < 5:
baby_flag = True
#(8)Ask LINE bot to reply messages
text = []
if baby_flag == True: #There is a baby
msg_approval = ['可愛い!', '天使!', '偉い!', '育児して偉い！', '最高!', 'すごい！', 'キャワワ!', '天才!', '【大当たり】おめでとうございます！']
text.append(TextSendMessage(text=random.choice(msg_approval)))
else: #There is no baby
text.append(TextSendMessage(text='No baby.'))
text.append(TextSendMessage(text=msg_label))
if human_flag == True: #There is a face
text.append(TextSendMessage(text=msg_face))
line_bot_api.reply_message(event.reply_token, text)
def detect_labels(image_bytes):
try:
res = rekognition_label.detect_labels(
Image={
'Bytes': image_bytes
},
MaxLabels=6,
MinConfidence=50
)
return res
except Exception as ex:
print("fail to detect labels. error = " + ex.message)
def detect_faces(image_bytes):
try:
res = rekognition_face.detect_faces(
Image={
'Bytes': image_bytes
},
Attributes=[
'ALL',
]
)
return res
except Exception as ex:
print("fail to detect faces. error = " + ex.message)
def send_line_bot(signature, body):
ok_json = {"isBase64Encoded": False,
"statusCode": 200,
"headers": {},
"body": ""}
error_json = {"isBase64Encoded": False,
"statusCode": 403,
"headers": {},
"body": "Error"}
try:
handler.handle(body, signature)
except LineBotApiError as e:
print("Got exception from LINE Messaging API: %s " % e.message)
for m in e.error.details:
print(" %s: %s" % (m.property, m.message))
return error_json
except InvalidSignatureError:
return error_json
return ok_json
def lambda_handler(event, context):
signature = event["headers"]["x-line-signature"]
body = event["body"]
res = send_line_bot(signature, body)
print(res)
return res
Pythonの場合、AWSの公式ライブラリboto3を使って、APIを叩くためのインスタンスを簡単に作れます。
(2)LINE Messaging APIのインスタンス生成
Messaging APIのアクセストークン（環境変数に登録）を使って、Messaging APIのインスタンスとハンドラーを作ります。
(3)Messaging APのHandler関数
@handler.add(MessageEvent)
def message(event):
(処理内容)
で、「LINE botがメッセージイベントを受信したとき」に起動する関数を作れます。
「メッセージイベント」とは、テキストや、スタンプ、画像などを含みます。
テキストだけに反応させたいときは、「@handler.add(MessageEvent, message=TextMessage)」とすればよいです。
参考：
GitHub - line/line-bot-sdk-python: LINE Messaging API SDK for Python
Messaging APIリファレンス
(4)LINEのメッセージ情報を取得
eventオブジェクトの中に、メッセージの様々な情報が含まれています。
今回は、event.message.typeで「画像メッセージかどうか」の情報を取得するのと、event.message.idで「メッセージごとに一意に付けられるid」を取得します。
(5)LINEから画像データを取得
画像メッセージだったときに、get_message_content()メソッドを使って、(4)で取得したidに紐づく画像データのバイナリを取得します。
バイナリはfor文でイテレーションできるので、結合して、content変数に画像データの全体を格納します。
(6)Rekognitionで画像のラベル検出
いよいよ、Rekognitionを使って画像を分析していきます。
自前で作ったdetect_labels()では、Rekognitionインスタンスのdetect_labels()メソッドを起動して、RekognitionのAPIに画像を受け渡して、結果をresに取得します。
Rekognitionへの画像データの受け渡し方は2通りあります。
(1)S3に保存済みの画像を使う。S3のキーを渡す。
(2)バイナリを直接送る。
どちらの実装でもよいのですが、(1)の場合には、「S3に画像を一時的に保存→Rekognitionに画像を送る→S3から画像を削除」という手続きを踏む必要があり、めんどうでした。
今回、LINEから画像データをバイナリで受け取っているので、ストレージを経由せず、直接渡す(2)の方式を選びました。
参考：DetectLabels - Amazon Rekognition
detect_labels()では、「最大ラベル取得数」と「信頼度の下限」をJSONで設定できます。
今回は、「最大ラベル数6、信頼度50%」と設定しました。
信頼度が50%以上のラベルの上位6個を返してくれます。
ラベルの識別結果は、labels['Labels']に格納されます。
これをfor文でイテレートすると、['Name']にラベル名が、['Confidence']に信頼度が入っています。
今回、'Baby'ラベルの有る無し、などを調べています。
(7)Rekognitionで顔検出と顔認識
人間のラベル（Baby, Human, Person, Newborn）が含まれていた場合には、さらにdetect_faces()のAPIを叩きます。
APIの使い方はdetect_labels()と同様に、JSONで画像のバイナリデータを送るだけです。カンタン！
結果は['FaceDetails']に格納されますが、注意として、顔が検出されなかった場合には['FaceDetails']に何も格納されません。
なので、「if len(FaceDetails['FaceDetails']) == 0:」でエラー処理を入れています。（他に良い方法をご存知の方は、教えて欲しいです）
顔の検出結果は、様々な属性がありますが、
今回は最大年齢['FaceDetails'][0]['AgeRange']['High']を使って、「4歳以下かどうか」を調べています。
参考：DetectFaces - Amazon Rekognition
(8)結果をLINE botに返す
line_bot_api.reply_message(event.reply_token, text)で、textの内容をLINE botに返事させることができます。
textには、TextSendMessage(text=<文章>)で定義したTextSendMessageオブジェクトを入れます。
textをTextSendMessageのリストにすると、リストの個数分のメッセージをbotが返事することになります。
今回は、
・赤ん坊だったら褒めるメッセージをランダムに1個返す。赤ん坊でなかったら「No baby.」と返す
・detect_labelsで識別したラベルを返す。
・detect_facesで識別した推定年齢を返す。ただし、顔が検出できなかったらナシ。
と返事をするよう実装しました。
ここで、LINE botではテキストを返すのは簡単ですが、画像を返すの難しいです。
画像を返す場合、画像を何らかの形でインターネット公開して、アドレスを渡す必要があります。
例えば、一旦S3に格納して、静的webホスティングでアドレスを与える、などすれば可能かと思います。
本当は、画像に対して「ここが赤ん坊と識別されました」のバウンディングボックスをつけた画像を返したかったのですが、今回は見送りました。
Lambdaの環境変数に、Messaging APIのアクセストークンを2個、登録する必要があります。
LINE_CHANNEL_SECRETとLINE_CHANNEL_ACCESS_TOKENです。
IAMロール
Lambdaには、Rekognitionを使うためのIAMポリシーをアタッチする必要があります。
AmazonRekognitionFullAccessと、CloudWatchLogsFullAccessのポリシーをアタッチしたIAMロールを作成し、このIAMロールをLambdaにアタッチします。
IAMロールが正しく機能していれば、LambdaのDesinger画面ではこのように、アクセス先のリソースにCloudWatchとRekognitionが表示されるはずです。
Lambdaのタイムアウトはデフォルトで3秒に設定されています。
Lambdaは使用した時間単位で課金されます。例えば無限ループに入ってしまうと、無限に課金されてしまうので、タイムアウトの設定は必須です。
今回、Rekognitionの応答に若干の時間がかかるので（特にdetect_facesに数秒かかる）、3秒だとタイムアウトしてしまいます。
10秒に設定すれば、タイムアウトを回避できました。
デモ
無事に、赤ん坊の画像を送ると褒めてくれる「育児応援bot」ができました！！
感想
AI楽しい！！！この一言に尽きます。
Rekognitionを使えば、学習とかめんどくさいステップを全て吹っ飛ばして、結果だけ返してくれるので、ラクチンでした。
最初は「LambdaにTensorFlowをインストールする必要があるのか・・・？」とか思っていましたが、そんな必要はなかったんや。
Rekognitionの精度は、AWSさんに委ねられているので、今後も頑張って精度を上げていってほしいです。
LINEbotを使い始めると、嫁も楽しんで写真を送ってくれるようになりました。
作ったオモチャを家族に使ってもらえるのは楽しいですね。
クラウドアプリなので、モノを管理する必要も無いので、運用が簡単なのもいいです。
みなさんも是非、Rekognitionで遊んでみてください！
以上、メタラーまとんでした。
ではでは。
