正しい姿勢でテレワークを健康的に ! GluonCV × AWS Lambda で姿勢をチェックしてみよう

from gluoncv import data
print(data.mscoco.keypoints.COCOKeyPoints.KEYPOINTS)

0: 'nose'
*1**:* ** *'left_eye'*
*2**:* ** *'right_eye'*
3: 'left_ear'
4: 'right_ear'
*5**:* ** *'left_shoulder'*
*6**:* ** *'right_shoulder'*
... (略)

import boto3
import sagemaker
from sagemaker import get_execution_role

# 基本的な情報の取得
role = get_execution_role()
region = boto3.session.Session().region_name
account_id = boto3.client('sts').get_caller_identity().get('Account')
session = sagemaker.Session()

# 任意の名前を設定
dirname = 'docker-gluon'
ecr_repository = 'gluon-ecr'

# ECRにアップロードするためのURIを作成
tag = ':latest'
uri_suffix = 'amazonaws.com'
processing_repository_uri = f'{account_id}.dkr.ecr.{region}.{uri_suffix}/{ecr_repository + tag}'

!mkdir -p $dirname

%%writefile $dirname/app.py

from gluoncv import model_zoo, data, utils
from gluoncv.data.transforms.pose import detector_to_alpha_pose, heatmap_to_coord_alpha_pose
import sagemaker
import os

# 指定したディレクトリの中身を消去する
def emptyDir(folder):    
    fileList = os.listdir(folder)  
    for f in fileList:        
        filePath = folder + '/'+f        
        if os.path.isfile(filePath):            
            os.remove(filePath)        
                        
def handler(event, context):
    print(f"Received event:\n{event}\nWith context:\n{context}")
    
    # データがあれば中身を削除する
    data_path = '/tmp/'
    emptyDir(data_path)    
    
    # S3にPUTされたオブジェクト情報を取得する
    bucket = event['Records'][0]['s3']['bucket']['name']
    object_name = event['Records'][0]['s3']['object']['key']
    print(f"Bucket name:\n{bucket}\nObject name:\n{object_name}")
    
    # 画像をS3からダウンロードする
    sess = sagemaker.Session()
    sagemaker.session.Session.download_data(sess, path=data_path, bucket=bucket, key_prefix=object_name)
    
    im_fname = data_path + object_name 
    x, img = data.transforms.presets.yolo.load_test(im_fname, short=512)

    # 物体検出するモデルを使う
    detector = model_zoo.get_model('yolo3_mobilenet1.0_coco', pretrained=True, root='/tmp')
    detector.reset_class(["person"], reuse_weights=['person'])
    class_IDs, scores, bounding_boxs = detector(x)
    pose_input, upscale_bbox = detector_to_alpha_pose(img, class_IDs, scores, bounding_boxs)

    # モデルを削除する
    emptyDir(data_path)  

    # 検知した人のポーズを推定モデルを使う
    pose_net = model_zoo.get_model('alpha_pose_resnet101_v1b_coco', pretrained=True, root='/tmp')
    predicted_heatmap = pose_net(pose_input)
    pred_coords, confidence = heatmap_to_coord_alpha_pose(predicted_heatmap, upscale_bbox)

   # 姿勢推定の結果から姿勢が正しいかどうかを求める
    np_coords=pred_coords.asnumpy()
    
    left_shoulder = np_coords[0, 5]
    right_shoulder = np_coords[0, 6]
    left_eye = np_coords[0, 1]
    right_eye = np_coords[0, 2]
    
    shoulder = abs((left_shoulder[1]-right_shoulder[1])/(left_shoulder[0]-right_shoulder[0]))
    eye = abs((left_eye[1]-right_eye[1])/(left_eye[0]-right_eye[0]))

    result = f'object={object_name}\n'
    if shoulder < 0.1 and eye < 0.1:
        result+="Good: 姿勢を保ちましょう"
    else:
        result+="Bad: 姿勢を見直しましょう"

    #モデルとダウンロードした画像を削除する
    emptyDir(data_path)
        
    return result

import json
import urllib
import urllib.request
import os

# 定数の設定
SLACK_URL = "https://slack.com/api/chat.postMessage"
TOKEN = os.environ['token']
CHANNNEL_ID= os.environ['channel_id']

def lambda_handler(event, context):
    # SNSから送られてきたテキストをjsonにパースする
    json_dict = json.loads(event['Records'][0]['Sns']['Message'])
    text = json_dict['responsePayload']

    # 送信するデータをまとめる
    data = urllib.parse.urlencode(
        (("token", TOKEN), ("channel", CHANNNEL_ID), ("text", text)))
    data = data.encode("ascii")

    # 送信するリクエストの設定
    request = urllib.request.Request(SLACK_URL, data=data, method="POST")
    request.add_header("Content-Type", "application/x-www-form-urlencoded")

    # リクエストの送信
    return urllib.request.urlopen(request).read())