[PYTHON] Du masque blanc au masque de monstre, développement de l'application de traitement de photos LINE sans serveur sur AWS

0. introduction

Ravi de vous rencontrer, je suis Pong de Chine. Je suis un nouvel ingénieur travaillant au Nomura Research Institute. Je ne suis toujours pas bon en japonais, alors pardonnez-moi si vous avez un japonais étrange. Merci beaucoup.

Il y a trop de masques blancs pour les photos de voyage pendant la période Corona, Vous ne pouvez plus le supporter, non? En ce moment, il y a de la formation pour les nouveaux arrivants, et j'en ai fait un sujet de formation. Pour résoudre ce problème, nous avons développé une application qui convertit le masque blanc de la photo en masque de monstre. Basé sur le concept de "réduire autant que possible la quantité de travail" Il a été développé en tant qu'application de traitement de photos LINE sans serveur en utilisant divers services AWS. ** Pour ceux qui en ont assez des photos masquées ** et ** pour ceux qui s'intéressent au sans serveur ** Veuillez apprécier ce rapport de développement.

1. 1. Pourquoi développer cette application?

J'ai voyagé avec elle à Choshi à Chiba pendant les vacances d'été. J'ai joué dans la mer, escaladé le phare et pris beaucoup de photos commémoratives. Mais malheureusement, le protagoniste de la photo n'était pas un humain ou un paysage, mais un masque blanc. Sur les photos de l'ère Corona (temps), le taux d'apparition des masques blancs est le plus élevé, et ils apparaissent partout. Quand elle a vu une telle image, elle s'est plainte de ne plus vouloir voir le masque blanc, alors pourquoi ne pas convertir le masque blanc de l'image en autre chose?

Tout comme elle et moi aimons les films de super-héros, j'adore le masque de monstre (par exemple, le monstre de Batman, Bane). Ne serait-il pas agréable que le masque blanc devienne un masque de monstre?

※This work is a derivative of "[Bane](https://www.flickr.com/photos/istolethetv/30216006787/)"by[istolethetv](https://www.flickr.com/people/istolethetv/),usedunder[CCBY2.0](https://creativecommons.org/licenses/by/2.0/)

C'est pourquoi j'ai eu l'idée et j'ai décidé de développer cette application de traitement de photos. Mais il y a trois problèmes devant moi.

Tout d'abord, il existe différentes options pour le formulaire de demande. Application Web en tant que page Web? Une application iOS ou Android pour smartphones uniquement? Vous devez concevoir non seulement le traitement backend mais également l'interface frontend. Compte tenu de diverses choses, je pense que l'application LINE est la plus appropriée. Il y a trois raisons:

  1. Facile à utiliser: presque tout le monde a LINE, et l'application LINE (bot) est très facile à envoyer et à recevoir, et tout le monde peut l'utiliser.
  2. Moins de travail: pour les applications Web et iOS, c'est ennuyeux pour être honnête, y compris la conception de l'interface. Mais avec l'application LINE, vous n'avez pas à y penser et ce sera plus facile.
  3. Facile à partager: La caractéristique de SNS est qu'il est facile à partager. Non seulement les photos converties, mais aussi cette application sera plus facile à partager.

Par conséquent, ** j'ai décidé d'utiliser l'application LINE comme formulaire de candidature! ** **

Et le prochain défi est de savoir où configurer le serveur Sera-t-il construit sur une machine physique telle que Raspberry pi? Utilisez-vous un serveur cloud tel qu'AWS EC2? De plus, le serveur a besoin non seulement de construction, mais aussi de gestion de la maintenance plus tard. Je suis paresseux et j'ai l'idée de "réduire le plus possible la quantité de travail", donc je ne veux pas faire ça. .. .. Alors, pourquoi ne pas développer sans serveur sans avoir besoin d'un serveur? D'après mes recherches, ** Avec AWS API Gateway et Lambda, le sans serveur peut être réalisé, et il n'y a pas de construction de serveur ni de gestion de la maintenance **! D'accord, j'ai choisi toi! !!

Enfin, cette fois, nous avons besoin d'une IA de reconnaissance faciale pour traiter la photo du visage. En conséquence, des questions telles que "quelle structure de modèle d'IA utilisez-vous?", "Où obtenez-vous les données d'entraînement?" Et "quel type d'étiquette voulez-vous étiqueter les données?" Je me suis dit "J'aurais aimé avoir une IA de reconnaissance faciale qui puisse être utilisée immédiatement", alors j'ai cherché sur AWS et les résultats sont vraiment ressortis! Il existe un service AWS qui analyse des images ou des vidéos appelé Rekognition (pas de reconnaissance). ** Pas besoin de créer une IA, appelez simplement Rekognition pour reconnaître et analyser le visage sur la photo **. Avec cela, vous pouvez réaliser "réduire la quantité de travail autant que possible".

Dans cet esprit, nous avons décidé de développer une application de traitement de photos LINE sans serveur sur AWS!

2. Présentation du système

Nous avons déjà décidé du formulaire de candidature, alors construisons le système maintenant! L'image globale du système créé cette fois est la suivante:

Ici, on suppose que la communication avec l'utilisateur est un smartphone. (La version PC LINE est également disponible) Le frontal est LINE Bot. Tous les backends sont traités par AWS Cloud. Pour réaliser un traitement sans serveur, le traitement est effectué sur trois Lambda: "contrôleur", "reconnaissance faciale" et "nouvelle génération d'image". Compte tenu du flux de traitement, ce système peut être divisé en 5 parties comme le montre la figure ci-dessous:

Ensuite, j'expliquerai ces cinq parties à partir du flux de traitement.

3. 3. Explication pour chaque partie

3-1 Partie d'entrée d'image

Flux de processus

La première partie est la partie d'entrée. La fonction est littéralement de charger l'image que l'utilisateur a envoyée au Bot LINE. Les entités liées à cette partie sont "LINE Bot", "API Gateway" et "Controller Lambda". Le flux de traitement est le suivant:

Tout d'abord, l'utilisateur envoie une image photo à LINE Bot. Le bot LINE enveloppe ensuite l'image dans line_event et l'envoie à la passerelle API. API Gateway envoie un événement au contrôleur Lambda sans aucune modification.

Créer un robot LINE

Pour créer cette pièce, créez d'abord un Bot LINE (messagingApi) comme porte d'entrée. Cliquez ici pour savoir comment faire: LINE Official Document: Get Started with Messaging API Après avoir créé le canal, deux paramètres sont encore nécessaires. La première consiste à émettre un «jeton d'accès au canal» pour l'authentification avec Lambda. La seconde consiste à désactiver la fonction de réponse de l'API de messagerie et à activer la fonction webhook. N'entrez pas l'URL du webhook maintenant, mais après avoir configuré API Gateway.

Créer un contrôleur Lambda

Vient ensuite la création d'un rôle IAM qui exécute des services tels que Lambda. Entrez le service IAM à partir du tableau de bord et créez un nouveau rôle. Le nouveau rôle IAM nomme serverless-linebot etc. et le service utilisé est Lambda. Les stratégies sont «Amazon S3FullAccess», «AmazonRekognitionFullAccess» et «CloudWatchLogsFullAccess». De plus, puisque le contrôleur Lambda appelle un autre Lambda, ajoutez la stratégie suivante:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor0",
            "Effect": "Allow",
            "Action": [
                "lambda:InvokeFunction",
                "lambda:InvokeAsync"
            ],
            "Resource": [
                "Reconnaissance faciale Lambda ARN",
                "Nouvelle génération d'images Lambda ARN"
            ]
        }
    ]
}

"Reconnaissance faciale Lambda arn" et "Nouvelle génération d'image Lambda arn" ne sont pas encore disponibles, alors n'oubliez pas de les réécrire après avoir créé la fonction Lambda. Tout ce traitement est exécuté dans ce rôle.

Créer la fonction Lambda du contrôleur

Puisque API Gateway est une "connexion", nous devons créer le bot LINE aux deux extrémités et la fonction Lambda du contrôleur avant de la créer, nous allons donc créer la fonction Lambda du contrôleur. Puisque python est utilisé pour la création de fonctions cette fois, sélectionnez python3.x (3.6 ~ 3.8) comme moteur d'exécution. Le rôle IAM à exécuter est celui que vous avez créé précédemment.

Après l'avoir créé, réglez d'abord la mémoire à 512 Mo et le délai d'expiration à 1 min dans "Paramètres de base". Définissez ensuite les variables d'environnement suivantes:

Clé valeur
LINE_CHANNEL_ACCESS_TOKEN Jeton d'accès LINE Bot Channel
LINE_CHANNEL_SECRET Secret du canal du bot LINE

En ce qui concerne le contenu de la fonction Lambda, le contrôleur Lambda communique avec le bot LINE, le package "line-bot-sdk" est donc requis. Pour installer sur Lambda, installez d'abord line-bot-sdk localement dans le nouveau dossier à l'aide de la commande suivante:

python -m pip install line-bot-sdk -t <new_folder>

Après cela, créez un fichier lambda_function.py (Lambda le reconnaît comme la fonction principale avec ce nom, alors assurez-vous de le nommer) dans le même dossier et entrez le code suivant:

lambda_function_for_controller.py


import os
import sys
import logging

import boto3
import json

from linebot import LineBotApi, WebhookHandler
from linebot.models import MessageEvent, TextMessage, TextSendMessage, ImageMessage, ImageSendMessage
from linebot.exceptions import LineBotApiError, InvalidSignatureError


logger = logging.getLogger()
logger.setLevel(logging.ERROR)

#Lire les jetons d'accès et les secrets du canal de bot de ligne à partir des variables d'environnement
channel_secret = os.getenv('LINE_CHANNEL_SECRET', None)
channel_access_token = os.getenv('LINE_CHANNEL_ACCESS_TOKEN', None)
if channel_secret is None:
    logger.error('Specify LINE_CHANNEL_SECRET as environment variable.')
    sys.exit(1)
if channel_access_token is None:
    logger.error('Specify LINE_CHANNEL_ACCESS_TOKEN as environment variable.')
    sys.exit(1)

# api&Générer un gestionnaire
line_bot_api = LineBotApi(channel_access_token)
handler = WebhookHandler(channel_secret)

#Connectez-vous avec le compartiment S3
s3 = boto3.client("s3")
bucket = "<Nom du compartiment S3>"

#Fonction principale de Lambda
def lambda_handler(event, context):

    #X pour authentification-Line-En-tête de signature
    signature = event["headers"]["X-Line-Signature"]

    body = event["body"]

    #Définition de la valeur de retour
    ok_json = {"isBase64Encoded": False,
               "statusCode": 200,
               "headers": {},
               "body": ""}
    error_json = {"isBase64Encoded": False,
                  "statusCode": 403,
                  "headers": {},
                  "body": "Error"}

    @handler.add(MessageEvent, message=ImageMessage)
    def message(line_event):

        #Profil de l'utilisateur
        profile = line_bot_api.get_profile(line_event.source.user_id)

        #Extraire l'ID de l'utilisateur qui a envoyé(push_Utiliser si message,Pas nécessaire pour la réponse)
        # user_id = profile.user_id

        #Extraire l'ID de message
        message_id = line_event.message.id

        #Extraire le fichier image
        message_content = line_bot_api.get_message_content(message_id)
        content = bytes()
        for chunk in message_content.iter_content():
            content += chunk

        #Enregistrer le fichier image
        key = "origin_photo/" + message_id
        new_key = message_id[-3:]
        s3.put_object(Bucket=bucket, Key=key, Body=content)

        #Reconnaissance faciale des appels lambda
        lambdaRekognitionName = "<C'est l'arn de la reconnaissance faciale lambda>"
        params = {"Bucket": bucket, "Key": key}  #Informations sur le chemin du fichier image
        payload = json.dumps(params)
        response = boto3.client("lambda").invoke(
            FunctionName=lambdaRekognitionName, InvocationType="RequestResponse", Payload=payload)
        response = json.load(response["Payload"])

        #Appeler la nouvelle génération d'images lambda
        lambdaNewMaskName = "<Voici la nouvelle génération d'image lambda arn>"
        params = {"landmarks": str(response),
                  "bucket": bucket,
                  "photo_key": key,
                  "new_photo_key": new_key}
        payload = json.dumps(params)
        boto3.client("lambda").invoke(FunctionName=lambdaNewMaskName,
                                      InvocationType="RequestResponse", Payload=payload)

        #Génération d'URL signée
        presigned_url = s3.generate_presigned_url(ClientMethod="get_object", Params={
                                                  "Bucket": bucket, "Key": new_key}, ExpiresIn=600)

        #Répondre au nouveau message image
        line_bot_api.reply_message(line_event.reply_token, ImageSendMessage(
            original_content_url=presigned_url, preview_image_url=presigned_url))

    try:
        handler.handle(body, signature)
    except LineBotApiError as e:
        logger.error("Got exception from LINE Messaging API: %s\n" % e.message)
        for m in e.error.details:
            logger.error("  %s: %s" % (m.property, m.message))
        return error_json
    except InvalidSignatureError:
        return error_json

    return ok_json

Ci-dessus se trouve toute la fonction Lambda du contrôleur, qui est associée aux cinq parties. La partie de cette première partie est:

lambda_function_for_controller.py


#Lire les jetons d'accès et les secrets du canal de bot de ligne à partir des variables d'environnement
channel_secret = os.getenv('LINE_CHANNEL_SECRET', None)
channel_access_token = os.getenv('LINE_CHANNEL_ACCESS_TOKEN', None)
if channel_secret is None:
    logger.error('Specify LINE_CHANNEL_SECRET as environment variable.')
    sys.exit(1)
if channel_access_token is None:
    logger.error('Specify LINE_CHANNEL_ACCESS_TOKEN as environment variable.')
    sys.exit(1)

# api&Générer un gestionnaire
line_bot_api = LineBotApi(channel_access_token)
handler = WebhookHandler(channel_secret)

lambda_function_for_controller.py


    #X pour authentification-Line-En-tête de signature
    signature = event["headers"]["X-Line-Signature"]

    body = event["body"]

Vous avez maintenant authentifié votre bot LINE et reçu les détails de l'événement. Après cela, compressez le contenu de ce dossier dans un zip et Téléchargez en sélectionnant «Code de fonction» -> «Action» -> «Télécharger un fichier .zip» de Lambda.

Créer une passerelle API

Le dernier est la création d'API Gateway en tant que connexion. Le type de passerelle API créé ici est l'API REST. Après avoir créé l'API, créez les ressources et les méthodes. La méthode est la méthode POST, le type d'intégration est la fonction Lambda et l'utilisation de l'intégration de proxy Lambda est également activée. La fonction Lambda sélectionne la fonction Lambda du contrôleur.

Aussi, à propos du paramétrage de la demande de méthode POST Tout d'abord, sélectionnez «Valider les paramètres et les en-têtes de la chaîne de requête» pour authentifier la demande. Et ajoutez l'en-tête suivant à l'en-tête de la requête HTTP:

Nom Obligatoire cache
X-Line-Signature

Une fois défini, déployons. Une fois le déploiement terminé, copiez l'URL d'appel de méthode sur scène et Collez-le dans l'URL du webhook LINE Bot. Ceci termine la première partie.

3-2 Partie de stockage d'image

Flux de processus

La deuxième partie est la partie stockage d'image. Cette partie est très simple, il suffit de sauvegarder l'image chargée par le contrôleur Lambda dans le compartiment S3. Le flux de traitement est le suivant:

Créer un compartiment S3

Tout d'abord, créez un compartiment S3 pour votre travail. Dans ce projet, si le nom du bucket est trop long, un "problème de longueur d'URL signée" se produira (voir [3-5](#signed url) pour plus de détails). Rendez le nom du bucket aussi court que possible (4 caractères anglais dans mon cas). De plus, vous ne voulez pas que les autres voient votre photo, non? Pour protéger votre vie privée Cochez "Bloquer tous les accès publics" dans les paramètres d'autorisation pour créer un bucket. Après création, un dossier nommé "origin_photo" pour enregistrer les photos téléchargées par l'utilisateur, Créez un dossier appelé «masques» pour enregistrer les images de masque. Cela termine le travail du côté S3.

Fonction Lambda du contrôleur

La fonction Lambda du contrôleur a été renseignée dans la première partie, il n'y a donc rien de spécial à faire ici. Expliquez simplement le code de cette partie et le contenu est:

lambda_function_for_controller.py


#Connectez-vous avec le compartiment S3
s3 = boto3.client("s3")
bucket = "<Nom du compartiment S3>"

lambda_function_for_controller.py


        #Extraire l'ID de message
        message_id = line_event.message.id

        #Extraire le fichier image
        message_content = line_bot_api.get_message_content(message_id)
        content = bytes()
        for chunk in message_content.iter_content():
            content += chunk

        #Enregistrer le fichier image
        key = "origin_photo/" + message_id
        new_key = message_id[-3:]
        s3.put_object(Bucket=bucket, Key=key, Body=content)

Ici, renommez le fichier image avec l'ID de message LINE, Plusieurs utilisateurs pourront distinguer.

3-3 Partie reconnaissance faciale

La troisième partie est la reconnaissance des photos enregistrées. Plus précisément, il reconnaît le contour du visage et les positions des yeux et du nez, et l'utilise pour une image et une combinaison ultérieures du masque. Avec le concept de "réduire autant que possible la quantité de travail" Je ne veux pas entraîner moi-même l'IA de reconnaissance faciale à partir de zéro Les visages sont reconnus à l'aide d'un service appelé «Rekognition» sur AWS.

Qu'est-ce que la reconnaissance?

Rekognition est un service qui «utilise l'apprentissage automatique pour automatiser l'analyse des images et des vidéos». En termes simples, cela ressemble à "utiliser l'IA entraînée telle quelle". Voici une introduction à Rekognition: Amazon Rekognition

Rekognition a diverses fonctions telles que la détection d'objets et de scènes et la comparaison de visages, et peut traiter non seulement des images mais aussi des vidéos. Cette fois, nous utiliserons la fonction "détection de visage" pour obtenir la position du visage. Les informations de localisation que vous souhaitez obtenir sont appelées «point de repère». La figure ci-dessous est une image d'un point de repère:

Résultat d'analyse de cette figure:

<détails>

Résultat de reconnaissance de reconnaissance </ summary>

{
    "FaceDetails": [
        {
            "AgeRange": {
                "High": 43,
                "Low": 26
            },
            "Beard": {
                "Confidence": 97.48941802978516,
                "Value": true
            },
            "BoundingBox": {
                "Height": 0.6968063116073608,
                "Left": 0.26937249302864075,
                "Top": 0.11424895375967026,
                "Width": 0.42325547337532043
            },
            "Confidence": 99.99995422363281,
            "Emotions": [
                {
                    "Confidence": 0.042965151369571686,
                    "Type": "DISGUSTED"
                },
                {
                    "Confidence": 0.002022328320890665,
                    "Type": "HAPPY"
                },
                {
                    "Confidence": 0.4482877850532532,
                    "Type": "SURPRISED"
                },
                {
                    "Confidence": 0.007082826923578978,
                    "Type": "ANGRY"
                },
                {
                    "Confidence": 0,
                    "Type": "CONFUSED"
                },
                {
                    "Confidence": 99.47616577148438,
                    "Type": "CALM"
                },
                {
                    "Confidence": 0.017732391133904457,
                    "Type": "SAD"
                }
            ],
            "Eyeglasses": {
                "Confidence": 99.42405700683594,
                "Value": false
            },
            "EyesOpen": {
                "Confidence": 99.99604797363281,
                "Value": true
            },
            "Gender": {
                "Confidence": 99.722412109375,
                "Value": "Male"
            },
            "Landmarks": [
                {
                    "Type": "eyeLeft",
                    "X": 0.38549351692199707,
                    "Y": 0.3959200084209442
                },
                {
                    "Type": "eyeRight",
                    "X": 0.5773905515670776,
                    "Y": 0.394561767578125
                },
                {
                    "Type": "mouthLeft",
                    "X": 0.40410104393959045,
                    "Y": 0.6479480862617493
                },
                {
                    "Type": "mouthRight",
                    "X": 0.5623446702957153,
                    "Y": 0.647117555141449
                },
                {
                    "Type": "nose",
                    "X": 0.47763553261756897,
                    "Y": 0.5337067246437073
                },
                {
                    "Type": "leftEyeBrowLeft",
                    "X": 0.3114689588546753,
                    "Y": 0.3376390337944031
                },
                {
                    "Type": "leftEyeBrowRight",
                    "X": 0.4224424660205841,
                    "Y": 0.3232649564743042
                },
                {
                    "Type": "leftEyeBrowUp",
                    "X": 0.36654090881347656,
                    "Y": 0.3104579746723175
                },
                {
                    "Type": "rightEyeBrowLeft",
                    "X": 0.5353175401687622,
                    "Y": 0.3223199248313904
                },
                {
                    "Type": "rightEyeBrowRight",
                    "X": 0.6546239852905273,
                    "Y": 0.3348073363304138
                },
                {
                    "Type": "rightEyeBrowUp",
                    "X": 0.5936762094497681,
                    "Y": 0.3080498278141022
                },
                {
                    "Type": "leftEyeLeft",
                    "X": 0.3524211347103119,
                    "Y": 0.3936865031719208
                },
                {
                    "Type": "leftEyeRight",
                    "X": 0.4229775369167328,
                    "Y": 0.3973258435726166
                },
                {
                    "Type": "leftEyeUp",
                    "X": 0.38467878103256226,
                    "Y": 0.3836822807788849
                },
                {
                    "Type": "leftEyeDown",
                    "X": 0.38629674911499023,
                    "Y": 0.40618783235549927
                },
                {
                    "Type": "rightEyeLeft",
                    "X": 0.5374732613563538,
                    "Y": 0.39637991786003113
                },
                {
                    "Type": "rightEyeRight",
                    "X": 0.609208345413208,
                    "Y": 0.391626238822937
                },
                {
                    "Type": "rightEyeUp",
                    "X": 0.5750962495803833,
                    "Y": 0.3821527063846588
                },
                {
                    "Type": "rightEyeDown",
                    "X": 0.5740782618522644,
                    "Y": 0.40471214056015015
                },
                {
                    "Type": "noseLeft",
                    "X": 0.4441811740398407,
                    "Y": 0.5608476400375366
                },
                {
                    "Type": "noseRight",
                    "X": 0.5155643820762634,
                    "Y": 0.5569332242012024
                },
                {
                    "Type": "mouthUp",
                    "X": 0.47968366742134094,
                    "Y": 0.6176465749740601
                },
                {
                    "Type": "mouthDown",
                    "X": 0.4807897210121155,
                    "Y": 0.690782368183136
                },
                {
                    "Type": "leftPupil",
                    "X": 0.38549351692199707,
                    "Y": 0.3959200084209442
                },
                {
                    "Type": "rightPupil",
                    "X": 0.5773905515670776,
                    "Y": 0.394561767578125
                },
                {
                    "Type": "upperJawlineLeft",
                    "X": 0.27245330810546875,
                    "Y": 0.3902156949043274
                },
                {
                    "Type": "midJawlineLeft",
                    "X": 0.31561678647994995,
                    "Y": 0.6596118807792664
                },
                {
                    "Type": "chinBottom",
                    "X": 0.48385748267173767,
                    "Y": 0.8160444498062134
                },
                {
                    "Type": "midJawlineRight",
                    "X": 0.6625112891197205,
                    "Y": 0.656606137752533
                },
                {
                    "Type": "upperJawlineRight",
                    "X": 0.7042999863624573,
                    "Y": 0.3863988518714905
                }
            ],
            "MouthOpen": {
                "Confidence": 99.83820343017578,
                "Value": false
            },
            "Mustache": {
                "Confidence": 72.20288848876953,
                "Value": false
            },
            "Pose": {
                "Pitch": -4.970901966094971,
                "Roll": -1.4911699295043945,
                "Yaw": -10.983647346496582
            },
            "Quality": {
                "Brightness": 73.81391906738281,
                "Sharpness": 86.86019134521484
            },
            "Smile": {
                "Confidence": 99.93638610839844,
                "Value": false
            },
            "Sunglasses": {
                "Confidence": 99.81478881835938,
                "Value": false
            }
        }
    ]
}

Ce que je veux obtenir cette fois, c'est l'élément "points de repère". "Type" est le nom du point (voir l'image ci-dessus). Cependant, x et y ne sont pas les coordonnées de points de pixels spécifiques. Affiche le rapport à la largeur de l'image.

Flux de processus

Le flux de traitement de la troisième partie est le suivant: La reconnaissance a deux mécanismes pour lire les images. La première consiste à charger à l'aide du compartiment S3 ou de l'URL de l'image sur Internet. La seconde consiste à envoyer le fichier et à le lire directement. Cette fois, nous utiliserons la première méthode URL. Par conséquent, ce n'est pas l'image qui est transmise du contrôleur Lambda au Lambda de reconnaissance faciale, mais les informations d'emplacement de stockage du fichier. Il en va de même pour la reconnaissance faciale Lambda pour passer à Rekognition.

Le rôle IAM qui exécute la reconnaissance faciale Lambda ici est le rôle créé dans la première partie. J'ai l'autorité d'utiliser S3 et Rekognition, donc Même si le compartiment S3 est privé, Rekognition peut lire les images qu'il contient et il n'y a aucun problème.

Et le résultat renvoyé par Rekognition semble être un exemple du résultat ci-dessus. Il contient plusieurs éléments tels que "l'âge" et le "sexe", Je ne veux utiliser que des "points de repère" cette fois. Par conséquent, la reconnaissance faciale Lambda extrait les points de repère des résultats.

En outre, il existe de nombreux points de repère, Il y a certains points (bouche, etc.) qui ne peuvent pas être bien reconnus à cause du masque, et il y a quelques points supplémentaires (yeux, etc.) qui sont trop fins. Par conséquent, il extrait simplement les 5 points de repère suivants et les renvoie au contrôleur Lambda.

Nom du repère position
eyeLeft oeil gauche
eyeRight l'œil droit
upperJawlineLeft Komekami gauche
upperJawlineRight Riz droit
chinBottom Menton
  • La traduction des points de repère peut être un peu étrange, veuillez donc vous référer à la figure.

Créer une fonction Lambda de reconnaissance faciale

Afin de séparer les rôles, créez une autre fonction Lambda de reconnaissance faciale en plus de la fonction Lambda du contrôleur. Lors de la création, tout comme la fonction Lambda du contrôleur, Sélectionnez python3.x et le rôle d'exécution est le même. Réglez également le délai d'expiration de 1 min et la mémoire de 512 Mo de la même manière dans "Paramètres de base".

Après l'avoir créé, il n'y a pas de package à introduire ici, donc Pas besoin de télécharger un zip, Tout ce que vous avez à faire est de renseigner le code Lambda_function.py généré automatiquement ci-dessous.

  • N'oubliez pas d'ajouter le lien arn Lambda de reconnaissance faciale à la stratégie d'appel du rôle IAM.

lambda_function_for_rekognition.py


import json
import boto3

rekognition = boto3.client("rekognition")


def lambda_handler(event, context):

    #Récupère le chemin du fichier image à partir de l'événement
    bucket = event["Bucket"]
    key = event["Key"]

    #Appeler Rekognition pour effectuer la reconnaissance faciale
    response = rekognition.detect_faces(
        Image={'S3Object': {'Bucket': bucket, 'Name': key}}, Attributes=['ALL'])

    #Combien de personnes sont sur la photo
    number_of_people = len(response["FaceDetails"])

    #Faites une liste de tous les points de repère requis
    all_needed_landmarks = []
    #Processus par le nombre de personnes
    for i in range(number_of_people):
        #Ceci est une liste de dictionnaires
        all_landmarks_of_one_person = response["FaceDetails"][i]["Landmarks"]
        #Cette fois, eyeLeft, eyeRight, upperJawlineLeft, upperJawlineRight,Utiliser uniquement chinBottom
        # needed_Extrait aux points de repère
        needed_landmarks = []
        for type in ["eyeLeft", "eyeRight", "upperJawlineLeft", "upperJawlineRight", "chinBottom"]:
            landmark = next(
                item for item in all_landmarks_of_one_person if item["Type"] == type)
            needed_landmarks.append(landmark)
        all_needed_landmarks.append(needed_landmarks)

    return all_needed_landmarks

Fonction Lambda du contrôleur

La fonction Lambda du contrôleur étant déjà remplie, Ceci est juste une description du code de la partie 3.

  • Reconnaissance faciale des appels Lambda réponse est les 5 repères que vous avez acquis.

lambda_function_for_controller.py


        lambdaRekognitionName = "<C'est l'arn de la reconnaissance faciale lambda>"
        params = {"Bucket": bucket, "Key": key}  #Informations sur le chemin du fichier image
        payload = json.dumps(params)
        response = boto3.client("lambda").invoke(
            FunctionName=lambdaRekognitionName, InvocationType="RequestResponse", Payload=payload)
        response = json.load(response["Payload"])

3-4 Nouvelle partie de génération d'image

Flux de processus

La quatrième partie est la nouvelle partie de génération d'images. En d'autres termes, c'est la partie qui combine l'image photographique et la nouvelle image de masque suivante:

Nom Bane Joker Immortan Joe
Image de masque
※1

※2

※3
La source La nuit noire se lève chevaliernoir MadMaxAngryDeathRoad

※1:This work is a derivative of "Bane"byistolethetv,usedunderCCBY2.0. ※2:This work is a derivative of this photo,usedunderCC01.0. ※3:This work, "joe's mask" is a derivative of "File:Fan_Expo_2015_-Immortan_Joe(21147179383).jpg"byGabboT,usedunderCCBY-SA2.0."joe'smask"islicensedCCBY-SA2.0 by y2-peng.

Le flux de traitement dans AWS est le suivant:

  • Tout d'abord, le contrôleur Lambda transmet «Informations de stockage d'image photo (nom du compartiment S3 et chemin du fichier)», «5 informations de repère» et «Nouveau nom de fichier image» à la nouvelle génération d'images Lambda.

  • Ensuite, la nouvelle génération d'image Lambda charge l'image photo et l'image de masque à partir du compartiment S3 à l'aide des informations d'enregistrement de fichier. De plus, il est nécessaire d'enregistrer l'image de masque dans le compartiment S3 à l'avance et d'enregistrer le chemin du fichier dans la nouvelle génération d'image Lambda. (Reportez-vous au code pour les paramètres détaillés tels que le chemin du fichier)

  • Ensuite, combinez l'image photographique et l'image du masque autant de fois que le nombre de personnes. Sélectionnez au hasard et utilisez une image de masque à chaque fois. L'ordre des travaux d'assemblage est le suivant:

※This work is a derivative of "[Bane](https://www.flickr.com/photos/istolethetv/30216006787/)"by[istolethetv](https://www.flickr.com/people/istolethetv/),usedunder[CCBY2.0](https://creativecommons.org/licenses/by/2.0/).
  • Enfin, nommez la nouvelle image avec le "nouveau nom de fichier image" et enregistrez-la dans le compartiment S3.

C'est tout pour le traitement.

Créer une nouvelle génération d'images Lambda

Tout d'abord, créez une nouvelle fonction Lambda dans AWS Lambda. Les rôles d'exécution et d'exécution sont les mêmes qu'auparavant. De même, comme auparavant, définissez la mémoire et le délai d'expiration à partir des "Paramètres de base".

Cette fois, la combinaison d'images nécessite deux packages python, "oreiller" et "numpy". Par conséquent, créez d'abord un nouveau dossier et installez le package à l'aide de la commande suivante.

python -m pip install pillow numpy -t <new_folder>

Ensuite, créez "lambda_function.py" dans ce dossier et entrez le code suivant.

lambda_function_for_new_image_gengeration.py


import json
import boto3

import numpy as np

from PIL import Image, ImageFile
from operator import sub
from io import BytesIO
from random import choice

s3 = boto3.client("s3")


class NewPhotoMaker:
    def __init__(self, all_landmarks, bucket, photo_key, new_photo_key):
        self.all_landmarks = eval(all_landmarks)
        self.bucket = bucket
        self.photo_key = photo_key
        self.new_photo_key = new_photo_key

    #Charger l'image photographique
    def load_photo_image(self):
        s3.download_file(self.bucket, self.photo_key, "/tmp/photo_file")
        self.photo_image = Image.open("/tmp/photo_file")

    #Chargez l'image du masque
    def load_mask_image(self):
        #fléau (Batman),joker (Batman),Sélection aléatoire de Immortan Joe (Mad Max)
        mask_key = "masks/" + choice(["bane", "joker", "joe"]) + ".png "
        s3.download_file(self.bucket, mask_key, "/tmp/mask_file")
        self.mask_image = Image.open("/tmp/mask_file")

    #Passer d'un repère (ratio) à un point spécifique
    def landmarks_to_points(self):
        upperJawlineLeft_landmark = next(
            item for item in self.landmarks if item["Type"] == "upperJawlineLeft")
        upperJawlineRight_landmark = next(
            item for item in self.landmarks if item["Type"] == "upperJawlineRight")
        eyeLeft_landmark = next(
            item for item in self.landmarks if item["Type"] == "eyeLeft")
        eyeRight_landmark = next(
            item for item in self.landmarks if item["Type"] == "eyeRight")

        self.upperJawlineLeft_point = [int(self.photo_image.size[0] * upperJawlineLeft_landmark["X"]), 
                                       int(self.photo_image.size[1] * upperJawlineLeft_landmark["Y"])]
        self.upperJawlineRight_point = [int(self.photo_image.size[0] * upperJawlineRight_landmark["X"]), 
                                        int(self.photo_image.size[1] * upperJawlineRight_landmark["Y"])]
        self.eyeLeft_point = [int(self.photo_image.size[0] * eyeLeft_landmark["X"]),
                              int(self.photo_image.size[1] * eyeLeft_landmark["Y"])]
        self.eyeRight_point = [int(self.photo_image.size[0] * eyeRight_landmark["X"]),
                               int(self.photo_image.size[1] * eyeRight_landmark["Y"])]

    #Redimensionner l'image du masque pour l'adapter à la largeur de votre visage
    def resize_mask(self):
        face_width = int(np.linalg.norm(list(map(sub, self.upperJawlineLeft_point, self.upperJawlineRight_point))))
        new_hight = int(self.mask_image.size[1]*face_width/self.mask_image.size[0])
        self.mask_image = self.mask_image.resize((face_width, new_hight))

    #Faites pivoter l'image du masque en fonction de l'angle du visage (pas de la diagonale du visage due à la rotation du cou)
    def rotate_mask(self):
        angle = np.arctan2(self.upperJawlineRight_point[1] - self.upperJawlineLeft_point[1],
                           self.upperJawlineRight_point[0] - self.upperJawlineLeft_point[0])
        angle = -np.degrees(angle)  # radian to dgree
        self.mask_image = self.mask_image.rotate(angle, expand=True)

    #Combinez une image photographique et une image de masque
    def match_mask_position(self):
        #Correspondance en utilisant la position des yeux
        face_center = [int((self.eyeLeft_point[0] + self.eyeRight_point[0])/2),
                       int((self.eyeLeft_point[1] + self.eyeRight_point[1])/2)]
        mask_center = [int(self.mask_image.size[0]/2),
                       int(self.mask_image.size[1]/2)]
        x = face_center[0] - mask_center[0]
        y = face_center[1] - mask_center[1]
        self.photo_image.paste(self.mask_image, (x, y), self.mask_image)

    #Enregistrer le nouveau fichier image sur S3
    def save_new_photo(self):
        new_photo_byte_arr = BytesIO()
        self.photo_image.save(new_photo_byte_arr, format="JPEG")
        new_photo_byte_arr = new_photo_byte_arr.getvalue()
        s3.put_object(Bucket=self.bucket, Key=self.new_photo_key,
                      Body=new_photo_byte_arr)

    #Courir
    def run(self):

        self.load_photo_image()

        #Traitement pour le nombre de personnes
        for i in range(len(self.all_landmarks)):
            self.load_mask_image()  #Chargez un nouveau masque à chaque fois
            self.landmarks = self.all_landmarks[i]
            self.landmarks_to_points()
            self.resize_mask()
            self.rotate_mask()
            self.match_mask_position()

        self.save_new_photo()

#fonction principale lambda


def lambda_handler(event, context):
    landmarks = event["landmarks"]
    bucket = event["bucket"]
    photo_key = event["photo_key"]
    new_photo_key = event["new_photo_key"]

    photo_maker = NewPhotoMaker(landmarks, bucket, photo_key, new_photo_key)
    photo_maker.run()

Enfin, compressez tout le contenu du dossier et téléchargez-le sur Lambda. Ceci termine la création de la nouvelle génération d'images.

  • N'oubliez pas d'ajouter le lien arn Lambda de nouvelle génération d'image à la stratégie d'appel du rôle IAM.

Fonction Lambda du contrôleur

Le code du contrôleur Lambda pour cette partie est:

lambda_function_for_controller.py


        #Appeler la nouvelle génération d'images lambda
        lambdaNewMaskName = "<Voici la nouvelle génération d'image lambda arn>"
        params = {"landmarks": str(response),
                  "bucket": bucket,
                  "photo_key": key,
                  "new_photo_key": new_key}
        payload = json.dumps(params)
        boto3.client("lambda").invoke(FunctionName=lambdaNewMaskName,
                                      InvocationType="RequestResponse", Payload=payload)

3-5 Nouvelle partie de sortie d'image

Sortie d'image sur LINE Bot

La dernière partie est la partie sortie de la nouvelle image. Cette application utilise LINE Bot pour entrer et sortir des images, et lors de la saisie, elle transmet directement le fichier image, La sortie ne peut pas envoyer le fichier image directement.

Document de message d'image dans l'API de message du bot LINE est une méthode de transmission d'image à l'utilisateur. Est stipulé. Ce n'est pas le fichier image que l'API reçoit, mais l'URL de l'image. Selon la documentation, La communication entre l'utilisateur et LINE Bot se fait via la plateforme LINE. Donc, ce processus de transmission est

  1. "Envoyer l'URL d'image du Bot LINE vers la plateforme LINE"
  2. "La plate-forme LINE charge les images stockées dans le compartiment S3"
  3. "La plate-forme LINE envoie des images aux utilisateurs"

Il est devenu. Mais ce processus fait des ** droits d'accès au compartiment S3 un problème **. Si le droit d'accès est défini sur "privé", la plateforme LINE ne pourra pas lire l'image et l'image donnée par l'utilisateur ressemblera à ceci: Si le droit d'accès est défini sur "public", n'importe qui peut y accéder en connaissant l'URL de l'objet S3 de l'image. Cela signifie que vos photos peuvent être vues par d'autres personnes, ce qui est un problème de confidentialité.

Pour le moment, j'ai pensé à utiliser DynamoDB etc. pour authentifier les utilisateurs LINE, La quantité de travail a considérablement augmenté et elle se heurte au concept de «réduire autant que possible la quantité de travail». Pour être honnête, je ne veux pas le faire.

Après de nombreuses recherches, j'ai finalement trouvé un bon moyen. C'est une "URL signée".

URL signée

Pour protéger votre vie privée, rendez l'accès au compartiment S3 "privé". Même si je connais l'URL de l'objet S3 de l'image, je ne peux pas y accéder. Mais si vous utilisez l 'URL signée émise avec l'autorité du rôle IAM, elle est privée. L'accès à des objets spécifiques dans le compartiment S3 est possible. Cela ressemble à une URL de conférence avec un mot de passe de zoom.

Vous pouvez également définir une date d'expiration pour cette URL signée. Lorsqu'elle expire, l'URL devient inutilisable, ce qui la rend encore plus sûre: Mais une chose à noter est la longueur de l'URL signée. L'URL signée émise avec l'autorité du rôle IAM contient des informations de jeton pour un accès temporaire, de sorte que l'URL sera assez longue. Cependant, selon les règles de l'API de message d'image de LINE Bot, la longueur maximale de l'URL pouvant être reçue est de 1000 caractères. Par conséquent, si le nom du compartiment S3, le chemin du fichier image et le nom du fichier image sont trop longs, l'URL dépassera 1 000 caractères et ne pourra pas être envoyée. Ainsi, lorsque j'ai créé le compartiment S3 pour la deuxième partie, j'ai parfois dit: «Le nom du compartiment doit être aussi court que possible». Pour la même raison, le nouveau nom de fichier image doit être composé des 3 derniers caractères de l'ID de message (raccourcissez le nom du fichier). J'enregistre également le nouveau fichier image dans le dossier roll du seau S3 (raccourcissez le chemin du fichier). Cela a résolu le problème de longueur d'URL signée.

Supplément: Il existe en fait une autre solution au problème de longueur d'URL signée. Il s'agit de publier l'URL avec les privilèges de l'utilisateur IAM, et non le rôle IAM. Les URL émises par les utilisateurs IAM ne nécessitent pas de jetons et peuvent être raccourcies, Vous devez utiliser l '«ID de clé d'accès» et la «clé d'accès secrète» de l'utilisateur IAM. Pour des raisons de sécurité, nous vous déconseillons d'émettre des URL en tant qu'utilisateurs IAM.

Flux de processus

Maintenant que nous avons résolu le problème des autorisations du compartiment S3, implémentons cette partie. Le déroulement de cette partie est le suivant:

Tout d'abord, la fonction Lambda du contrôleur transmet l'URL signée de la nouvelle image à LINE Bot. Ensuite, LINE Bot lit le fichier image à partir du compartiment S3 (la lecture réelle se fait sur la plateforme LINE), Envoyer au dernier utilisateur. C'est la fin du processus.

Fonction Lambda du contrôleur

Semblable à la partie ci-dessus, je vais expliquer le code de fonction Lambda du contrôleur pour cette partie.

  • Générer une URL signée La période de validité est fixée à 600 secondes.

lambda_function_for_controller.py


        #Génération d'URL signée
        presigned_url = s3.generate_presigned_url(ClientMethod="get_object", Params={
                                                  "Bucket": bucket, "Key": new_key}, ExpiresIn=600)
  • Envoyer une nouvelle image

lambda_function_for_controller.py


        #Répondre au nouveau message image
        line_bot_api.reply_message(line_event.reply_token, ImageSendMessage(
            original_content_url=presigned_url, preview_image_url=presigned_url))

4. Résultat actuel

Essayons l'application que nous avons créée!

interface

Le premier est l'envoi et la réception via l'interface LINE. Il existe un code QR pour Bot dans "Paramètres de l'API de messagerie" de LINE Bot, et vous pouvez l'utiliser pour l'ajouter à vos amis. Je vous l'enverrai plus tard. .. .. ※This work, "wearing joe's mask" is a derivative of "File:Fan_Expo_2015_-Immortan_Joe(21147179383).jpg"byGabboT,usedunderCCBY-SA2.0."wearingjoe'smask"islicensedCCBY-SA2.0 by y2-peng.

Vous l'avez fait! Voyons maintenant quels modèles fonctionnent et ce qui ne fonctionne pas!

Modèle réussi

description before after
1 personne avant IMG_1593.jpg IMG_1603.JPG※1
1 personne avant (avec rotation) IMG_1597.jpg IMG_1604.JPG※2
Devant plusieurs personnes 1.jpg 2.jpg※3
Même si le visage est trop grand IMG_1606.jpg IMG_1608.JPG※4

※1:This work is a derivative of this photo,usedunderCC01.0. ※2:This work, "result 2" is a derivative of "File:Fan_Expo_2015_-Immortan_Joe(21147179383).jpg"byGabboT,usedunderCCBY-SA2.0."result2"islicensedCCBY-SA2.0 by y2-peng. ※3:This work, "masked 4" is a derivative of "File:Fan_Expo_2015_-Immortan_Joe(21147179383).jpg" by GabboT, used under CC BY-SA 2.0, "Bane"byistolethetv,usedunderCCBY2.0, and this photo,usedunderCC01.0. "masked 4" is licensed CC BY-SA 2.0 by y2-peng. ※4:This work is a derivative of "Bane"byistolethetv,usedunderCCBY2.0.

Des modèles qui ne fonctionnent pas

description before after
Face diagonale 3.jpg 4.jpg※1
Visage trop petit (la personne à l'arrière) 5.jpg 6.jpg※2
Flou (la personne derrière) 7.jpg 8.jpg※3

※1:This work, "standing 2" is a derivative of "File:Fan_Expo_2015_-Immortan_Joe(21147179383).jpg" by GabboT, used under CC BY-SA 2.0 and "Bane"byistolethetv,usedunderCCBY2.0. "standing 2" is licensed CC BY-SA 2.0 by y2-peng. ※2:This work, "standing 4" is a derivative of "File:Fan_Expo_2015_-Immortan_Joe(21147179383).jpg" by GabboT, used under CC BY-SA 2.0 and "Bane"byistolethetv,usedunderCCBY2.0. "standing 4" is licensed CC BY-SA 2.0 by y2-peng. ※3:This work is a derivative of "Bane"byistolethetv,usedunderCCBY2.0.

une analyse

En fonction du résultat, s'il est clair et clair, le traitement peut être effectué grossièrement. S'il y a un flou, le visage ne peut pas être reconnu et le traitement ne sera pas effectué. Si le visage ou le visage incliné est trop petit, il sera traité, mais le résultat n'est pas correct.

5. Résumé et impressions

Résumé

Cette fois, nous avons développé une application LINE qui change le masque blanc de la photo en masque de monstre. En utilisant les services AWS, il était possible de le réaliser sans serveur, et nous avons pu mettre en œuvre de manière approfondie le concept de «réduire autant que possible la quantité de travail». Si la photo est claire à l'avant, le processus de conversion est généralement correct. Cependant, le traitement des faces diagonales et des visages flous sera un problème pour l'avenir.

Tâches futures

  1. Face diagonale: Actuellement, le traitement des faces diagonales est incorrect. La raison en est que le masque est uniquement destiné à l'avant, pas à la face diagonale. Comme solution future, je pense à faire pivoter le masque 2D vers le système de coordonnées 3D, puis à le combiner, ou à préparer une image de masque pour les faces diagonales.
  2. Visage trop petit ou flou: La reconnaissance faciale actuelle utilise AWS Rekognition et ses performances définissent la limite supérieure des performances de cette application. Si je pouvais développer moi-même un système de reconnaissance faciale plus précis, je pense que je pourrais résoudre ce problème. (Mais c'est un conflit avec "réduire la quantité de travail autant que possible" :()
  3. Sélection du masque: Actuellement, j'utilise le masque fantôme au hasard parmi les trois, mais j'aimerais l'augmenter à l'avenir. En outre, je veux permettre aux utilisateurs de sélectionner, pas seulement une sélection aléatoire. Marquer le masque pour qu'il puisse répondre à toutes les demandes de l'utilisateur telles que "Je veux porter un masque XX" et "Je veux un masque mignon".

Autres impressions

  1. Commodité du serverless: Ce que j'ai ressenti le plus cette fois, c'est le charme du serverless. Si vous avez un serveur, non seulement la construction de l'environnement, mais également la gestion de la maintenance sont nécessaires, ce qui prend un temps considérable. Mais le développement sans serveur a pu les ignorer et gagner du temps. Il peut être utilisé pour le développement agile. Cependant, le traitement sans serveur dans Lambd a des limitations de performances, donc si vous avez un traitement compliqué, démarrons le serveur.
  2. AWS gratuit pendant un an au mieux! !! : Les nouveaux comptes AWS ont un "niveau gratuit" d'un an et toute utilisation dans une certaine plage est gratuite. Les Lambda, API Gateway, S3, Rekognition et Cloudwatch utilisés pour ce développement ont tous été conçus pour 0 yen, ce qui était une bonne affaire. J'aimerais essayer diverses choses pendant la période gratuite des quelques mois restants. Si vous êtes intéressé, faites-le! C'est gratuit!

6. Tous les codes

lambda_function_for_controller.py

lambda_function_for_controller.py


import os
import sys
import logging

import boto3
import json

from linebot import LineBotApi, WebhookHandler
from linebot.models import MessageEvent, TextMessage, TextSendMessage, ImageMessage, ImageSendMessage
from linebot.exceptions import LineBotApiError, InvalidSignatureError


logger = logging.getLogger()
logger.setLevel(logging.ERROR)

#Lire les jetons d'accès et les secrets du canal de bot de ligne à partir des variables d'environnement
channel_secret = os.getenv('LINE_CHANNEL_SECRET', None)
channel_access_token = os.getenv('LINE_CHANNEL_ACCESS_TOKEN', None)
if channel_secret is None:
    logger.error('Specify LINE_CHANNEL_SECRET as environment variable.')
    sys.exit(1)
if channel_access_token is None:
    logger.error('Specify LINE_CHANNEL_ACCESS_TOKEN as environment variable.')
    sys.exit(1)

# api&Générer un gestionnaire
line_bot_api = LineBotApi(channel_access_token)
handler = WebhookHandler(channel_secret)

#Connectez-vous avec le compartiment S3
s3 = boto3.client("s3")
bucket = "<Nom du compartiment S3>"

#Fonction principale de Lambda
def lambda_handler(event, context):

    #X pour authentification-Line-En-tête de signature
    signature = event["headers"]["X-Line-Signature"]

    body = event["body"]

    #Définition de la valeur de retour
    ok_json = {"isBase64Encoded": False,
               "statusCode": 200,
               "headers": {},
               "body": ""}
    error_json = {"isBase64Encoded": False,
                  "statusCode": 403,
                  "headers": {},
                  "body": "Error"}

    @handler.add(MessageEvent, message=ImageMessage)
    def message(line_event):

        #Profil de l'utilisateur
        profile = line_bot_api.get_profile(line_event.source.user_id)

        #Extraire l'ID de l'utilisateur qui a envoyé(push_Utiliser si message,Pas nécessaire pour la réponse)
        # user_id = profile.user_id

        #Extraire l'ID de message
        message_id = line_event.message.id

        #Extraire le fichier image
        message_content = line_bot_api.get_message_content(message_id)
        content = bytes()
        for chunk in message_content.iter_content():
            content += chunk

        #Enregistrer le fichier image
        key = "origin_photo/" + message_id
        new_key = message_id[-3:]
        s3.put_object(Bucket=bucket, Key=key, Body=content)

        #Reconnaissance faciale des appels lambda
        lambdaRekognitionName = "<C'est l'arn de la reconnaissance faciale lambda>"
        params = {"Bucket": bucket, "Key": key}  #Informations sur le chemin du fichier image
        payload = json.dumps(params)
        response = boto3.client("lambda").invoke(
            FunctionName=lambdaRekognitionName, InvocationType="RequestResponse", Payload=payload)
        response = json.load(response["Payload"])

        #Appeler la nouvelle génération d'images lambda
        lambdaNewMaskName = "<Voici la nouvelle génération d'image lambda arn>"
        params = {"landmarks": str(response),
                  "bucket": bucket,
                  "photo_key": key,
                  "new_photo_key": new_key}
        payload = json.dumps(params)
        boto3.client("lambda").invoke(FunctionName=lambdaNewMaskName,
                                      InvocationType="RequestResponse", Payload=payload)

        #Génération d'URL signée
        presigned_url = s3.generate_presigned_url(ClientMethod="get_object", Params={
                                                  "Bucket": bucket, "Key": new_key}, ExpiresIn=600)

        #Répondre au nouveau message image
        line_bot_api.reply_message(line_event.reply_token, ImageSendMessage(
            original_content_url=presigned_url, preview_image_url=presigned_url))

    try:
        handler.handle(body, signature)
    except LineBotApiError as e:
        logger.error("Got exception from LINE Messaging API: %s\n" % e.message)
        for m in e.error.details:
            logger.error("  %s: %s" % (m.property, m.message))
        return error_json
    except InvalidSignatureError:
        return error_json

    return ok_json

lambda_function_for_rekognition.py

lambda_function_for_rekognition.py


import json
import boto3

rekognition = boto3.client("rekognition")


def lambda_handler(event, context):

    #Récupère le chemin du fichier image à partir de l'événement
    bucket = event["Bucket"]
    key = event["Key"]

    #Appeler Rekognition pour effectuer la reconnaissance faciale
    response = rekognition.detect_faces(
        Image={'S3Object': {'Bucket': bucket, 'Name': key}}, Attributes=['ALL'])

    #Combien de personnes sont sur la photo
    number_of_people = len(response["FaceDetails"])

    #Faites une liste de tous les points de repère requis
    all_needed_landmarks = []
    #Processus par le nombre de personnes
    for i in range(number_of_people):
        #Ceci est une liste de dictionnaires
        all_landmarks_of_one_person = response["FaceDetails"][i]["Landmarks"]
        #Cette fois, eyeLeft, eyeRight, upperJawlineLeft, upperJawlineRight,Utiliser uniquement chinBottom
        # needed_Extrait aux points de repère
        needed_landmarks = []
        for type in ["eyeLeft", "eyeRight", "upperJawlineLeft", "upperJawlineRight", "chinBottom"]:
            landmark = next(
                item for item in all_landmarks_of_one_person if item["Type"] == type)
            needed_landmarks.append(landmark)
        all_needed_landmarks.append(needed_landmarks)

    return all_needed_landmarks

lambda_function_for_new_image_gengeration.py

lambda_function_for_new_image_gengeration.py


import json
import boto3

import numpy as np

from PIL import Image, ImageFile
from operator import sub
from io import BytesIO
from random import choice

s3 = boto3.client("s3")


class NewPhotoMaker:
    def __init__(self, all_landmarks, bucket, photo_key, new_photo_key):
        self.all_landmarks = eval(all_landmarks)
        self.bucket = bucket
        self.photo_key = photo_key
        self.new_photo_key = new_photo_key

    #Charger l'image photographique
    def load_photo_image(self):
        s3.download_file(self.bucket, self.photo_key, "/tmp/photo_file")
        self.photo_image = Image.open("/tmp/photo_file")

    #Chargez l'image du masque
    def load_mask_image(self):
        #fléau (Batman),joker (Batman),Sélection aléatoire de Immortan Joe (Mad Max)
        mask_key = "masks/" + choice(["bane", "joker", "joe"]) + ".png "
        s3.download_file(self.bucket, mask_key, "/tmp/mask_file")
        self.mask_image = Image.open("/tmp/mask_file")

    #Passer d'un repère (ratio) à un point spécifique
    def landmarks_to_points(self):
        upperJawlineLeft_landmark = next(
            item for item in self.landmarks if item["Type"] == "upperJawlineLeft")
        upperJawlineRight_landmark = next(
            item for item in self.landmarks if item["Type"] == "upperJawlineRight")
        eyeLeft_landmark = next(
            item for item in self.landmarks if item["Type"] == "eyeLeft")
        eyeRight_landmark = next(
            item for item in self.landmarks if item["Type"] == "eyeRight")

        self.upperJawlineLeft_point = [int(self.photo_image.size[0] * upperJawlineLeft_landmark["X"]), 
                                       int(self.photo_image.size[1] * upperJawlineLeft_landmark["Y"])]
        self.upperJawlineRight_point = [int(self.photo_image.size[0] * upperJawlineRight_landmark["X"]), 
                                        int(self.photo_image.size[1] * upperJawlineRight_landmark["Y"])]
        self.eyeLeft_point = [int(self.photo_image.size[0] * eyeLeft_landmark["X"]),
                              int(self.photo_image.size[1] * eyeLeft_landmark["Y"])]
        self.eyeRight_point = [int(self.photo_image.size[0] * eyeRight_landmark["X"]),
                               int(self.photo_image.size[1] * eyeRight_landmark["Y"])]

    #Redimensionner l'image du masque pour l'adapter à la largeur de votre visage
    def resize_mask(self):
        face_width = int(np.linalg.norm(list(map(sub, self.upperJawlineLeft_point, self.upperJawlineRight_point))))
        new_hight = int(self.mask_image.size[1]*face_width/self.mask_image.size[0])
        self.mask_image = self.mask_image.resize((face_width, new_hight))

    #Faites pivoter l'image du masque en fonction de l'angle du visage (pas de la diagonale du visage due à la rotation du cou)
    def rotate_mask(self):
        angle = np.arctan2(self.upperJawlineRight_point[1] - self.upperJawlineLeft_point[1],
                           self.upperJawlineRight_point[0] - self.upperJawlineLeft_point[0])
        angle = -np.degrees(angle)  # radian to dgree
        self.mask_image = self.mask_image.rotate(angle, expand=True)

    #Combinez une image photographique et une image de masque
    def match_mask_position(self):
        #Correspondance en utilisant la position des yeux
        face_center = [int((self.eyeLeft_point[0] + self.eyeRight_point[0])/2),
                       int((self.eyeLeft_point[1] + self.eyeRight_point[1])/2)]
        mask_center = [int(self.mask_image.size[0]/2),
                       int(self.mask_image.size[1]/2)]
        x = face_center[0] - mask_center[0]
        y = face_center[1] - mask_center[1]
        self.photo_image.paste(self.mask_image, (x, y), self.mask_image)

    #Enregistrer le nouveau fichier image sur S3
    def save_new_photo(self):
        new_photo_byte_arr = BytesIO()
        self.photo_image.save(new_photo_byte_arr, format="JPEG")
        new_photo_byte_arr = new_photo_byte_arr.getvalue()
        s3.put_object(Bucket=self.bucket, Key=self.new_photo_key,
                      Body=new_photo_byte_arr)

    #Courir
    def run(self):

        self.load_photo_image()

        #Traitement pour le nombre de personnes
        for i in range(len(self.all_landmarks)):
            self.load_mask_image()  #Chargez un nouveau masque à chaque fois
            self.landmarks = self.all_landmarks[i]
            self.landmarks_to_points()
            self.resize_mask()
            self.rotate_mask()
            self.match_mask_position()

        self.save_new_photo()

#fonction principale lambda


def lambda_handler(event, context):
    landmarks = event["landmarks"]
    bucket = event["bucket"]
    photo_key = event["photo_key"]
    new_photo_key = event["new_photo_key"]

    photo_maker = NewPhotoMaker(landmarks, bucket, photo_key, new_photo_key)
    photo_maker.run()

Recommended Posts

Du masque blanc au masque de monstre, développement de l'application de traitement de photos LINE sans serveur sur AWS
Procédure du développement AWS CDK (Python) à la construction de ressources AWS * Pour les débutants