概要

外出自粛や在宅勤務による運動不足も問題としてあげられて、スクワットなど自宅での手軽な運動が流行しています。トレーニングに集中してどこまでカウントしたかを忘れたことがよくあります。これらを背景にAI姿勢推定技術を用いてスクワットのトレーニング動画から反復運動を自動的にカウントしてみました。

今回やること

AI姿勢推定エンジンAnyMotionを用いてスクワット動画から人間の骨格位置情報を推定
位置情報から姿勢の特徴量を算出
k-NNを用いてスクアウトの「しゃがむ」と「立つ」二種類の姿勢を分類
推定結果からスクアウトを計測

事前準備

フォルダー構成

datasetフォルダーに今回利用するデータを格納します

|-- dataset
|   |-- down
|   |   |-- down_(0).jpg
|   |   |-- down_(1).png
|   |   |-- ...
|   |   `-- down_(31).jpg
|   |-- down.json
|   |-- up
|   |   |-- up_(0).png
|   |   |-- ...
|   |   `-- up_(31).png
|   |-- up.json
|   |-- squat.mp4
|   `-- squat.json
|-- main.ipynb
`-- src
    |-- classifier.py　
    |-- counter.py
    |-- embedder.py
    |-- keypoint.py
    |-- sample.py

データを準備

スクワットの計測するため、動画のフレームごとに「しゃがむ」と「立つ」二種類の姿勢を分類する必要があります。

分類器を構築するため二種類の姿勢のサンプル画像を準備します。
- フォルダーdownに「しゃがむ」の画像データを格納します。
- down.json は「しゃがむ」の骨格位置推定情報を記載します。
- フォルダーupに「立つ」の画像データを格納します。
- up.jsonは「立つ」の骨格位置推定情報を記載します。
検証用のスクワット動画を準備します。
　- squat.mp4 はスクワット動画です。
　- squat.json はスクワット動画の各フレームの骨格位置推定情報を記載します。

骨格位置推定情報

AnyMotion SDK を利用してサンプル画像データ、スクワット動画を姿勢推定し、骨格位置情報をjsonデータに保存します。

import anymotion_sdk

# 初期化
client = anymotion_sdk.Client(client_id="your_client_id", client_secret="your_client_secret")

# 動画をアップロード
upload_result = client.upload("squat.mp4")

# 骨格位置情報を推定する
keypoint_id = client.extract_keypoint(movie_id=upload_result.movie_id)

# 骨格位置情報を取得する
extraction_result = client.wait_for_extraction(keypoint_id)
keypoint = extraction_result.json

データフォマット

AnyMotionから取得した骨格位置推定情報とファイルパスを次のフォーマットでjsonファイルに保存します。

[
	...
    {
        "path": "./data/down/down_(2).png",
        "keypoint": [
            [68, 35],
            [71, 32],
            ...
            [32, 156],
        ],
    },
    ...
]

実装

骨格位置情報の格納

上記のjsonファイルを読み取り、Keypointクラスに情報を保持します。

ファイルのパス
分類の名前
骨格位置情報
骨格位置情報から計算された特徴量

from __future__ import annotations

import numpy as np
import numpy.typing as npt


class Keypoint:
    def __init__(self, coordinates: list[list]) -> None:
        self.names = [
            "nose",
            "left_eye",
            "right_eye",
            "left_ear",
            "right_ear",
            "left_shoulder",
            "right_shoulder",
            "left_elbow",
            "right_elbow",
            "left_wrist",
            "right_wrist",
            "left_hip",
            "right_hip",
            "left_knee",
            "right_knee",
            "left_ankle",
            "right_ankle",
        ]
        self._coordinates = np.copy(coordinates)
        self.class_name = ("",)
        self.embedding = (None,)

    @property
    def coordinates(self) -> npt.NDArray:
        return self._coordinates

    def __add__(self, other: npt.NDArray) -> Keypoint:
        return Keypoint(self._coordinates + other)

    def __sub__(self, other: npt.NDArray) -> Keypoint:
        return Keypoint(self._coordinates - other)

    def __truediv__(self, other: npt.NDArray) -> Keypoint:
        return Keypoint(self._coordinates / other)

    def __mul__(self, other: npt.NDArray) -> Keypoint:
        return Keypoint(self._coordinates * other)

    def get_coordinate_by_name(self, name: str) -> npt.NDArray:
        return self._coordinates[self.names.index(name)]

    # 二つの点の中心座標を返す
    def get_midpoint_by_names(self, name_from: str, name_to: str) -> npt.NDArray:
        point_from = self.get_coordinate_by_name(name_from)
        point_to = self.get_coordinate_by_name(name_to)
        return (point_from + point_to) * 0.5

    # 二つの点の距離を返す
    def get_distance_by_names(self, name_from: str, name_to: str) -> npt.NDArray:
        point_from = self.get_coordinate_by_name(name_from)
        point_to = self.get_coordinate_by_name(name_to)
        return self.get_distance(point_from, point_to)

    def get_distance(
        self, point_from: npt.NDArray, point_to: npt.NDArray
    ) -> npt.NDArray:
        return point_to - point_from

特徴量算出

骨格位置情報から特徴量ベクトルを算出します。

正規化（Normalization）
- 胴体の長さと各点から中心の距離の大きいほうを正規化のスケールします
- 原点をヒップの中心にして、正規化を行います
特徴量ベクトル
- 間接点間の位置情報からユークリッド距離を計算して特徴量空間に格納します
- 今回は23種類の特徴量を利用します

from __future__ import annotations

import numpy as np
import numpy.typing as npt

from src.keypoint import Keypoint


class KeypointEmbedder:
    def __init__(self) -> None:
        self.multiplier = 2.5

    def __call__(self, keypoint: Keypoint) -> npt.NDArray:
        keypoint = self._normalize(keypoint)
        embedding = self._get_distance_embedding(keypoint)
        return embedding

    # 原点をヒップの中心にして、正規化する
    def _normalize(self, keypoint: Keypoint) -> Keypoint:
        keypoint_center = keypoint.get_midpoint_by_names("left_hip", "right_hip")
        keypoint = keypoint - keypoint_center

        scale = self._get_scale(keypoint)
        keypoint = keypoint / scale
        return keypoint

    # 胴体の長さと各点から中心の距離の大きいほうを正規化スケールにする
    def _get_scale(self, keypoint: Keypoint) -> float:
        hip_center = keypoint.get_midpoint_by_names("left_hip", "right_hip")
        shoulder_center = keypoint.get_midpoint_by_names(
            "left_shoulder", "right_shoulder"
        )
        body_size = np.linalg.norm(shoulder_center - hip_center)

        keypoint_center = keypoint.get_midpoint_by_names("left_hip", "right_hip")
        max_dist = np.max(
            np.linalg.norm(keypoint.coordinates - keypoint_center, axis=1)
        )

        return max(body_size * self.multiplier, max_dist)

    # 特徴量の計算
    def _get_distance_embedding(self, keypoint: Keypoint) -> npt.NDArray:
        embedding = np.array(
            [
                keypoint.get_distance(
                    keypoint.get_midpoint_by_names("left_hip", "right_hip"),
                    keypoint.get_midpoint_by_names("left_shoulder", "right_shoulder"),
                ),
                keypoint.get_distance_by_names("left_shoulder", "left_elbow"),
                keypoint.get_distance_by_names("right_shoulder", "right_elbow"),
                keypoint.get_distance_by_names("left_elbow", "left_wrist"),
                keypoint.get_distance_by_names("right_elbow", "right_wrist"),
                keypoint.get_distance_by_names("left_hip", "left_knee"),
                keypoint.get_distance_by_names("right_hip", "right_knee"),
                keypoint.get_distance_by_names("left_knee", "left_ankle"),
                keypoint.get_distance_by_names("right_knee", "right_ankle"),
                keypoint.get_distance_by_names("left_shoulder", "left_wrist"),
                keypoint.get_distance_by_names("right_shoulder", "right_wrist"),
                keypoint.get_distance_by_names("left_hip", "left_ankle"),
                keypoint.get_distance_by_names("right_hip", "right_ankle"),
                keypoint.get_distance_by_names("left_hip", "left_wrist"),
                keypoint.get_distance_by_names("right_hip", "right_wrist"),
                keypoint.get_distance_by_names("left_shoulder", "left_ankle"),
                keypoint.get_distance_by_names("right_shoulder", "right_ankle"),
                keypoint.get_distance_by_names("left_hip", "left_wrist"),
                keypoint.get_distance_by_names("right_hip", "right_wrist"),
                keypoint.get_distance_by_names("left_elbow", "right_elbow"),
                keypoint.get_distance_by_names("left_knee", "right_knee"),
                keypoint.get_distance_by_names("left_wrist", "right_wrist"),
                keypoint.get_distance_by_names("left_ankle", "right_ankle"),
            ]
        )

        return embedding

k-NNによる分類器の構築

サンプルデータの特徴量ベクトルからk-NN分類器を構築します。

k-NNによる分類
1. 対象動画フレームの位置値情報(Keypoint)を入力します
2. 対象フレームと各サンプル、特徴量空間における最大距離で並び替えて小さい30個サンプルを利用します
3. 対象フレームと残した30個サンプル、特徴量空間における平均距離で並び替えて小さい10個サンプルを残します
4. 残した10個サンプル中、「しゃがむ」と「立つ」の数を計測、数の多いほうに分類します
5. {'up': 2, 'down': 8} のような結果がアウトプットされて、'up': 2は「立つ」の数が２個を表し、'down': 8は「しゃがむ」の数が２個を表します
6. 数が多いほうの分類を対象フレームの分類として推定します

from __future__ import annotations

import numpy as np

from src.embedder import KeypointEmbedder
from src.keypoint import Keypoint


class KNN:
    def __init__(self, data: dict, embedder: KeypointEmbedder) -> None:
        self._embedder = embedder
        self.keypoint_samples = self._load_keypoint_samples(data, embedder)

    # サンプルデータの特徴量ベクトルを格納
    def _load_keypoint_samples(
        self, data: dict, embedder: KeypointEmbedder
    ) -> list[Keypoint]:
        keypoint_samples = []
        for class_name in data.keys():
            for item in data[class_name]:
                keypoint = Keypoint(item["keypoint"])
                keypoint.class_name = class_name
                keypoint.embedding = embedder(keypoint)
                keypoint_samples.append(keypoint)
        return keypoint_samples

    # k-NNによる分類を行う
    def __call__(self, keypoint: Keypoint) -> dict:
        # 対象データの特徴量ベクトル、ひっくり返った特徴量ベクトルを計算
        embedding = self._embedder(keypoint)
        flipped_embedding = self._embedder(keypoint * np.array([-1, 1]))

        # 対象データと各サンプル、特徴量空間における最大距離で並び替えて近い30個サンプルを残す
        max_distances = []
        for idx, sample in enumerate(self.keypoint_samples):
            max_dist = min(
                np.max(np.abs(sample.embedding - embedding)),
                np.max(np.abs(sample.embedding - flipped_embedding)),
            )
            max_distances.append([max_dist, idx])

        max_distances = sorted(max_distances, key=lambda x: x[0])
        max_distances = max_distances[:30]
        mean_distances = []

        # 対象データと残したサンプル、特徴量空間における平均距離で並び替えて近い10個サンプルを残す
        for _, idx in max_distances:
            sample = self.keypoint_samples[idx]
            mean_dist = min(
                np.mean(np.abs(sample.embedding - embedding)),
                np.mean(np.abs(sample.embedding - flipped_embedding)),
            )
            mean_distances.append([mean_dist, idx])

        mean_distances = sorted(mean_distances, key=lambda x: x[0])
        mean_distances = mean_distances[:10]

        # 残した10個サンプル中、「しゃがむ」と「立つ」の数を計測、数の多いほうに分類する
        class_names = [
            self.keypoint_samples[idx].class_name for _, idx in mean_distances
        ]
        result = {
            class_name: class_names.count(class_name) for class_name in set(class_names)
        }

        return result

動画のスクアウトを計測

動画は全てのフレームから時系列アウトプットが生成されます。滑らかな曲線で動作の傾向を表すために指数平滑移動平均(EMA)を時系列アウトプットにかけます

from __future__ import annotations

import cv2
import matplotlib.pyplot as plt
import numpy as np

from src.classifier import KNN
from src.keypoint import Keypoint


# 指数平滑移動平均を利用して分類器のアウトプットを滑らかにする
class ExponentialMovingAverage:
    def __init__(self) -> None:
        self._window_size = 10
        self._alpha = 0.2
        self._data_in_window: list[dict] = []

    # numerator = D_n + (1 - α) * D_(n-1) + (1 - α)^2 * D_(n-2) ...
    # denominator = 1 + 1 - α + (1 - α)^2 ...
    # EMA = numerator / denominator
    def __call__(self, data: dict, class_names: set) -> dict:
        # 新しいデータを入れてwindow_size分だけ保持する
        self._data_in_window.insert(0, data)
        self._data_in_window = self._data_in_window[: self._window_size]

        averaged_data = dict()
        for name in class_names:
            factor = 1.0
            numerator = 0.0
            denominator = 0.0
            for data in self._data_in_window:
                value = data.get(name, 0.0)
                numerator += factor * value
                denominator += factor
                factor *= 1.0 - self._alpha

            averaged_data[name] = numerator / denominator

        return averaged_data

指数平滑移動平均の結果を用いてスクワットの回数を計測します。スクワット開始からスクワット終了までを1回とします。

スクワット開始閾値： 6
スクワット終了閾値： 4

class Counter:
    def __init__(self, keypoint_data: list[dict], classifier: KNN) -> None:
        self.ema = ExponentialMovingAverage()
        self.classification = [
            classifier(Keypoint(item["keypoint"])) for item in keypoint_data
        ]
        self.moving_average: list[dict] = []

    def __call__(self) -> int:
        repeats = 0
        squating = False
        start_th = 6
        stop_th = 4

        class_names = set(
            [key for data in self.classification for key, _ in data.items()]
        )

        for idx, item in enumerate(self.classification):
            ema_item = self.ema(item, class_names)
            self.moving_average.append(ema_item)

            if ema_item.get("down", 0.0) > start_th:
                squating = True
            if ema_item.get("down", 0.0) < stop_th and squating:
                squating = False
                repeats = repeats + 1

        return repeats

実験結果

実際の動画squat.mp4を使って実験してみました。


# サンプルデータを読み込み
from src.sample import Dataset
sample = Dataset(['./dataset/down.json','./dataset/up.json'])

# サンプルデータからデータ算出
from src.embedder import KeypointEmbedder
embedder = KeypointEmbedder()

# サンプルデータからKNN分類器を構築
from src.classifier import KNN
classifier = KNN(sample.data, embedder)

# テスト動画のスクアウトを計測
from src.counter import Counter
test = Dataset(['./dataset/video.json'])
counter = Counter(test.data['video'], classifier)
print("repeats =", counter())
#repeats = 3
counter.plot_result('result.png')

スクワット回数は３回
フレームごとに「しゃがむ」と「立つ」の数は次の図のようになりました
- 縦軸は指数平滑移動平均したアウトプット
- 横軸は動画フレーム数

おわりに

今回はスクワットのトレーニング動画から回数を計測してみました。
処理としては、次の2つのことを行いました。

AnyMotion APIを用いて動画の骨格位置情報を推定
k-NNを用いてスクワットの「しゃがむ」と「立つ」姿勢を2分類

参考サイト