WebサービスをAmazon Rekognition x Serverlessで開発してみた

pip install -r requirements.txt

def detectFacesByRekognition(image_binary: bytes) -> List[List[float]]:
    client = boto3.client('rekognition')
    response = client.detect_faces(
        Image={'Bytes': image_binary}, Attributes=['ALL'])

    faces = list()
    for face_info in response["FaceDetails"]:
        faces.append(face_info["BoundingBox"])
        print(face_info["BoundingBox"])

    return faces

def convertBinaryToImageArray(binary: bytes) -> np.ndarray:
    # 一旦Pillow用の画像データにする
    pil_image = Image.open(io.BytesIO(binary))

    # PIL.JpegImagePlugin.JpegImageFile -> numpy.ndarray(RGB)に変換
    image_array = np.asarray(pil_image)

    return image_array

def pasteEmoji(image: np.ndarray, face: List[float]) -> np.ndarray:
    height, width, _ = image.shape[:3]

    face_top = round(height * face["Top"])
    face_left = round(width * face["Left"])
    face_height = round(height * face["Height"])
    face_width = round(height * face["Width"])

    # 顔の枠を正方形にする
    face_size = round(max(face_height, face_width))

    # 絵文字が顔の真ん中にくるように座標を調整
    if face_height > face_width:
        face_left -= round((face_size - face_width)/2)
    elif face_width > face_height:
        face_top -= round((face_size - face_height)/2)

    # 顔が画像からはみ出しそうなときはサイズを小さくする
    if face_top + face_size > height or face_left + face_size > width:
        face_size = min(height - face_top, width - face_left)

    # -1 : アルファチャンネルで読み込む
    emoji = cv2.imread(
        "emoji/{:0=3}.png".format(random.randrange(1, 71, 1)), -1)

    # BGRA -> RGBAに変換
    emoji = cv2.cvtColor(emoji, cv2.COLOR_BGRA2RGBA)

    # 絵文字の画像サイズを顔の大きさに揃える
    resized_emoji = cv2.resize(
        emoji, (face_size, face_size))

    # 透過処理
    # https://blanktar.jp/blog/2015/02/python-opencv-overlay.html
    mask = resized_emoji[:, :, 3]
    mask = cv2.cvtColor(mask, cv2.COLOR_GRAY2BGR)
    mask = mask / 255.0

    # 予め型キャストしておく
    image = image.astype("float64")

    # 顔に画像を貼り付け
    image[face_top:face_top + face_size,
          face_left:face_left + face_size] *= 1 - mask

    image[face_top:face_top + face_size,
          face_left:face_left + face_size] += mask * resized_emoji[:, :, 0:3]

    return image

def lambda_handler(event, context):
    binary_image = base64.b64decode(event["body"])
    faces = detectFacesByRekognition(binary_image)

    image = convertBinaryToImageArray(binary_image)
    for face in faces:
        image = pasteEmoji(image, face)

    # レスポンス用のバイナリイメージを作成
    # 参考 : https://stackoverrun.com/ja/q/7582988
    response_image = io.BytesIO()
    Image.fromarray(image.astype('uint8')).save(
        response_image, "JPEG")
    body = base64.b64encode(response_image.getvalue()).decode('utf-8')

    # 204で顔が検出されなかったことを表現
    statusCode = 200
    if len(faces) == 0:
        statusCode = 204

    response = {
        "statusCode": statusCode,
        "headers": {
            "Content-Type": "image/jpeg",
            "Access-Control-Allow-Origin": os.environ["EMOJIC_ORIGIN"]
        },
        "body": body,
        "isBase64Encoded": True
    }
    return response

# 抜粋
provider:
    stage: ${opt:stage, self:custom.defaultStage}
    apiKeys:
        - emojic_api_key_${self:provider.stage}

custom:
    defaultStage: dev
    origin:
        dev: http://localhost:3000
        prod: https://emojic.ch
    needApi:
        dev: false
        prod: true

functions:
    upload:
        environment:
            EMOJIC_ORIGIN: ${self:custom.origin.${self:provider.stage}}
        events:
            - http:
                  private: ${self:custom.needApi.${self:provider.stage}}
                  cors:
                      origin: ${self:custom.origin.${self:provider.stage}}

plugins:
    - serverless-python-requirements
    - serverless-apigw-binary
    - serverless-prune-plugin

    apigwBinary:
        types:
            - 'image/jpeg' #image/jpegのレスポンス、リクエストをバイナリで返す

    prune:
        automatic: true
        number: 3

# 抜粋
version: 2
jobs:
    deploy-serverless:
        docker:
            - image: shinsukeabe/circleci-serverless-python3
        steps:
            - checkout
            - restore_cache:
                  keys:
                      - v1-{{ checksum "lambda/package-lock.json" }}
            - run:
                  name: Install Packages
                  command: |
                      cd lambda
                      npm install --save serverless-python-requirements serverless-prune-plugin serverless-apigw-binary
            - save_cache:
                  paths:
                      - lambda/node_modules
                  key: v1-{{ checksum "lambda/package-lock.json" }}
            - run:
                  name: Deploy lambda function in production environment
                  command: |
                      cd lambda
                      serverless deploy -v -s prod

# 抜粋
version: 2
jobs:
    deploy-frontend:
        docker:
            - image: cdssnc/aws-cli
        steps:
            - checkout
            - attach_workspace:
                  at: .
            - deploy:
                  name: Deploy web page
                  command: |
                      cd frontend
                      ./deploy_to_s3.sh

#!/bin/sh
aws s3 sync ./dist s3://$EMOJIC_BUCKET/ --delete
aws cloudfront create-invalidation --distribution-id $EMOJIC_DISTRIBUTION --paths '/*'

# 抜粋
workflows:
    version: 2
    build-and-deploy:
        jobs:
            - test-lambda
            - build
            - deploy-frontend:
                  requires:
                      - build
                      - test-lambda
                  filters:
                      branches:
                          only: master
            - deploy-serverless:
                  requires:
                      - build
                      - test-lambda
                  filters:
                      branches:
                          only: master