機械学習を使って魚の画像判定アプリを作ってみた !

import os
import logging
import tempfile
import boto3
from linebot import LineBotApi, WebhookHandler
from linebot.models import TextSendMessage

s3 = boto3.resource('s3')
# 環境変数を取得
channel_access_token = os.getenv('CHANNEL_ACCESS_TOKEN')
backet_name = os.getenv('BUCKET_NAME')
line_bot_api = LineBotApi(channel_access_token)

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)

def lambda_handler(event, context):
    logger.info('image event')
    imageid = event["imageid"]
    replytoken = event["replytoken"]
    user_id = event["user_id"]
    image = line_bot_api.get_message_content(imageid)

    with tempfile.TemporaryFile() as tmp:
        for chunk in image.iter_content():
            tmp.write(chunk)
        tmp.seek(0)

    bucket = s3.Bucket(backet_name)
    bucket.put_object(
            Body=tmp,
            Key=f'{imageid}.png'
        )
    line_bot_api.reply_message(
            replytoken,
            TextSendMessage(text='アップロードに成功しました')
        )
    return {
        "bucketname": backet_name,
        "imagename": str(imageid) + ".png",
        "user_id": user_id
    }

{
    "StartAt": "LINEImageS3",
    "States": {
        "LINEImageS3": {
            "Type": "Task",
            "Resource": "${LineS3Function}",
            "Next": "RekognitionDB",
        },
        "RekognitionDB": {
            "Type": "Task",
            "Resource": "${RekognitionFunction}",
            "Next": "LineReply",
        },
        "LineReply": {
            "Type": "Task",
            "Resource": "${LineReplyFunction}",
            "End": true
        }
    }
}

import json
import os
from types import BuiltinMethodType
import datetime
import base64
import hashlib
import hmac
import boto3
from linebot import LineBotApi, WebhookHandler
from linebot.models import TextSendMessage

client = boto3.client('stepfunctions')

# 環境変数を取得
channel_secret = os.getenv('CHANNEL_SECRET')
channel_access_token = os.getenv('CHANNEL_ACCESS_TOKEN')
statemachine_arn = os.getenv('STATEMACHINE_ARN')
line_bot_api = LineBotApi(channel_access_token)

def lambda_handler(event, context):
    # 署名の検証(後述)
    body = event["body"]
    signature = event["headers"]["x-line-signature"]
    hash = hmac.new(channel_secret.encode('utf-8'),
                    body.encode('utf-8'), hashlib.sha256).digest()
    if signature != base64.b64encode(hash).decode():
        return {
            "statusCode": 401,
            "body": 'Unauthorized'
        }

    now = datetime.datetime.now().strftime("%m-%d-%Y-%H-%M-%S")
    jsonbody = json.loads(body)
    imageid = jsonbody["events"][0]["message"]["id"]
    replytoken = jsonbody["events"][0]["replyToken"]
    user_id = jsonbody["events"][0]["source"]["userId"]

    type = jsonbody["events"][0]["message"]["type"]
    if type == "image":
        client.start_execution(
            stateMachineArn=statemachine_arn,
            name="job-" + str(now),
            input=json.dumps({
                "imageid": imageid,
                "replytoken": replytoken,
                "user_id": user_id
            })
        )
    # きたメッセージが画像以外の場合、ステートマシンは実行せず、「判定したい魚の写真を送ってください」と返信して終了。
    else:
        line_bot_api.reply_message(replytoken, TextSendMessage(
            text='判定したい魚の写真を送ってください'))
    return {'body': 'ok'}

import os
import boto3

# 環境変数を取得
project_arn = os.getenv('PROJECT_ARN')
table_name = os.getenv('TABLE_NAME')

rekognition = boto3.client('rekognition')
dynamodb = boto3.client('dynamodb')

def lambda_handler(event, context):
    response = rekognition.detect_custom_labels(
        ProjectVersionArn=project_arn,
        Image={
            'S3Object': {
                'Bucket': event["bucketname"],
                'Name': event["imagename"],
            }
        },
    )
    user_id = event["user_id"]

    if not response["CustomLabels"]:
        message = "この写真ではうまく判定できませんでした・・・。"

    else:
        resultname = response['CustomLabels'][0]['Name']
        resultpercent = response['CustomLabels'][0]['Confidence']
        user_id = event["user_id"]

        response = dynamodb.get_item(
            TableName=table_name,
            Key={
                "FishName": {"S": resultname}
            }
        )
        item = response["Item"]
        per = round(resultpercent, 2)
        env = item["Env"]["S"]
        bool = item["bool"]["S"]

        message = '''**判定結果**

名前 : {0}

信頼度 : {1}%

生息地 : {2}

人への影響 : {3}'''.format(resultname, str(per), env, bool)

    return {
        "user_id": user_id,
        "message": message
    }

import os
from linebot import LineBotApi, WebhookHandler
from linebot.models import TextSendMessage

# 環境変数を取得
channel_access_token = os.getenv('CHANNEL_ACCESS_TOKEN')
line_bot_api = LineBotApi(channel_access_token)

def lambda_handler(event, context):
    line_bot_api.push_message(event["user_id"], TextSendMessage(
        text=event["message"]))

def lambda_handler(event, context):
    body = event["body"]
    signature = event["headers"]["x-line-signature"]
    hash = hmac.new(channel_secret.encode('utf-8'),
                    body.encode('utf-8'), hashlib.sha256).digest()
    if signature != base64.b64encode(hash).decode():
        return {
            "statusCode": 401,
            "body": 'Unauthorized'
        }

{
    "StartAt": "LINEImageS3",
    "States": {
        "LINEImageS3": {
            "Type": "Task",
            "Resource": "${LineS3Function}",
            "Next": "RekognitionDB",
            "Catch": [
              {
                  "ErrorEquals": [
                      "States.ALL"
                  ],
                  "Next": "FailFunction",
                  "ResultPath": null
              }
            ]
        },
        "FailFunction": {
            "Type": "Task",
            "Resource": "${FailFunction}",
            "Next": "LineReply"
        },
        "RekognitionDB": {
            "Type": "Task",
            "Resource": "${RekognitionFunction}",
            "Next": "LineReply",
            "Catch": [
                {
                    "ErrorEquals": [
                        "States.ALL"
                    ],
                    "Next": "FailFunction",
                    "ResultPath": null
                }
            ]
        },
        "LineReply": {
            "Type": "Task",
            "Resource": "${LineReplyFunction}",
            "End": true
        }
    }
}

def lambda_handler(event, context):
    return{
        "user_id": event["user_id"],
        "message": "エラーが起きました。"
    }