5. Pose estimation 에서 스쿼트 자세를 측정하는 방법

이 글은 프로젝트가 마무리된 후 작성된 글입니다.

그래서 시각적 자료가 부족할 수 있습니다.

 

 

우선 이전 글에서 MediaPipe를 다운로드받고

CPU만으로 사람의 각관절을 detection했습니다.

 

예시를 들기위해 배경을 삭제함.

의자에 앉아있는 내 모습을 detection한 모습인데

 

GPU없이도 사람을 정확하게 detection하며

설치도 간편하고, 응용도 쉽습니다.

 

 

 

그러면 이의 좌표값을 통해 접근을 해야하는 상황에서

도큐먼트에 존재하는 32가지의 랜드마크들 중에서 원하는 값을 가져오면 됩니다.

 

 

일단 도큐먼트의 기초적인 설명이 부족하기 때문에

참고한 사이트는

Pradnya1208/Squats-angle-detection-using-OpenCV-and-mediapipe_v1: The purpose of this project is to detect the squat angles which will be helpful for the fitness instructors to provide corrective advice where appropriate. (github.com)

GitHub - Pradnya1208/Squats-angle-detection-using-OpenCV-and-mediapipe_v1: The purpose of this project is to detect the squat an

MIT 에서 스쿼트의 자세를 판단하는 알고리즘에서

좌표값과 각관절에 대한 알고리즘을 참고했다.

 

def calculate_angle(a,b,c):
    a = np.array(a) # First
    b = np.array(b) # Mid
    c = np.array(c) # End
    
    radians = np.arctan2(c[1]-b[1], c[0]-b[0]) - np.arctan2(a[1]-b[1], a[0]-b[0])
    angle = np.abs(radians*180.0/np.pi)
    
    if angle >180.0:
        angle = 360-angle
        
    return angle

좌표값을 입력하면 관절각도를 리턴하는 함수

 

 

그리고 조금 더 코드를 확인해서 확인한건

 

randmark에서 좌표값을 가져오는 방법

try:
	landmarks = results.pose_landmarks.landmark
    shoulder = [landmarks[mp_pose.PoseLandmark.RIGHT_SHOULDER.value].x,landmarks[mp_pose.PoseLandmark.RIGHT_SHOULDER.value].y]
    angle = calculate_angle(shoulder, elbow, wrist)
    # elbow와 wrist가 생략되어있다.
   
 except:
 	pass

랜드마크에 결과값을 저장하고

어깨x, y 

이를통해 관절각도를 계산하는 과정이다.

 

중요한게 try:, except로 

detection 과정은 무한루프로 돌고 있기 때문에

사람이 detection되지 않을 때, 예외처리를 해주어야한다.

 

이정도만 참고를 하였고,

이제 직접 만들기 시작했다.

 

 

먼저 데이터 수집을 위한 코드를 작성했다.

import cv2
import mediapipe as mp
import numpy as np
import pandas as pd
import datetime as dt

# 스쿼트 측정용 코드
def return_today():
    year = dt.datetime.now().year
    month = dt.datetime.now().month
    day = dt.datetime.now().day
    hour = dt.datetime.now().hour
    minute = dt.datetime.now().minute
    second = dt.datetime.now().second
        
    return str(year) + "_"  + str(month) + "_" +\
           str(day) + "_"  + str(hour) + "_"  +\
           str(minute) + "_"  + str(second)



def calculate_angle(a,b,c):
    a = np.array(a) # First
    b = np.array(b) # Mid
    c = np.array(c) # End
    
    radians = np.arctan2(c[1]-b[1], c[0]-b[0]) - np.arctan2(a[1]-b[1], a[0]-b[0])
    angle = np.abs(radians*180.0/np.pi)
    
    if angle >180.0:
        angle = 360-angle
        
    return angle

mp_drawing = mp.solutions.drawing_utils
mp_drawing_styles = mp.solutions.drawing_styles
mp_pose = mp.solutions.pose

col_list = ["Rshoulder_x", "Rshoulder_y",
            "Lshoulder_x", "Lshoulder_y", 
            "Rhip_x", "Rhip_y",
            "Lhip_x", "Lhip_y", 
            "Rknee_x", "Rknee_y",
            "Lknee_x", "Lknee_y",
            "Rankle_x", "Rankle_y",
            "Lankle_x", "Lankle_y",
           "Rknee_angle", "Lknee_angle",
           "Rhip_angle", "Lhip_angle",
           "labels"]

           # labels 추가 해줘야함.

df = pd.DataFrame(columns = col_list)
count = 0


cap = cv2.VideoCapture(0)
with mp_pose.Pose(
    min_detection_confidence=0.5,
    min_tracking_confidence=0.5) as pose:
  while cap.isOpened():
    success, image = cap.read()
    if not success:
      print("Ignoring empty camera frame.")
      # If loading a video, use 'break' instead of 'continue'.
      continue

    # To improve performance, optionally mark the image as not writeable to
    # pass by reference.
    image.flags.writeable = False
    image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
    results = pose.process(image)

    # Draw the pose annotation on the image.
    image.flags.writeable = True
    image = cv2.cvtColor(image, cv2.COLOR_RGB2BGR)

    mp_drawing.draw_landmarks(
        image,
        results.pose_landmarks,
        mp_pose.POSE_CONNECTIONS,
        landmark_drawing_spec=mp_drawing_styles.get_default_pose_landmarks_style())
    # Flip the image horizontally for a selfie-view display.

    # 렌드마크가0~32 를 가진다.
   
    try:
      landmarks = results.pose_landmarks.landmark
      Rshoulder = [landmarks[mp_pose.PoseLandmark.RIGHT_SHOULDER.value].x,landmarks[mp_pose.PoseLandmark.RIGHT_SHOULDER.value].y]
      Rhip = [landmarks[mp_pose.PoseLandmark.RIGHT_HIP.value].x,landmarks[mp_pose.PoseLandmark.RIGHT_HIP.value].y]
      Rknee = [landmarks[mp_pose.PoseLandmark.RIGHT_KNEE.value].x,landmarks[mp_pose.PoseLandmark.RIGHT_KNEE.value].y]
      Rankle = [landmarks[mp_pose.PoseLandmark.RIGHT_ANKLE.value].x,landmarks[mp_pose.PoseLandmark.RIGHT_ANKLE.value].y]
      
      Lshoulder = [landmarks[mp_pose.PoseLandmark.LEFT_SHOULDER.value].x,landmarks[mp_pose.PoseLandmark.LEFT_SHOULDER.value].y]
      Lhip = [landmarks[mp_pose.PoseLandmark.LEFT_HIP.value].x,landmarks[mp_pose.PoseLandmark.LEFT_HIP.value].y]
      Lknee = [landmarks[mp_pose.PoseLandmark.LEFT_KNEE.value].x,landmarks[mp_pose.PoseLandmark.LEFT_KNEE.value].y]
      Lankle = [landmarks[mp_pose.PoseLandmark.LEFT_ANKLE.value].x,landmarks[mp_pose.PoseLandmark.LEFT_ANKLE.value].y]

      Rangle_knee = calculate_angle(Rhip, Rknee, Rankle)
      Rknee_angle = 180-Rangle_knee

      Langle_knee = calculate_angle(Lhip, Lknee, Lankle)
      Lknee_angle = 180-Langle_knee

      Rangle_hip = calculate_angle(Rshoulder, Rhip, Rknee)
      Rhip_angle = 180-Rangle_hip

      Langle_hip = calculate_angle(Lshoulder, Lhip, Lknee)
      Lhip_angle = 180-Langle_hip

      print("detect")
    
    

      if count > 7:
        df.loc[len(df)] = [Rshoulder[0], Rshoulder[1], Lshoulder[0], Lshoulder[1],
                           Rhip[0], Rhip[1], Lhip[0], Lhip[1], 
                           Rknee[0], Rknee[1], Lknee[0], Lknee[1], 
                           Rankle[0], Rankle[1], Lankle[0], Lankle[1], 
                           Rknee_angle, Lknee_angle, Rhip_angle, Lhip_angle, 2 ] # 끝에 라벨 추가 0: 준비///  1: 스쿼트 // 2: 틀린자세
        count = 0
      
    except:
      print("not find")

    print(count)
    cv2.imshow('MediaPipe Pose', cv2.flip(image, 1))
    count += 1


    if cv2.waitKey(5) & 0xFF == 27:
      title = "./" + return_today() + '.csv'
      df.to_csv(title, encoding="UTF-8")
      break

cap.release()

 

프로그래밍상에서 버튼을 누르면 저장을 멈추고, 버튼을 누르면 시작하게 만들려고 했는데, 

그냥 귀찮은거 때문에 걍 자세를 잡으면 프로그램을  실행하고 말지 하고 작성을 했다.

 

스쿼트를 총 2가지로 나누어 접근을 하였다.

1. 준비자세 (서있는 자세) (라벨 0)

 

 

2. 스쿼트 자세 (라벨 1)

그리고 모든 방향에서 적용시키려면 360도 방향에서 모든 데이터를 수집해야한다.

이는 난이도도 난이도지만, 데이터 측정의 어려움때문에 카메라를 고정하는 것으로 결정하였다.

 

 

이 두개의 데이터를 측정하고, 어떤 모델을 사용해야할지 고민을 했다.

결국 각각의 좌표값은 조건문으로 해결될 것이라 판단하였고

결정나무 모델로 제작을 진행하기로 했다.

 

다음은 결정나무 모델을 만드는 코드이다.

 

import pandas as pd
import numpy as np
from glob import glob

대충 파일을 하나로 묶어주는 모듈과 데이터분석에 필요한 모듈을 import하고

 

csv_list = glob('./*.csv')

col_list = ["Rshoulder_x", "Rshoulder_y",
            "Lshoulder_x", "Lshoulder_y", 
            "Rhip_x", "Rhip_y",
            "Lhip_x", "Lhip_y", 
            "Rknee_x", "Rknee_y",
            "Lknee_x", "Lknee_y",
            "Rankle_x", "Rankle_y",
            "Lankle_x", "Lankle_y",
           "Rknee_angle", "Lknee_angle",
           "Rhip_angle", "Lhip_angle",
           "labels"]

df = pd.DataFrame(columns = col_list)

csv 파일을 모두 불러온다.

그리고 각 관절에 관한 list형식의 변수를 선언하고

이를 DataFrame화 한다.

 

for i in range(len(csv_list)):
    df_temp = pd.read_csv(csv_list[i])
    df = pd.concat([df, df_temp])

df

그러면  csv 파일을 모두 하나의 파일로 합친다.

 

labels (라벨인지 레이블인지)

여기서는 2로 표시되는데 그 이야기는 나중에 하도록 하자.

 

df = df.reset_index()
df = df.drop(['index', 'Unnamed: 0'], axis = 1 )

쓸모없는 행을 삭제하고 인덱스를 맞춘다.

 

이제 결정나무 모델을 제작해보자.

from sklearn.tree import DecisionTreeClassifier
from sklearn.metrics import accuracy_score


from sklearn.model_selection import train_test_split
features = df.drop(['labels'],  axis = 1 )
labels = df["labels"]
X_train, X_test, y_train, y_test = train_test_split(features, labels, test_size=0.2, random_state=13)

from sklearn.model_selection import GridSearchCV
from sklearn.tree import DecisionTreeClassifier

params = {'max_depth' : [1, 2, 3,  4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16]}
sq = DecisionTreeClassifier(max_depth=2, random_state=13)

gridsearch = GridSearchCV(estimator=sq, param_grid=params, cv=5)
gridsearch.fit(features, labels)

(지금와서 생각해보면 train_test_split은 할 필요가 없을것이라 생각한다)

우선 GridSearch를 통해 max_depth가 몇개일때 최적인지 찾아낸다.

 

 

 

 

from sklearn.metrics import accuracy_score

y_pred_tr = sq.predict(X_train)
accuracy_score(y_train, y_pred_tr)

어큐러씨를 측정하면

 

보통 0.99~1.0 이 나오는데

어큐러씨가 높다고 좋은 모델이라고 할 수가 없다. 그 이유는 나중에 언급하도록하고

 

 

from mlxtend.plotting import plot_decision_regions
import matplotlib.pyplot as plt
from sklearn import tree

fig = plt.figure(figsize=(15, 8))
_ = tree.plot_tree(sq,
                  feature_names=X_train.columns,
                  class_names = ['0.0', '1.0', '2.0'],
                   rounded=True,
                  filled=True)
plt.figure(figsize=(14, 8))

뭐 이렇게 하면 컴퓨터가 어떻게 판단했는지 조건문형식으로 볼 수 있다.

 

아마 이런식으로 그래프가 그려지지 않을 것이다.

(자료가 아무래도 다끝나고 하는 것이기 때문에 차이가 있다.)

import pickle
import joblib

pickle.dump(sq, open('./val_model.sav', 'wb'))

 

이 모델을 저장하고 활용하면된다.

 

여튼 가장 처음 만들었던 모델은 이렇게 판단했었다.

어깨 높이가 높으면 ? 준비자세

낮으면 스쿼트자세네? 라고 판단을 해버리는 상황이 발생했다.

 

 

그래서 어깨좌표값을 빼고 학습을 시켰을때

스쿼트가 잘 되긴 하나

 

살짝만 움직여도 스쿼트 자세로 인식한다거나

 

 

무릎만 움직여도 스쿼트네? 

라고 인식했다.

 

정확도가 1인 이유는 이 때문이였다.

이 때까지만 해도 어떤점이 문제인지 찾질 못했다.

도대체 뭐가 문제일까?

 

 

 

여기서 한가지 찾아낸 점은 바로

틀린 자세를 데이터로 집어넣어야한다는 것이다.

 

 

스쿼트를 측정할때 우리를 속썩였던 

무릎을 굽히는 자세라던가

스쿼트를 진행중일 때의 자세라던가

 

이를 라벨 2를 달아 학습을 시켰고

 

이처럼 꽤 복잡한 모델이 만들어졌다.

결과는 성공적이였다.

 

 

 

결국 

1. 준비자세

2. 정자세

3. 틀린자세

를 측정하면 어떤 운동이건 구현할  수 있다는 것을 알게 되었다.

 

이는 푸쉬업의 모델이다.

 

 

여튼 이렇게 만들어진 데이터는 

자세를 측정하는 코드를 통해 실행된다.

 

여기서 추가된건 스레드에 관한 내용이다.

import cv2
import mediapipe as mp
import numpy as np
import pandas as pd
import datetime as dt
import pickle
from sklearn.metrics import accuracy_score
from sklearn import tree
from sklearn.tree import DecisionTreeClassifier
from playsound import playsound
import threading
import time
from queue import Queue

# 스쿼트 판단용 코드



def play_sound(q):
  while True:
    data = q.get()
    if data is None:
      pass

    else: 
      data = str(data)

      if data == "1000":
        playsound("./ready.mp3")
      
      elif data == "10000":
        break

      elif len(data) == 1:
        if data == "0":
          playsound("./pp.mp3")

        elif data[-1] > "0":
          playsound("./sound/" + str(data) + ".mp3" )
    
      elif len(data) == 2:
        if data[-1] == "0":
          if data[0] == "1":
            playsound("./sound/10.mp3")
          
          elif data[0] > "1":
            playsound("./sound/" + data[0] + ".mp3" ) 
            playsound("./sound/10.mp3")
          
        elif data[-1] > "0":
          playsound("./sound/" + data[-1] + ".mp3" )

      
      



def calculate_angle(a,b,c):
    a = np.array(a) # First
    b = np.array(b) # Mid
    c = np.array(c) # End
    
    radians = np.arctan2(c[1]-b[1], c[0]-b[0]) - np.arctan2(a[1]-b[1], a[0]-b[0])
    angle = np.abs(radians*180.0/np.pi)
    
    if angle >180.0:
        angle = 360-angle
        
    return angle

def play_cam(q):

  mp_drawing = mp.solutions.drawing_utils
  mp_drawing_styles = mp.solutions.drawing_styles
  mp_pose = mp.solutions.pose

  # 자세측정 지연시간?
  time_count = 0

  # 운동 카운트
  ex_count = 0

  # 운동 상태
  ex_status = 0

  status = "start"

  loaded_model = pickle.load(open('./val_model.sav', 'rb')) 

  cap = cv2.VideoCapture(0)
  with mp_pose.Pose(
      min_detection_confidence=0.5,
      min_tracking_confidence=0.5) as pose:
    while cap.isOpened():
      success, image = cap.read()
      if not success:
        print("Ignoring empty camera frame.")
        # If loading a video, use 'break' instead of 'continue'.
        continue

      # To improve performance, optionally mark the image as not writeable to
      # pass by reference.
      image.flags.writeable = False
      image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
      results = pose.process(image)

      # Draw the pose annotation on the image.
      image.flags.writeable = True
      image = cv2.cvtColor(image, cv2.COLOR_RGB2BGR)

      mp_drawing.draw_landmarks(
          image,
          results.pose_landmarks,
          mp_pose.POSE_CONNECTIONS,
          landmark_drawing_spec=mp_drawing_styles.get_default_pose_landmarks_style())
      # Flip the image horizontally for a selfie-view display.11

      # 렌드마크가0~32 를 가진다.
    
      try:
        landmarks = results.pose_landmarks.landmark
        Rshoulder = [landmarks[mp_pose.PoseLandmark.RIGHT_SHOULDER.value].x,landmarks[mp_pose.PoseLandmark.RIGHT_SHOULDER.value].y]
        Rhip = [landmarks[mp_pose.PoseLandmark.RIGHT_HIP.value].x,landmarks[mp_pose.PoseLandmark.RIGHT_HIP.value].y]
        Rknee = [landmarks[mp_pose.PoseLandmark.RIGHT_KNEE.value].x,landmarks[mp_pose.PoseLandmark.RIGHT_KNEE.value].y]
        Rankle = [landmarks[mp_pose.PoseLandmark.RIGHT_ANKLE.value].x,landmarks[mp_pose.PoseLandmark.RIGHT_ANKLE.value].y]
        
        Lshoulder = [landmarks[mp_pose.PoseLandmark.LEFT_SHOULDER.value].x,landmarks[mp_pose.PoseLandmark.LEFT_SHOULDER.value].y]
        Lhip = [landmarks[mp_pose.PoseLandmark.LEFT_HIP.value].x,landmarks[mp_pose.PoseLandmark.LEFT_HIP.value].y]
        Lknee = [landmarks[mp_pose.PoseLandmark.LEFT_KNEE.value].x,landmarks[mp_pose.PoseLandmark.LEFT_KNEE.value].y]
        Lankle = [landmarks[mp_pose.PoseLandmark.LEFT_ANKLE.value].x,landmarks[mp_pose.PoseLandmark.LEFT_ANKLE.value].y]

        Rangle_knee = calculate_angle(Rhip, Rknee, Rankle)
        Rknee_angle = 180-Rangle_knee

        Langle_knee = calculate_angle(Lhip, Lknee, Lankle)
        Lknee_angle = 180-Langle_knee

        Rangle_hip = calculate_angle(Rshoulder, Rhip, Rknee)
        Rhip_angle = 180-Rangle_hip

        Langle_hip = calculate_angle(Lshoulder, Lhip, Lknee)
        Lhip_angle = 180-Langle_hip

      except:
        pass

      try:
        if time_count > 10:
          time_count = 0
          pre = np.array([[Rshoulder[0], Rshoulder[1], Lshoulder[0], Lshoulder[1],
                            Rhip[0], Rhip[1], Lhip[0], Lhip[1], 
                            Rknee[0], Rknee[1], Lknee[0], Lknee[1], 
                            Rankle[0], Rankle[1], Lankle[0], Lankle[1], 
                            Rknee_angle, Lknee_angle, Rhip_angle, Lhip_angle
                            ]])
          

          #print(np.argmax(loaded_model.predict_proba(pre).tolist()))
          if 0 == np.argmax(loaded_model.predict_proba(pre).tolist()):
            if ex_status == 0:
              ex_status = 1
              status = "ready"
              q.put(1000)

            elif ex_status == 2:
              ex_status = 1
              status = "ready"
              ex_count += 1
              q.put(ex_count)

          elif 1 == np.argmax(loaded_model.predict_proba(pre).tolist()):
            if ex_status == 1:
              ex_status = 2
              status = "squat"
              q.put(0)
            
      except:
        print("model error")
          
          


      cv2.flip(image, 1)
      cv2.putText(image, str(ex_count), org=(30, 60), 
          fontFace=cv2.FONT_HERSHEY_SIMPLEX, fontScale=1, 
          color=(0,0,255),thickness=3, lineType=cv2.LINE_AA)

      cv2.putText(image, status, org=(30, 30), 
          fontFace=cv2.FONT_HERSHEY_SIMPLEX, fontScale=1, 
          color=(0,0,255),thickness=3, lineType=cv2.LINE_AA)

      cv2.imshow('MediaPipe Pose',image)
      time_count += 1


      if cv2.waitKey(5) & 0xFF == 27:
        playsound("./sound/finish.mp3")
        q.put(10000)
        break

  cap.release()
  q.put(None)



if __name__ == '__main__':
  playsound("./sound/start.mp3")
  q = Queue()
  t1 = threading.Thread(target=play_sound, args=(q, ))
  t2 = threading.Thread(target=play_cam, args=(q, ))
  t2.start()
  t1.start()

스레드를 사용한건

소리를 재생하면 소리가 재생되는 시간동안 멈추기 때문이였다.

 

 

이 스레드가 진짜 머리를 터지게 할 줄은 이때까지는 몰랐다.

 

여튼 이 코드를 활용해서 결과를 보여주는게 스쿼트는 없고

푸쉬업을 측정하는 결과물이 있어서 gif로 남긴다.

이 gif에는 스레드가 걸려있지 않네..

여튼 스레드가 걸린건 나중에 시연영상을 올리면 보시도록

 

 

 

 

이제 다음글은 데이터베이스와 여러운동을 하나로 구현하는 과정을 설명하겠습니다.