はじめに
LINExAWSでほげほげする回の続編です。これまで
i. LINEでLambdaに画像を送信
ii. Rekognitionで顔情報の取得(表情判定)
iii. SageMakerでセマンティックセグメンテーションモデル作成
をしてきました。今回はi, iiiを用いて、背景切り抜きするLINEアプリを作ってみました。具体的には
- LINEでLambdaに画像を送信して
- セマンティックセグメンテーションモデルを用いて背景画像を切り抜き
- 切り抜いた画像をLINEに送り返す
をしてみました。基本的には組み合わせです。
完成物
LINEで画像をおくると、背景が切り抜かれた画像が返却されます。
いぬ。かわゆ。
「あれ、処理後の画質悪くなってね?」はちょっと聞こえません。切り抜き精度もご愛嬌でお願いいたします。
では書いていきます。
#バイナリ画像処理に慣れていないので、冗長な部分もあると思いますが、ご容赦ください...
LINE->Lambdaで画像受信できるようにするconfig
おさらいになりますが、まずはLINEMessagingAPIで送られてきた画像をLambdaで受信できるようにする設定から。
LINE_CHANNEL_ACCESS_TOKEN = os.environ["LINE_CHANNEL_ACCESS_TOKEN"]
#-- Headerの生成
HEADER = {
'Content-type':
'application/json',
'Authorization':
'Bearer ' + LINE_CHANNEL_ACCESS_TOKEN,
}
BUCKET_NAME='xxxxxxxx'
画像を受信してバイナリ化(SageMaker呼び出し用)
画像を読み込んで、SageMakerへ入力できるようにバイナリに変換します。
def lambda_handler(event, context):
body = json.loads(event['body'])
for event in body['events']:
#-- ImageMessageが来た時
if event['message']['type'] == 'image':
#-- 1. 画像を読み込み
MessageId = event['message']['id'] # メッセージID
ImageFile = requests.get('https://api-data.line.me/v2/bot/message/'+ MessageId +'/content',headers=HEADER) #Imagecontent取得
print('get image!')
#-- 2. ImageFile(LINEから送られてきた画像)をPILで読込&pngで保存
img = Image.open(BytesIO(ImageFile.content))
img.save('/tmp/receive.png','PNG')
#-- 3. bytesioに変換
img_bytes = BytesIO()
img.save(img_bytes, format='PNG',optimize=True,quality=80)
image_data = img.getvalue() #これが bytes
1はLINEからの画像取得。 2はPILで読み込んでreceive.pngとして保存しておきます(後で使用)。3はSageMakerに入力するためにbyteにします。
SageMakerで学習したモデル を呼び出して推論
さて、画像が用意できたところで、いつぞやの記事で学習したモデルを使って推論してみます!
#-- 4. sagemakerにアクセスする
runtime = boto3.client("sagemaker-runtime", region_name="your-region")
#-- 5. 作成したエンドポイント名称
endpoint_name='semantic-segmentation'
#-- 6. 指定したエンドポイントにデータを渡す
res = runtime.invoke_endpoint(EndpointName=endpoint_name,
Body=image_data,
ContentType='image/jpeg',
Accept='image/png'
)
#-- 7. 推論結果の画像情報を抽出
body = res['Body'].read()
image_decoded = bytearray(body) #-- ここでハマったー!bytearrayとすること。
#-- 8. analysis.pngとして保存
with open('/tmp/analysis.jpg',"wb") as f:
f.write(image_decoded)
pil_img = Image.open('/tmp/analysis.jpg','r')
pil_img.save('/tmp/analysis.png', 'PNG')
ここで5はSageMakerで作成したエンドポイント名を入力します。6ではjpegもしくはpngを読み込むように設定しています。読み込み対象のimage_dataは前述の3で作っています。
7で推論結果の画像情報を抽出して、8でpngに保存しています(jpg->pngのやり方が遠回りな気がする...)
背景の切り抜き
さて、入力画像(receive.png)と推論画像(analysis.png)を用いて、背景の切り抜きをしてみます。
推論画像をちゃんとしたマスク画像にするために、ちょいと画像処理します。
'''
検出物体(黒色以外)を白色に変換
'''
#-- i1. 推論結果をRGBAに変換してRGB分ける
org = Image.open('/tmp/analysis.png')
rgb_img = org.convert('RGBA')
r, g, b,_ = rgb_img.split()
#-- i2. RGB各々において、色がないところは0,それ以外は1
_r = r.point(lambda _: 0 if _ == 0 else 1, mode="1")
_g = g.point(lambda _: 0 if _ == 0 else 1, mode="1")
_b = b.point(lambda _: 0 if _ == 0 else 1, mode="1")
#-- i3. mask(111->1, 000->0)
mask = ImageChops.logical_or(_r, _g)
mask = ImageChops.logical_or(mask, _b)
trans=mask.convert('RGBA')
trans.save('/tmp/mask.png')
#-- i4. 白黒をグレイスケール変換
mask_gry=Image.open('/tmp/mask.png').convert('L').resize(img_input_png.size) #Grayスケール変換&resize
#-- i5. img_input_png+img_msk
img_input_png=Image.open('/tmp/receive.png')
img_mask=Image.open('/tmp/mask.png')
im=Image.composite(img_input_png,img_mask,mask_gry)
#-- i6. 背景を透過
transparent = Image.new("RGBA", im.size, (0, 0, 0, 0))
transparent.paste(im,(0,0),mask_gry.split()[0])
i2,i3で検出物体の画素を白色に、そのほかを黒色に変換しています。こちらのサイトを参考にさせていただきました。
i2 では、RGB各々に対して、色がないところは0, それ以外は1にする処理をしています。mode=1にすることで、色深度が1ビットの画像生成されます。
i3 では、各々のorをとっています。つまり、RGBどれかひとつでも1があれば残し、全て0であれば要らない子(背景)とみなします(「すべて1の画素を残す」でもそれなりに分離できてた。このあたりの最適解が分からない...)。
i4 では、合成時にマスクするために、生成されたmask画像に対しグレイスケール変換を施します。
i5 では、入力画像と白黒マスク画像 を合成しています。引数が3つありますが、これは「img_input_pngとimg_maskを合成するよ、ただしmask_gryでマスクするよ」という意味です。これによって「切り抜き」が実現されます。
i6 で最後に背景を透過しています。
ちなみに、入力画像(LINEから受信)、マスク画像(~i3で作成)、出力(i5,i6で作成)はそれぞれこんな感じです。
いぬ。かわゆ。
LINEに送り返す
これにも一手間必要です。画像をS3バケットに保存し、署名付きのURLを取得する必要があります。
S3に保存された画像をLINE Messaging APIで取得する時に、時間制限をつけてアクセス可能にします。時間制限付きのIAM Roleがつくイメージですね。
#-- 9. resultをS3に保存(.jpg)
filename_output='output.jpg'
image_bytes = BytesIO()
transparent.save(image_bytes, format="png")
image_bytes = image_bytes.getvalue()
obj = s3.Object(BUCKET_NAME,filename_output)
obj.put( Body=image_bytes)
#-- 10. S3へアップロードした画像の署名付きURLを取得する
s3_client = boto3.client('s3')
s3_image_url = s3_client.generate_presigned_url(
ClientMethod = 'get_object',
Params = {'Bucket': BUCKET_NAME, 'Key': file_name_output},
ExpiresIn = 10,
HttpMethod = 'GET'
)
#-- 11. 署名付きURLを用いてimage_urlを送る
REQUEST_MESSAGE = [
{
'type': 'image',
'originalContentUrl': s3_image_url,
'previewImageUrl': s3_image_url,
},
]
payload = {'replyToken': event['replyToken'], 'messages': REQUEST_MESSAGE}
10.のExpiresInで、有効時間(秒)を決めています。
これでLINEに送信できました!
まとめ
今回はLINEMessagingAPIと、SageMakerを使って、「写真送ったら切り抜いて送り返してくれる」簡単なアプリを作ってみました。
普段は音声屋の身分ですが、画像の信号処理も勉強になりました。
Rekognitionと組み合わせて、証明写真ジェネレータとか作れそう。