【LINEbot】Pythonで女優識別botを作ろう Part5【Bot作成編】

こんにちは！

「女優識別botを作ろう」の第５回目かつ最終回となる今回は、前回までに作成した画像識別モデルをLINEbot化する方法を解説していきます。

今回の内容

推論用プログラム作成
モデルのLINEbot化

第４回目の記事はこちら。

>>【LINEbot】Pythonで女優識別botを作ろう Part4【モデル構築編】

では、さっそく見ていきましょう。

LINEbotの基本的な使い方を押さえていない方は、先にこちらの記事を読むことを推奨します。

>>【入門用】PythonによるLINEbot作り方

1 全体のイメージ
2 推論用コード
3 サーバー側のコード
4 おわりに

全体のイメージ

まずは、いきなりコードを確認する前に、全体的な処理手順について軽く確認していきます。

ざっくりしたイメージはこんな感じになります。

実際はサーバー側の処理が面倒だったりするんですけど、まあ大体こんな感じだと思ってもらえればOKです。

ちなみに、今回作成するLINEbotプログラムでは、サーバー側のプログラム（app.py）と推論用プログラム（pred.py）は別々のファイルとして作成します。

上の図で言うところの右の２つが別々のファイルで処理を行います。

サーバー側のファイルで推論用のファイルをインポートするだけですね。

では、次の章から実際のコードを見ていきます。

推論用コード

ここでは、推論用コードを確認します。

# ライブラリのインポート
import numpy as np
from PIL import Image, ImageFilter

import torch
import torch.nn as nn

from torchvision.models import vgg16
from torchvision import transforms


data_labels = ("橋本環奈", "石原さとみ", "深田恭子", "新垣結衣", "本田翼")
MODEL_PATH = "./model/model.pt"   # 学習済みパラメータのパスを指定

size = 224
std = (0.281, 0.275, 0.274)
mean = (0.591, 0.520, 0.491)


# 中央上部を切り取る（前処理）
class CenterTopCrop:
    def __init__(self, size):
        self.size = size

    def __call__(self, img):
        img = np.array(img)
        shape = img.shape
        h, w = shape[0], shape[1]
        padding_side = int((w - self.size) / 2)
        img = img[:self.size, padding_side:w-padding_side]
        if img.shape[1] == 225:
            img = img[:, 1:]

        return Image.fromarray(np.uint8(img))


class Net(nn.Module):
    def __init__(self):
        super(Net, self).__init__()
        self.model = vgg16(pretrained=True)
        self.classifier = nn.Linear(1000, 5)

    def forward(self, x):
        x = self.model(x)
        x = self.classifier(x)
        return x


def pred_actress(img_path):
    img = Image.open(img_path)
    transformed = transform(img)  # 前処理をかける
    transformed.unsqueeze_(0)  # 次元の追加

    if torch.cuda.is_available():
        transformed = transformed.to("cuda:0")

    with torch.no_grad():
        calc_score = model(transformed)

    sum = torch.sum(np.exp(calc_score), dim=1)
    max = torch.max(np.exp(calc_score), dim=1)
    y_label = max.indices
    score = max.values / sum
    act_name = data_labels[y_label]

    return act_name, score


transform = transforms.Compose([
    transforms.Resize(size),   # リサイズ
    CenterTopCrop(size),   # リサイズした画像の中央の正方形を切り取る
    transforms.ToTensor(),   # Tensor型に変換
    transforms.Normalize(mean=mean, std=std)   # 標準化
])

if torch.cuda.is_available():
    device = "cuda:0"
    model = Net().to(device)
    model.load_state_dict(torch.load(MODEL_PATH, map_location=device))
else:
    model = Net()
    model.load_state_dict(torch.load(MODEL_PATH, map_location="cpu"))

# 推論モードに切り替える
model.eval()

# ライブラリのインポート

import numpy as np

from PIL import Image, ImageFilter

import torch

import torch.nn as nn

from torchvision.models import vgg16

from torchvision import transforms

data_labels = ("橋本環奈", "石原さとみ", "深田恭子", "新垣結衣", "本田翼")

MODEL_PATH = "./model/model.pt" # 学習済みパラメータのパスを指定

size = 224

std = (0.281, 0.275, 0.274)

mean = (0.591, 0.520, 0.491)

# 中央上部を切り取る（前処理）

class CenterTopCrop:

def __init__(self, size):

self.size = size

def __call__(self, img):

img = np.array(img)

shape = img.shape

h, w = shape[0], shape[1]

padding_side = int((w - self.size) / 2)

img = img[:self.size, padding_side:w-padding_side]

if img.shape[1] == 225:

img = img[:, 1:]

return Image.fromarray(np.uint8(img))

class Net(nn.Module):

def __init__(self):

super(Net, self).__init__()

self.model = vgg16(pretrained=True)

self.classifier = nn.Linear(1000, 5)

def forward(self, x):

x = self.model(x)

x = self.classifier(x)

return x

def pred_actress(img_path):

img = Image.open(img_path)

transformed = transform(img) # 前処理をかける

transformed.unsqueeze_(0) # 次元の追加

if torch.cuda.is_available():

transformed = transformed.to("cuda:0")

with torch.no_grad():

calc_score = model(transformed)

sum = torch.sum(np.exp(calc_score), dim=1)

max = torch.max(np.exp(calc_score), dim=1)

y_label = max.indices

score = max.values / sum

act_name = data_labels[y_label]

return act_name, score

transform = transforms.Compose([

transforms.Resize(size), # リサイズ

CenterTopCrop(size), # リサイズした画像の中央の正方形を切り取る

transforms.ToTensor(), # Tensor型に変換

transforms.Normalize(mean=mean, std=std) # 標準化

])

if torch.cuda.is_available():

device = "cuda:0"

model = Net().to(device)

model.load_state_dict(torch.load(MODEL_PATH, map_location=device))

else:

model = Net()

model.load_state_dict(torch.load(MODEL_PATH, map_location="cpu"))

# 推論モードに切り替える

model.eval()

既にモデル構築編までの記事を読んだ方であれば、このコードの説明はほとんど必要ないかと思います。

なので、ここではポイントだけを説明します。

49行目に pred_actress という関数があります。

この関数の引数として画像配列を渡すことで、関数内で推論を行っています。

処理の流れとしては、まず受け取った画像配列に前処理をかけ、次に学習済みモデルに入力します。

出力されたスコアのsoftmaxを計算することで、推論結果を確率に変換しています。

これにより推論結果が表す女優名と、その確率を返り値として吐き出します。

内容自体は特に難しいことは行っていないため、こちらのコードの説明はここまでとします。

サーバー側のコード

では、ここではサーバー側のコードを確認します。

import config
import pred
import os
import numpy as np
from pathlib import Path
from flask import Flask, request, abort
from linebot import (
    LineBotApi, WebhookHandler
)
from linebot.exceptions import (
    InvalidSignatureError
)
from linebot.models import (
    MessageEvent, TextMessage, TextSendMessage, ImageMessage, ImageSendMessage
)


SAVE_DIR = "./static/images"   # 一時的に画像を保存するディレクトリ名
if not os.path.isdir(SAVE_DIR):
    os.mkdir(SAVE_DIR)

app = Flask(__name__, static_folder="static", static_url_path="")

line_bot_api = LineBotApi(config.LINE_CHANNEL_ACCESS_TOKEN)  # config.pyで設定したチャネルアクセストークン
handler = WebhookHandler(config.LINE_CHANNEL_SECRET)  # config.pyで設定したチャネルシークレット

message_temp = "name: {}\nscore: {}%"
SRC_IMG_PATH = "static/images/{}.jpg"


# 画像の保存
def save_img(message_id, src_img_path):
    # message_idから画像のバイナリデータを取得
    message_content = line_bot_api.get_message_content(message_id)
    with open(src_img_path, "wb") as f:
        # バイナリを1024バイトずつ書き込む
        for chunk in message_content.iter_content():
            f.write(chunk)


# 予測スコアを算出
def calc_score(src_img_path):
    pred_name, pred_score = pred.pred_actress(src_img_path)
    pred_score = str(np.round(pred_score.numpy() * 100, 1))[1:-1]

    reply = message_temp.format(pred_name, pred_score)
    return reply


@app.route("/")
def index():
    return "Hello World!"


@app.route("/callback", methods=['POST'])
def callback():
    # get X-Line-Signature header value
    signature = request.headers['X-Line-Signature']

    # get request body as text
    body = request.get_data(as_text=True)
    app.logger.info("Request body: " + body)

    # handle webhook body
    try:
        handler.handle(body, signature)
    except InvalidSignatureError:
        print("Invalid signature. Please check your channel access token/channel secret.")
        abort(400)

    return 'OK'


# 画像を受け取った際の処理
@handler.add(MessageEvent, message=ImageMessage)
def handle_image(event):
    message_id = event.message.id
    src_img_path = SRC_IMG_PATH.format(message_id)   # 保存する画像のパス

    save_img(message_id, src_img_path)   # 画像を一時保存する
    reply = calc_score(src_img_path)   # LINE側での出力内容を取得

    # LINE側で推論結果を出力する
    line_bot_api.reply_message(
        event.reply_token,
        TextSendMessage(text=reply)
    )

    # 一時保存していた画像を削除
    Path(SRC_IMG_PATH.format(message_id)).absolute().unlink()


if __name__ == "__main__":
    # Flaskを起動
    app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 5000)))

import config

import pred

import os

import numpy as np

from pathlib import Path

from flask import Flask, request, abort

from linebot import (

LineBotApi, WebhookHandler

)

from linebot.exceptions import (

InvalidSignatureError

)

from linebot.models import (

MessageEvent, TextMessage, TextSendMessage, ImageMessage, ImageSendMessage

)

SAVE_DIR = "./static/images" # 一時的に画像を保存するディレクトリ名

if not os.path.isdir(SAVE_DIR):

os.mkdir(SAVE_DIR)

app = Flask(__name__, static_folder="static", static_url_path="")

line_bot_api = LineBotApi(config.LINE_CHANNEL_ACCESS_TOKEN) # config.pyで設定したチャネルアクセストークン

handler = WebhookHandler(config.LINE_CHANNEL_SECRET) # config.pyで設定したチャネルシークレット

message_temp = "name: {}\nscore: {}%"

SRC_IMG_PATH = "static/images/{}.jpg"

# 画像の保存

def save_img(message_id, src_img_path):

# message_idから画像のバイナリデータを取得

message_content = line_bot_api.get_message_content(message_id)

with open(src_img_path, "wb") as f:

# バイナリを1024バイトずつ書き込む

for chunk in message_content.iter_content():

f.write(chunk)

# 予測スコアを算出

def calc_score(src_img_path):

pred_name, pred_score = pred.pred_actress(src_img_path)

pred_score = str(np.round(pred_score.numpy() * 100, 1))[1:-1]

reply = message_temp.format(pred_name, pred_score)

return reply

@app.route("/")

def index():

return "Hello World!"

@app.route("/callback", methods=['POST'])

def callback():

# get X-Line-Signature header value

signature = request.headers['X-Line-Signature']

# get request body as text

body = request.get_data(as_text=True)

app.logger.info("Request body: " + body)

# handle webhook body

try:

handler.handle(body, signature)

except InvalidSignatureError:

print("Invalid signature. Please check your channel access token/channel secret.")

abort(400)

return 'OK'

# 画像を受け取った際の処理

@handler.add(MessageEvent, message=ImageMessage)

def handle_image(event):

message_id = event.message.id

src_img_path = SRC_IMG_PATH.format(message_id) # 保存する画像のパス

save_img(message_id, src_img_path) # 画像を一時保存する

reply = calc_score(src_img_path) # LINE側での出力内容を取得

# LINE側で推論結果を出力する

line_bot_api.reply_message(

event.reply_token,

TextSendMessage(text=reply)

)

# 一時保存していた画像を削除

Path(SRC_IMG_PATH.format(message_id)).absolute().unlink()

if __name__ == "__main__":

# Flaskを起動

app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 5000)))

ちょっと長いですが、ブロックごとで役割が分担されているので順を追って見ていきましょう。

まずは各関数の役割を確認した後、どのようにプログラムが動くのかを確認します。

各関数の役割

save_img ：画像の保存を行う
calc_score： LINEbotで出力する内容を取得する
handle_image：画像受け取ると呼び出される関数。
内部で上の２つの関数を実行する。

プログラムの流れは以下のようになります。

ちなみに、これが最初に紹介したイメージ図の、サーバ側で行う処理になります。

ごちゃごちゃしてる気もしますが、単純に考えるとhandle_image関数内で受け取った画像の予測結果を求めて、それをLINEbotで表示しているだけです。

今はよくわからなくても、コードを何度か見てみればわかると思います。

ちなみに、画像を一時保存するのは、LINEbotから送られてきたデータの形式だとうまくモデルへ入力できないため、一度画像データとして保存しています。

では、ここからは各関数の中身を見ていきます。

まずは、32~38行目にある save_img関数から見てみましょう。

def save_img(message_id, src_img_path):
    # message_idから画像のバイナリデータを取得
    message_content = line_bot_api.get_message_content(message_id)
    with open(src_img_path, "wb") as f:
        # バイナリを1024バイトずつ書き込む
        for chunk in message_content.iter_content():
            f.write(chunk)

def save_img(message_id, src_img_path):

# message_idから画像のバイナリデータを取得

message_content = line_bot_api.get_message_content(message_id)

with open(src_img_path, "wb") as f:

# バイナリを1024バイトずつ書き込む

for chunk in message_content.iter_content():

f.write(chunk)

これは、 message_content = line_bot_api.get_message_content(message_id) の部分で、受信したデータ（ここでは画像データ）を message_contentという変数に格納しています。

その後、 with open()を用いて指定したパスに画像データを書き込みます。

これでひとまず画像の一時保存は完了です。

次に、42~47行目にある calc_score 関数を見ていきます。

def calc_score(src_img_path):
    pred_name, pred_score = pred.pred_actress(src_img_path)
    pred_score = str(np.round(pred_score.numpy() * 100, 1))[1:-1]
 
    reply = message_temp.format(pred_name, pred_score)
    return reply

def calc_score(src_img_path):

pred_name, pred_score = pred.pred_actress(src_img_path)

pred_score = str(np.round(pred_score.numpy() * 100, 1))[1:-1]

reply = message_temp.format(pred_name, pred_score)

return reply

まず、 pred_name, pred_score = pred.pred_actress(src_img_path) の部分で、先ほど保存した画像から女優名と、その確率をそれぞれ取得します。

pred_actress 関数は、推論用コードで紹介した関数ですね。

それぞれの情報を取得したら、すぐ下の行で確率を％表記に変換しています。

あとは、あらかじめ用意していたテンプレートに値を代入して、返り値として渡します。

最後は75~90行目にある handle_image 関数を見ていきます。

見ると言っても、LINEbot側で文字を出力するだけなので、先にLINEbotの記事を読んでいればさほど難しいことはありません。

@handler.add(MessageEvent, message=ImageMessage)
def handle_image(event):
    message_id = event.message.id
    src_img_path = SRC_IMG_PATH.format(message_id)   # 保存する画像のパス
 
    save_img(message_id, src_img_path)   # 画像を一時保存する
    reply = calc_score(src_img_path)   # LINE側での出力内容を取得
 
    # LINE側で推論結果を出力する
    line_bot_api.reply_message(
        event.reply_token,
        TextSendMessage(text=reply)
    )
 
    # 一時保存していた画像を削除
    Path(SRC_IMG_PATH.format(message_id)).absolute().unlink()

@handler.add(MessageEvent, message=ImageMessage)

def handle_image(event):

message_id = event.message.id

src_img_path = SRC_IMG_PATH.format(message_id) # 保存する画像のパス

save_img(message_id, src_img_path) # 画像を一時保存する

reply = calc_score(src_img_path) # LINE側での出力内容を取得

# LINE側で推論結果を出力する

line_bot_api.reply_message(

event.reply_token,

TextSendMessage(text=reply)

)

# 一時保存していた画像を削除

Path(SRC_IMG_PATH.format(message_id)).absolute().unlink()

80，81行目で上で見た２つの関数を使って、すでに出力用のテキストを生成しています。

後は、84~87行目でLINEbot側にテキストを送信して終わりです。

これでLINEbot側では、画像から予測した女優の名前とその確率が表示されます。

これでbot作成は終了です！

おわりに

ここまで全部で５つに分けて「女優識別bot」の解説をしてきましたが、いかがだったでしょうか。

コードを全て解説しようとすると、かなりの労力になるためところどころ抜粋したのですが、ポイントは押さえてきたつもりです。

今回扱ったものは「女優識別」という限定的なものですが、自分でオリジナルbotを作る場合でも基本的な考え方は同じなので是非試してみてください。

では、お疲れさまでした。

P.S.

このコードを書いた時期はまだPyTorchの基本的な使い方しか知らない状態だったので、今見返すと汚いコードがいくつかありました、、

余裕があるときに修正します。