AWS Rekognitionを使って人の存在を確認する

全体構成

本当はAWS IoT or Greegrassを使いたかったが社内の通信の都合で実現できず、代わりに直接S3にuploadして、それをトリガーにlambdaを実行する構成に変更

raspiの設定

適切な位置にUSBカメラを設置する
USBカメラで取得した画像をAWS S3にuploadする

camera.py

import cv2
import time
import datetime
import boto3
import os

resource_s3 = boto3.resource('s3')
cap = cv2.VideoCapture(0)
while True:
    d = datetime.datetime.today()
    now = d.strftime("%Y_%m_%d_%H_%M_%S")
    ret, frame = cap.read()
    image_url =  'image/' + now + ".jpg"
    cv2.imwrite(image_url, frame)
    resource_s3.Bucket('mitsu-face-check').upload_file(image_url, image_url)
    time.sleep(10)
    os.remove(image_url)

AWS lambdaの設定

S3へのuploadをトリガーとする

uploadされた画像に対しAWS Rekognitionを使って物体検出(ついでに顔認識)を行う。物体検出のラベルにHumanやPersonがあったら人がいるとする。Rekognitionの結果はS3にcsvファイルとしてuploadする。使用した画像は削除

lambda_function.py

from __future__ import print_function

import os
import json
import urllib
import boto3
import time
import datetime
import sys

s3 = boto3.resource('s3')
rekognition = boto3.client('rekognition')

def lambda_handler(event, context):
    jst = datetime.datetime.now()
    now = jst.strftime("%Y-%m-%d %H:%M:%S")
    bucket_name = 'mitsu-face-check'

    images_bucket = event['Records'][0]['s3']['bucket']['name']
    images_key = urllib.unquote_plus(event['Records'][0]['s3']['object']['key'].encode('utf8'))
    result_folder = 'result/' + images_key.rsplit('/', 1)[1].rsplit('.', 1)[0]
    results_bucket = s3.Bucket(bucket_name)


    label_record = [0]*5
    label_record[0] = (images_key)
    label_record[1] = (now)

    reko_response = rekognition.detect_labels(
            Image={
                'S3Object': {
                    'Bucket': images_bucket,
                    'Name': images_key,
                },
            },
            MaxLabels=20
        )

    for label in reko_response['Labels'] :
        if label["Name"] == "Human" or label["Name"] == "People" or label["Name"] == "Person":
            label_record[2] = (1)


    reko_response_face = rekognition.detect_faces(
        Image={
            'S3Object': {
                'Bucket': images_bucket,
                'Name': images_key,
            },
        },
        Attributes=[
            'ALL',
            ]
    )

    for label in reko_response_face['FaceDetails'] :
        if label["Smile"]['Value'] == True:
            smile_value =  1*label["Smile"]['Confidence']
            label_record[3] = (smile_value)
        if label["Smile"]['Value'] == False:
            smile_value = -1*label["Smile"]['Confidence']
            label_record[3] = (smile_value)

        if label["EyesOpen"]['Value'] == True:
            eye_value =  1*label["EyesOpen"]['Confidence']
            label_record[4] = (eye_value)
        if label["EyesOpen"]['Value'] == False:
            eye_value = -1*label["EyesOpen"]['Confidence']
            label_record[4] = (eye_value)

    label_records = ','.join(map(str, label_record)) + '\n'
    s3_response = results_bucket.put_object( \
        ACL='private', \
        Body=label_records, \
        Key=result_folder + ".csv", \
        ContentType='text/plain' \
    )

    boto3.client('s3').delete_object(Bucket=bucket_name, Key=images_key)
    return str(s3_response)

S3にcsvはuploadされているのを確認

結果の可視化

AWS Athena およびQuicksightを使用する
詳しくは下記サイトより
https://qiita.com/Ichiro_Tsuji/items/d366203c3621e6c15def

上記のサイトを参考にS3のデータからAthenaでデータベースを作成する

athenaの結果をQuicksightによって可視化する
カメラ設置は6階の私の机の前

AM11からPM12までの１時間毎の平均値(1:人がいる、０:人がいない)
AM11にはほぼずっと人がおり、休憩時間のAM12には人がいなくなり、PM1から人がいるようになり、PM8頃には帰宅によって人がいなくなるのがわかる

余談

顔認識の結果(smile度チェック)

AM11からPM12までの１時間毎のsmile度(100:smile、-100:not smile、人がいないと0)
時間が経つにつれてsmile度が下がっている(疲れてる)のか?