human pose recoginition vanila lstm model vs 가변 길이 lstm model

지난번에는 mediapipe를 이용한 pose estimation으로 사람의 skeleton data를 pose_landmarks & pose_world_landmarks 2가지 방식으로 x,y,z,visibility의 좌표값을 추출하는 방법을 알아보았다.

 

mediapipe pose estimation을 바탕으로 추출한 skeleton data를 sequence data 형식으로 변환하여 lstm model로 학습시켜 humen pose recoginition lstm model을 만들어보고자 한다. 

 

Action Recognition
Action Recognition은 사람이 하는 동작(Action)을 인식하는 기술이다. Action Recognition은 사람의 동작 분류, 인식 및 모니터링 등 다양한 분야에서 활용되고 있다. 예를 들어, 운동 분야에서는 사람의 운동 동작을 인식하여 자세 교정 등을 제공할 수 있으며, 보안 분야에서는 비정상적인 동작을 인식하여 위험 요소를 파악할 수 있다.

 

 

sequence data로 skeleton data 변형하기

Sequence Skeleton Data란?
Sequence Skeleton Data는 Skeleton Data좌표값을 시간의 흐름에 따라 연속적으로 측정한 데이터이다.

예를 들어, 사람이 걸을 때, 뛸 때, 앉을 때, 쓰러질 때, 던질 때 등의 동작은 각각 서로 다른 Skeleton Data로 구성되지만, 연속된 동작으로 발생한 skeleton data를 연속적으로 이어줄 때 Sequence Skeleton Data로 표현된다.

 

Sequence Skeleton Data는 시간의 흐름에 따라 변화하는 Skeleton Data의 연속적인 집합이기 때문에, LSTM과 같은 시계열 데이터를 다루는 딥러닝 모델에 적합하다.

 

먼저 60Frame으로 쌓은 ucf101이라는 dataset에서 추출한 skeleton data를 numpy형식으로 변형해준다.

numpy array 형식으로 변형해주는 이유는 lstm input으로 사용되는 형식이 3차원 tensor형식으로 들어가야하기 때문에 변형이 쉬운 numpy array로 바꿔주는 것이다.

import pandas as pd
import numpy as np

#seed 설정
seed = 0
np.random.seed(seed)
tf.random.set_seed(seed)


seq_length = 60

df = pd.read_csv('mediapipe/detect/pose/ucf101/sequence_3pose_ucf101.csv').set_index('Unnamed: 0')

# train split
df_pushup_train = df[:2460]
df_pullup_train = df[3060:5220]
df_lunge_train = df[6600:9480]

df_train = pd.concat([df_pushup_train, df_pullup_train, df_lunge_train])


# validation split
df_pushup_validation = df[2460:2820]
df_pullup_validation = df[5220:5520]
df_lunge_validation = df[9480:9900]

df_validation = pd.concat([df_pushup_validation, df_pullup_validation, df_lunge_validation])

# test split
df_pushup_test = df[2820:3060]
df_pullup_test = df[5520:6600]
df_lunge_test = df[9900:10740]

df_test = pd.concat([df_pushup_test, df_pullup_test, df_lunge_test])

 

skeleton data를 dataframe으로 60프레임마다 추가하는 방식으로 dataframe에 저장했기 때문에 train, validation, test data 형식으로 index 순서대로 8:1:1 비율로 나눠주었다.

 

from sklearn.preprocessing import LabelEncoder
# x, y data 나누기
x_train = df_train.drop('class', axis=1)
x_validation = df_validation.drop('class', axis=1)
x_test = df_test.drop('class', axis=1)

y_train = df_train['class'].reset_index(drop=True)
y_validation = df_validation['class'].reset_index(drop=True)
y_test = df_test['class'].reset_index(drop=True)

# y_data labeling
e = LabelEncoder()
e.fit(y_train)
e.fit(y_validation)
e.fit(y_test)
y_train = e.transform(y_train)
y_validation = e.transform(y_validation)
y_test = e.transform(y_test)

나눠주었던 train, validation, test dataset에서 x, y 값을 분류해준다.

 

데이터셋 마지막 columns에는 push up, pull up, lunge로 class를 분류해주었기 때문에 class 컬럼을 y_train, y_test, y_val로 따로 분류해준다

 

분류한 y data를 sklearn의 LabelEncoder를 이용하여 0~2 값으로 변형해주었다.

# x_dataset, y_dataset 나누기
x_dataset_train = []
x_dataset_validation = []
x_dataset_test = []

y_dataset_train = []
y_dataset_validation = []
y_dataset_test = []

# x_dataset, y_dataset 리스트에 append
for i in range(0, 7500, 60):
    _x = x_train[i:i + seq_length]
    x_dataset_train.append(_x)
    _y = y_train[i] 
    y_dataset_train.append(_y)

# x_dataset, y_dataset 리스트에 append
for i in range(0, 1080, 60):
    _x = x_validation[i:i + seq_length]
    x_dataset_validation.append(_x)
    _y = y_validation[i] 
    y_dataset_validation.append(_y)

for i in range(0, 2160, 60):
    _x = x_test[i:i + seq_length]
    x_dataset_test.append(_x)
    _y = y_test[i] 
    y_dataset_test.append(_y)

x_dataset_train = np.array(x_dataset_train)
x_dataset_validation = np.array(x_dataset_validation)
x_dataset_test = np.array(x_dataset_test)

y_dataset_train = np.array(y_dataset_train)
y_dataset_validation = np.array(y_dataset_validation)
y_dataset_test = np.array(y_dataset_test)

print(x_dataset_train.shape)
print(y_dataset_train.shape)

print(x_dataset_validation.shape)
print(y_dataset_validation.shape)

print(x_dataset_test.shape)
print(y_dataset_test.shape)

나눠준 TRAIN, TEST, VALIDATION DATA들을 60 sequence로 나누어 빈 list안에 저장해주어야 한다.

그러므로 dataframe에서 60개씩 잘라서 빈 리스트에 넣어주었다

 

마지막으로 numpy로 변형하여 (100, 60, 132)라는 3차원 numpy array 데이터를 만들어줄 수 있다.

 

(100, 60, 132)의 뜻은 (sequence 데이터 개수, time step, features)로

- 100은 학습될 데이터의 개수

- 60은 60sequence를 의미하고 1개의 데이터에 들어가는 60 sequence의 skeleton data

- 132는 skeleton data의 x,y,z,visibility 132개의 좌표값을 의미한다.

 

LSTM MODEL

 

LSTM 모델:
LSTM(Long Short-Term Memory) 모델은 RNN(Recurrent Neural Network)의 한 종류로, 시계열 데이터를 처리하는 데 특화된 딥러닝 모델이다. LSTM 모델은 과거의 데이터를 기억하고, 현재 입력 데이터와 함께 과거의 데이터를 활용하여 예측을 수행한다. 이를 통해, 시계열 데이터에서 발생하는 장기 의존성(long-term dependency) 문제를 해결할 수 있다.

적용 사례:
LSTM 모델을 이용한 Sequence Skeleton Data를 활용한 Action Recognition에는 다양한 적용 사례가 있다. 예를 들어, 스마트 홈 분야에서는 사람의 동작을 인식하여 자동으로 가전 제품을 제어할 수 있으며, 운동 분야에서는 사람의 운동 동작을 인식하여 자세 교정 등을 제공할 수 있다. 또한, 보안 분야에서는 CCTV를 이용하여 사람의 동작을 인식하여 비정상적인 동작을 감지하는 용도로 활용되고 있다.

 

- vailla lstm model

import os
import numpy as np
from tensorflow.keras.utils import to_categorical
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Masking
from tensorflow.keras.preprocessing.sequence import pad_sequences
from tensorflow.keras.layers import Input
import pandas as pd 
import tensorflow as tf

y_train = np.load('sequence_pose/3pose_sample_label.npy', allow_pickle=True)
y_train = pd.DataFrame(y_train)
y_train.columns = ['class']

y_val = np.load('sequence_pose/3pose_sample_label_val.npy', allow_pickle=True)
y_val = pd.DataFrame(y_val)
y_val.columns = ['class']

y_test = np.load('sequence_pose/3pose_sample_label_test.npy', allow_pickle=True)
y_test = pd.DataFrame(y_test)
y_test.columns = ['class']


# 클래스 레이블을 one-hot 벡터로 변환
y_train_label = to_categorical(y_train)

y_val_label = to_categorical(df_y_val)

y_test_label = to_categorical(df_y_test)


# LSTM 모델 구성
inputs = tf.keras.layers.Input(shape=(60, 132))
layer = tf.keras.layers.LSTM(32, return_sequences=False)(layer)
output = tf.keras.layers.Dense(10, activation='softmax')(layer)
model = tf.keras.models.Model(inputs, output)

model.compile(loss='categorical_crossentropy', optimizer=tf.keras.optimizers.Adam(), metrics=['accuracy'])
model.summary()

# 모델 학습
model.fit(X_pad_train, y_train_label, epochs=500, batch_size=5, validation_data=(X_pad_val,y_val_label))


# 모델 평가
loss, acc = model.evaluate(X_pad_test, y_test_label)
print('Test Loss:', loss)
print('Test Accuracy:', acc)

일반적인 vinilla lstm model은 time steps가 고정되어있는 input으로 모델에 들어가게 된다.

 

고정된 sequence data로 lstm을 데이터를 학습시킬 수 있다. 

 

 

- 가변 길이 lstm model

time step이 길이를 고정하지 않고  sequence가 자유롭게 input될 수 있는 lstm model이다.

 

# 시퀀스 데이터 길이를 100으로 맞추고 패딩 처리
# 100 초과시 자름
X_pad_train = pad_sequences(x_train, maxlen=100, dtype='float32', padding='post', truncating='post', value = 10.)
X_pad_val = pad_sequences(x_val, maxlen=100, dtype='float32', padding='post', truncating='post', value = 10.)
X_pad_test = pad_sequences(x_test, maxlen=100, dtype='float32', padding='post', truncating='post', value = 10.)

vaillna lstm과 다른점은 input을 0.0 혹은 1.0값으로 padding 시켜 다른 시퀀스를 같은 시퀀스 길이로 맞춰준다.

 

padding 된 값은 0.0이나 1.0 혹은 다른 숫자로도 적용해줄 수 있다.

 

pad_sequences(x_train, maxlen=100, dtype='float32', padding='post', truncating='post', value = 10.)

- x_train : padding 시킬 데이터

- maxlen : 시퀀스를 얼마까지 padding시킬 것인지 결정

- dtype : padding의 dtype을 결정

- padding : padding 의 위치를 결정, post는 뒤에 paddding, front는 앞에 패딩이 붙는다.

- truncating = 데이터의 길이가 padding의  maxlen을 초과할 때 sequence를 잘라준다.

- value : padding될 값을 설정할 수 있다.

 

# LSTM 모델 구성
inputs = tf.keras.layers.Input(shape=(None, 132))
layer = tf.keras.layers.Masking(mask_value=10.)(inputs)
layer = tf.keras.layers.LSTM(32, return_sequences=False)(layer)
layer = tf.keras.layers.BatchNormalization()(layer)
output = tf.keras.layers.Dense(10, activation='softmax')(layer)
model = tf.keras.models.Model(inputs, output)

model.compile(loss='categorical_crossentropy', optimizer=tf.keras.optimizers.Adam(), metrics=['accuracy'])
model.summary()

가변 time steps model이 vanila lstm과 다른점은 2가지다

 

우선 input 값의 shape에 time steps 값이 정해져있지 않기 때문에 shape=(None, 132)으로 설정한다.

 

다음은 Masking layer로 padding으로 추가한 value 값을 학습시킬 때 Masking layer로 해당 padding 값이 나왔을 때는 학습에 적용시키지 않게끔 설정할 수 있다.

 

만약에 padding value 가 0.0값이라면, 학습 input의 0.0값은 모두 학습에 적용되지 않는다.

 

그러므로 실제로 학습에 적용되어야할 sequence data와 padding value는 겹치지 않도록 주의해야 한다.

 

import os
import numpy as np
from tensorflow.keras.utils import to_categorical
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Masking
from tensorflow.keras.preprocessing.sequence import pad_sequences
from tensorflow.keras.layers import Input
import pandas as pd 
import tensorflow as tf

y_train = np.load('C:/Users/kwonk/Downloads/workspace/mediapipe_py/sequence_pose/3pose_sample_label.npy', allow_pickle=True)
y_train = pd.DataFrame(y_train)
y_train.columns = ['class']

y_val = np.load('C:/Users/kwonk/Downloads/workspace/mediapipe_py/sequence_pose/3pose_sample_label_val.npy', allow_pickle=True)
y_val = pd.DataFrame(y_val)
y_val.columns = ['class']

y_test = np.load('C:/Users/kwonk/Downloads/workspace/mediapipe_py/sequence_pose/3pose_sample_label_test.npy', allow_pickle=True)
y_test = pd.DataFrame(y_test)
y_test.columns = ['class']


# 시퀀스 데이터 길이를 100으로 맞추고 패딩 처리
# 100 초과시 자름
X_pad_train = pad_sequences(x_train, maxlen=100, dtype='float32', padding='post', truncating='post', value = 10.)
X_pad_val = pad_sequences(x_val, maxlen=100, dtype='float32', padding='post', truncating='post', value = 10.)
X_pad_test = pad_sequences(x_test, maxlen=100, dtype='float32', padding='post', truncating='post', value = 10.)

# 클래스 레이블을 one-hot 벡터로 변환
y_train_label = to_categorical(y_train)

y_val_label = to_categorical(df_y_val)

y_test_label = to_categorical(df_y_test)


# LSTM 모델 구성
inputs = tf.keras.layers.Input(shape=(None, 132))
layer = tf.keras.layers.Masking(mask_value=10.)(inputs)
layer = tf.keras.layers.LSTM(32, return_sequences=False)(layer)
layer = tf.keras.layers.BatchNormalization()(layer)
output = tf.keras.layers.Dense(10, activation='softmax')(layer)
model = tf.keras.models.Model(inputs, output)

model.compile(loss='categorical_crossentropy', optimizer=tf.keras.optimizers.Adam(), metrics=['accuracy'])
model.summary()

# 모델 학습
model.fit(X_pad_train, y_train_label, epochs=500, batch_size=5, validation_data=(X_pad_val,y_val_label))


# 모델 평가
loss, acc = model.evaluate(X_pad_test, y_test_label)
print('Test Loss:', loss)
print('Test Accuracy:', acc)