MoveNet および TensorFlow Lite での人間姿勢分類

このノートブックは、MoveNet および TensorFlow Lite を使用して、姿勢分類モデルを学習する方法について説明します。結果として、新しい TensorFlow Lite が MoveNet モデルの出力を入力として受け取り、ヨガのポーズの名前といった姿勢分類を出力します。

このノートブックの手順は次の 3 つの部分に分かれています。

  • 第 1 部: 姿勢分類学習データを、MoveNet モデルによって検出されたランドマーク (体の主要な点) とグラウンドトゥルース姿勢ラベルを指定する CSV ファイルに再処理します。
  • 第 2 部: CSV ファイルのランドマーク座標を入力として受け取り、予測されたラベルを出力する姿勢分類モデルを構築して学習させます。
  • 第 3 部: 姿勢分類モデルを TFLite に変換します。

既定では、このノートブックは、ヨガのポーズというラベルが付けられた画像データセットを使用しますが、第 1 部のセクションでは、独自の姿勢の画像データセットをアップロードできます。

TensorFlow.org で表示 Google Colab で実行 GitHubでソースを表示 ノートブックをダウンロード TF Hub モデルを見る

準備

このセクションでは、必要なライブラリをインポートし、複数の関数を定義して、学習画像を、ランドマーク座標とグラウンドトゥルースラベルを含む CSV に再処理します。

ここでは、観察可能な事象は発生しませんが、非表示のコードセルを拡大すると、後から呼び出す一部の関数の実装を確認することができます。

すべての詳細を知らずに CSV ファイルのみを作成したい場合は、このセクションの手順を実行し、第 1 部に進んでください。

pip install -q opencv-python

import csv
import cv2
import itertools
import numpy as np
import pandas as pd
import os
import sys
import tempfile
import tqdm

from matplotlib import pyplot as plt
from matplotlib.collections import LineCollection

import tensorflow as tf
import tensorflow_hub as hub
from tensorflow import keras

from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, classification_report, confusion_matrix

MoveNet を使用して姿勢推定を実行するコード

Functions to run pose estimation with MoveNet

Functions to visualize the pose estimation results.

Code to load the images, detect pose landmarks and save them into a CSV file

(Optional) Code snippet to try out the Movenet pose estimation logic

第 1 部: 入力画像の再処理

姿勢分類器の入力は MoveNet モデルからの出力ランドマークであるため、MoveNet 経由でラベル付けされた画像を実行し、すべてのランドマークデータとグラウンドトゥルースラベルを CSV ファイルに取り込んで、学習データセットを生成する必要があります。

このチュートリアル用に提供しているデータセットは、CG で生成されたヨガのポーズのデータセットです。CG で生成された、5 つのヨガのポーズを取っている複数のモデルの画像が含まれています。このディレクトリはすでにtrain データセットと test データセットに分割されています。

このセクションでは、ヨガデータセットをダウンロードして、それを MoveNet 経由で実行し、すべてのランドマークを CSV ファイルに取り込めるようにします。ただし、ヨガデータセットを MoveNet に入力し、この CSV ファイルを生成するには、約 15 分かかります。このため、代替策として、次の is_skip_step_1 パラメータを True に設定して、あらかじめ準備された既存のヨガデータセット CSV ファイルをダウンロードできます。このようにすると、この手順を省略して、この前処理ステップで作成されるのと同じ CSV ファイルをダウンロードできます。

逆に、自分の独自の画像データセットで姿勢分類器を学習させたい場合は、画像をアップロードして、この前処理ステップを実行 (is_skip_step_1 を False に設定すること)し、次の手順に従って独自の姿勢データセットをアップロードする必要があります。

(任意) 独自の姿勢データセットのアップロード

独自のラベル付けされた姿勢 (ヨガのポーズだけではなく、どのような姿勢でもかまいません) で姿勢分類器を学習させたい場合は、次の手順に従います。

  1. 上記の use_custom_dataset オプションを True に設定します。

  2. 画像データセットが入ったフォルダを含むアーカイブファイル (ZIP、TAR など) を準備します。フォルダには、次のように姿勢の画像がソートされた状態で格納されている必要があります。

すでにデータセットを学習セットとテストセットに分割している場合は、dataset_is_split を True に設定します。つまり、画像フォルダには、次のように、「train」ディレクトリと「test」ディレクトリが必要です。

yogaposes/ |_ train/ |__ downdog/ |______ 00000128.jpg |______ ... |__ test/ |__ downdog/ |______ 00000181.jpg |______ ...


Or, if your dataset is NOT split yet, then set
`dataset_is_split` to **False** and we'll split it up based
on a specified split fraction. That is, your uploaded images
folder should look like this:

yogaposes/ |_ downdog/ |______ 00000128.jpg |______ 00000181.jpg |______ ... |__ goddess/ |______ 00000243.jpg |______ 00000306.jpg |______ ...
  1. 左の Files タブ (フォルダアイコン) をクリックして、Upload to session storage (ファイルアイコン) をクリックします。
  2. アーカイブファイルを選択し、アップロードが完了するまで待ってから、続行します。
  3. 次のコードブロックを編集し、アーカイブファイルと画像ディレクトリの名前を指定します。(既定では、ZIP ファイルに設定されているため、別の形式のアーカイブファイルを使用する場合は、該当する部分を修正する必要があります。)
  4. ノートブックの残りの部分を実行します。

if use_custom_dataset:
  # ATTENTION:
  # You must edit these two lines to match your archive and images folder name:
  # !tar -xf YOUR_DATASET_ARCHIVE_NAME.tar
  !unzip -q YOUR_DATASET_ARCHIVE_NAME.zip
  dataset_in = 'YOUR_DATASET_DIR_NAME'

  # You can leave the rest alone:
  if not os.path.isdir(dataset_in):
    raise Exception("dataset_in is not a valid directory")
  if dataset_is_split:
    IMAGES_ROOT = dataset_in
  else:
    dataset_out = 'split_' + dataset_in
    split_into_train_test(dataset_in, dataset_out, test_split=0.2)
    IMAGES_ROOT = dataset_out

注意: split_into_train_test() を使用してデータセットを分割している場合は、すべての画像を PNG、JPEG、または BMP にする必要があります。他のファイルタイプは無視されます。

ヨガデータセットのダウンロード

if not is_skip_step_1 and not use_custom_dataset:
  !wget -O yoga_poses.zip http://download.tensorflow.org/data/pose_classification/yoga_poses.zip
  !unzip -q yoga_poses.zip -d yoga_cg
  IMAGES_ROOT = "yoga_cg"

TRAIN データセットの前処理

if not is_skip_step_1:
  images_in_train_folder = os.path.join(IMAGES_ROOT, 'train')
  images_out_train_folder = 'poses_images_out_train'
  csvs_out_train_path = 'train_data.csv'

  preprocessor = MoveNetPreprocessor(
      images_in_folder=images_in_train_folder,
      images_out_folder=images_out_train_folder,
      csvs_out_path=csvs_out_train_path,
  )

  preprocessor.process(per_pose_class_limit=None)

TEST データセットの前処理

if not is_skip_step_1:
  images_in_test_folder = os.path.join(IMAGES_ROOT, 'test')
  images_out_test_folder = 'poses_images_out_test'
  csvs_out_test_path = 'test_data.csv'

  preprocessor = MoveNetPreprocessor(
      images_in_folder=images_in_test_folder,
      images_out_folder=images_out_test_folder,
      csvs_out_path=csvs_out_test_path,
  )

  preprocessor.process(per_pose_class_limit=None)

第 2 部: ランドマーク座標を入力として受け取り、予測されたラベルを出力する姿勢分類モデルを学習させる。

ランドマーク座標を取り、入力画像内の人間の姿勢分類を予測する TensorFlow モデルを構築します。このモデルには次の下位モデルがあります。

  • 下位モデル 1 は、検出されたランドマーク座標から姿勢の埋め込み (特徴量ベクトル) を計算します。
  • 下位モデル 2 は、複数の Dense レイヤー経由で姿勢埋め込みを入力し、姿勢分類を予測します。

第 1 部で前処理されたデータセットに基づき、モデルを学習させます。

(任意) 第 1 部を実行しなかった場合に前処理されたデータセットをダウンロードする

# Download the preprocessed CSV files which are the same as the output of step 1
if is_skip_step_1:
  !wget -O train_data.csv http://download.tensorflow.org/data/pose_classification/yoga_train_data.csv
  !wget -O test_data.csv http://download.tensorflow.org/data/pose_classification/yoga_test_data.csv

  csvs_out_train_path = 'train_data.csv'
  csvs_out_test_path = 'test_data.csv'
  is_skipped_step_1 = True

前処理された CSV を TRAIN および TEST データセットに読み込む

def load_pose_landmarks(csv_path):
  """Loads a CSV created by MoveNetPreprocessor.

  Returns:
    X: Detected landmark coordinates and scores of shape (N, 17 * 3)
    y: Ground truth labels of shape (N, label_count)
    classes: The list of all class names found in the dataset
    dataframe: The CSV loaded as a Pandas dataframe features (X) and ground
      truth labels (y) to use later to train a pose classification model.
  """

  # Load the CSV file
  dataframe = pd.read_csv(csv_path)
  df_to_process = dataframe.copy()

  # Drop the file_name columns as you don't need it during training.
  df_to_process.drop(columns=['file_name'], inplace=True)

  # Extract the list of class names
  classes = df_to_process.pop('class_name').unique()

  # Extract the labels
  y = df_to_process.pop('class_no')

  # Convert the input features and labels into the correct format for training.
  X = df_to_process.astype('float64')
  y = keras.utils.to_categorical(y)

  return X, y, classes, dataframe

元の TRAIN データセットを読み込んで、TRAIN (データの 85%) と VALIDATE (残りの 15%) に分割します。

# Load the train data
X, y, class_names, _ = load_pose_landmarks(csvs_out_train_path)

# Split training data (X, y) into (X_train, y_train) and (X_val, y_val)
X_train, X_val, y_train, y_val = train_test_split(X, y,
                                                  test_size=0.15)

# Load the test data
X_test, y_test, _, df_test = load_pose_landmarks(csvs_out_test_path)

姿勢ランドマークを姿勢埋め込み (特徴量ベクトル) に変換し、姿勢分類を実行する関数を定義する

次に、次の手順で、ランドマーク座標を特徴量ベクトルに変換します。

  1. 姿勢の中心を元の位置に移動します。
  2. 姿勢のサイズが 1 になるように、姿勢を調整します。
  3. これらの座標を特徴量ベクトルに一次元化します。

この特徴量ベクトルを使用して、ニュートラルネットワークに基づく姿勢分類器を学習させます。

def get_center_point(landmarks, left_bodypart, right_bodypart):
  """Calculates the center point of the two given landmarks."""

  left = tf.gather(landmarks, left_bodypart.value, axis=1)
  right = tf.gather(landmarks, right_bodypart.value, axis=1)
  center = left * 0.5 + right * 0.5
  return center


def get_pose_size(landmarks, torso_size_multiplier=2.5):
  """Calculates pose size.

  It is the maximum of two values:
    * Torso size multiplied by `torso_size_multiplier`
    * Maximum distance from pose center to any pose landmark
  """
  # Hips center
  hips_center = get_center_point(landmarks, BodyPart.LEFT_HIP, 
                                 BodyPart.RIGHT_HIP)

  # Shoulders center
  shoulders_center = get_center_point(landmarks, BodyPart.LEFT_SHOULDER,
                                      BodyPart.RIGHT_SHOULDER)

  # Torso size as the minimum body size
  torso_size = tf.linalg.norm(shoulders_center - hips_center)

  # Pose center
  pose_center_new = get_center_point(landmarks, BodyPart.LEFT_HIP, 
                                     BodyPart.RIGHT_HIP)
  pose_center_new = tf.expand_dims(pose_center_new, axis=1)
  # Broadcast the pose center to the same size as the landmark vector to
  # perform substraction
  pose_center_new = tf.broadcast_to(pose_center_new,
                                    [tf.size(landmarks) // (17*2), 17, 2])

  # Dist to pose center
  d = tf.gather(landmarks - pose_center_new, 0, axis=0,
                name="dist_to_pose_center")
  # Max dist to pose center
  max_dist = tf.reduce_max(tf.linalg.norm(d, axis=0))

  # Normalize scale
  pose_size = tf.maximum(torso_size * torso_size_multiplier, max_dist)

  return pose_size


def normalize_pose_landmarks(landmarks):
  """Normalizes the landmarks translation by moving the pose center to (0,0) and
  scaling it to a constant pose size.
  """
  # Move landmarks so that the pose center becomes (0,0)
  pose_center = get_center_point(landmarks, BodyPart.LEFT_HIP, 
                                 BodyPart.RIGHT_HIP)
  pose_center = tf.expand_dims(pose_center, axis=1)
  # Broadcast the pose center to the same size as the landmark vector to perform
  # substraction
  pose_center = tf.broadcast_to(pose_center, 
                                [tf.size(landmarks) // (17*2), 17, 2])
  landmarks = landmarks - pose_center

  # Scale the landmarks to a constant pose size
  pose_size = get_pose_size(landmarks)
  landmarks /= pose_size

  return landmarks


def landmarks_to_embedding(landmarks_and_scores):
  """Converts the input landmarks into a pose embedding."""
  # Reshape the flat input into a matrix with shape=(17, 3)
  reshaped_inputs = keras.layers.Reshape((17, 3))(landmarks_and_scores)

  # Normalize landmarks 2D
  landmarks = normalize_pose_landmarks(reshaped_inputs[:, :, :2])

  # Flatten the normalized landmark coordinates into a vector
  embedding = keras.layers.Flatten()(landmarks)

  return embedding

姿勢分類の Keras モデルの定義

Keras モデルは検出された姿勢ランドマークを取り、姿勢埋め込みを計算し、姿勢分類を予測します。

# Define the model
inputs = tf.keras.Input(shape=(51))
embedding = landmarks_to_embedding(inputs)

layer = keras.layers.Dense(128, activation=tf.nn.relu6)(embedding)
layer = keras.layers.Dropout(0.5)(layer)
layer = keras.layers.Dense(64, activation=tf.nn.relu6)(layer)
layer = keras.layers.Dropout(0.5)(layer)
outputs = keras.layers.Dense(len(class_names), activation="softmax")(layer)

model = keras.Model(inputs, outputs)
model.summary()

model.compile(
    optimizer='adam',
    loss='categorical_crossentropy',
    metrics=['accuracy']
)

# Add a checkpoint callback to store the checkpoint that has the highest
# validation accuracy.
checkpoint_path = "weights.best.hdf5"
checkpoint = keras.callbacks.ModelCheckpoint(checkpoint_path,
                             monitor='val_accuracy',
                             verbose=1,
                             save_best_only=True,
                             mode='max')
earlystopping = keras.callbacks.EarlyStopping(monitor='val_accuracy', 
                                              patience=20)

# Start training
history = model.fit(X_train, y_train,
                    epochs=200,
                    batch_size=16,
                    validation_data=(X_val, y_val),
                    callbacks=[checkpoint, earlystopping])

# Visualize the training history to see whether you're overfitting.
plt.plot(history.history['accuracy'])
plt.plot(history.history['val_accuracy'])
plt.title('Model accuracy')
plt.ylabel('accuracy')
plt.xlabel('epoch')
plt.legend(['TRAIN', 'VAL'], loc='lower right')
plt.show()

# Evaluate the model using the TEST dataset
loss, accuracy = model.evaluate(X_test, y_test)

モデルパフォーマンスを効果的に理解するための混同行列を描画する

def plot_confusion_matrix(cm, classes,
                          normalize=False,
                          title='Confusion matrix',
                          cmap=plt.cm.Blues):
  """Plots the confusion matrix."""
  if normalize:
    cm = cm.astype('float') / cm.sum(axis=1)[:, np.newaxis]
    print("Normalized confusion matrix")
  else:
    print('Confusion matrix, without normalization')

  plt.imshow(cm, interpolation='nearest', cmap=cmap)
  plt.title(title)
  plt.colorbar()
  tick_marks = np.arange(len(classes))
  plt.xticks(tick_marks, classes, rotation=55)
  plt.yticks(tick_marks, classes)
  fmt = '.2f' if normalize else 'd'
  thresh = cm.max() / 2.
  for i, j in itertools.product(range(cm.shape[0]), range(cm.shape[1])):
    plt.text(j, i, format(cm[i, j], fmt),
              horizontalalignment="center",
              color="white" if cm[i, j] > thresh else "black")

  plt.ylabel('True label')
  plt.xlabel('Predicted label')
  plt.tight_layout()

# Classify pose in the TEST dataset using the trained model
y_pred = model.predict(X_test)

# Convert the prediction result to class name
y_pred_label = [class_names[i] for i in np.argmax(y_pred, axis=1)]
y_true_label = [class_names[i] for i in np.argmax(y_test, axis=1)]

# Plot the confusion matrix
cm = confusion_matrix(np.argmax(y_test, axis=1), np.argmax(y_pred, axis=1))
plot_confusion_matrix(cm,
                      class_names,
                      title ='Confusion Matrix of Pose Classification Model')

# Print the classification report
print('\nClassification Report:\n', classification_report(y_true_label,
                                                          y_pred_label))

(任意) 予測ミスの調査

誤って予測された TEST データセットの姿勢を調査し、モデル精度を改善できるかどうかを確認できます。

注意: この方法では、ローカルコンピュータの姿勢画像ファイルを表示する必要があるため、手順 1 を実行した場合にのみ動作します。

if is_skip_step_1:
  raise RuntimeError('You must have run step 1 to run this cell.')

# If step 1 was skipped, skip this step.
IMAGE_PER_ROW = 3
MAX_NO_OF_IMAGE_TO_PLOT = 30

# Extract the list of incorrectly predicted poses
false_predict = [id_in_df for id_in_df in range(len(y_test)) \
                if y_pred_label[id_in_df] != y_true_label[id_in_df]]
if len(false_predict) > MAX_NO_OF_IMAGE_TO_PLOT:
  false_predict = false_predict[:MAX_NO_OF_IMAGE_TO_PLOT]

# Plot the incorrectly predicted images
row_count = len(false_predict) // IMAGE_PER_ROW + 1
fig = plt.figure(figsize=(10 * IMAGE_PER_ROW, 10 * row_count))
for i, id_in_df in enumerate(false_predict):
  ax = fig.add_subplot(row_count, IMAGE_PER_ROW, i + 1)
  image_path = os.path.join(images_out_test_folder,
                            df_test.iloc[id_in_df]['file_name'])

  image = cv2.imread(image_path)
  plt.title("Predict: %s; Actual: %s"
            % (y_pred_label[id_in_df], y_true_label[id_in_df]))
  plt.imshow(cv2.cvtColor(image, cv2.COLOR_BGR2RGB))
plt.show()

第 3 部: 姿勢分類モデルを TFLite に変換する

Keras 姿勢分類モデルを TensorFlow Lite 形式に変換し、モバイルアプリ、Web ブラウザ、エッジデバイスに展開できるようにします。モデルを変換するときには、ダイナミックレンジ量子化を適用し、大きい精度損失がない状態で、姿勢分類 TensorFlow Lite モデルサイズを約 1/4 まで小さくします。

注意: TensorFlow Lite は複数の量子化スキームをサポートしています。詳細については、ドキュメントを参照してください。

converter = tf.lite.TFLiteConverter.from_keras_model(model)
converter.optimizations = [tf.lite.Optimize.DEFAULT]
tflite_model = converter.convert()

print('Model size: %dKB' % (len(tflite_model) / 1024))

with open('pose_classifier.tflite', 'wb') as f:
  f.write(tflite_model)

クラスインデックスと人間が読み取れるクラス名とのマッピングを含むラベルファイルを作成します。

with open('pose_labels.txt', 'w') as f:
  f.write('\n'.join(class_names))

量子化を適用してモデルサイズを小さくしたので、量子化された TFLite モデルを評価して、精度の低下が許容可能かどうかを確認します。

def evaluate_model(interpreter, X, y_true):
  """Evaluates the given TFLite model and return its accuracy."""
  input_index = interpreter.get_input_details()[0]["index"]
  output_index = interpreter.get_output_details()[0]["index"]

  # Run predictions on all given poses.
  y_pred = []
  for i in range(len(y_true)):
    # Pre-processing: add batch dimension and convert to float32 to match with
    # the model's input data format.
    test_image = X[i: i + 1].astype('float32')
    interpreter.set_tensor(input_index, test_image)

    # Run inference.
    interpreter.invoke()

    # Post-processing: remove batch dimension and find the class with highest
    # probability.
    output = interpreter.tensor(output_index)
    predicted_label = np.argmax(output()[0])
    y_pred.append(predicted_label)

  # Compare prediction results with ground truth labels to calculate accuracy.
  y_pred = keras.utils.to_categorical(y_pred)
  return accuracy_score(y_true, y_pred)

# Evaluate the accuracy of the converted TFLite model
classifier_interpreter = tf.lite.Interpreter(model_content=tflite_model)
classifier_interpreter.allocate_tensors()
print('Accuracy of TFLite model: %s' %
      evaluate_model(classifier_interpreter, X_test, y_test))

TFLite モデル (pose_classifier.tflite) とラベルファイル (pose_labels.txt) をダウンロードして、カスタム姿勢を分類します。TFLite 姿勢分類モデルの使用方法に関するエンドツーエンドの例については、Android および Python/Raspberry Pi サンプルアプリを確認してください。

zip pose_classifier.zip pose_labels.txt pose_classifier.tflite

# Download the zip archive if running on Colab.
try:
  from google.colab import files
  files.download('pose_classifier.zip')
except:
  pass