目次
概要
作成したもの
- 動画の中で自衛隊体操の統制運動をしている場面を検知するアルゴリズム。
使用した技術・ライブラリ
- OpenPose
- LSTM
- OpenCV
- Pillow
- pydub
動機
画像の類似度を測るコードは、動画の類似度を測るアルゴリズムがあまり公開されていなかったのと、自衛隊運動を世間に認知してもらうことが目的で今回コードを書きました。
自衛隊体操とは
自衛官の基礎体力の向上を目的とした体操であり、２１種類の動的運動と静的運動を音楽に合わせて行います。ハードな体操である為、筋を痛めたり怪我をする危険性があるので、自衛隊体操のための準備運動が必要です。。。くれぐれも自衛隊体操を実施する際は怪我に気をつけてください。
OpenPoseとは
OpenPoseはカーネギーメロン大学（CMU）の Zhe Caoら が「Realtime Multi-Person pose estimation」の論文で発表した、深層学習を用いて人物のポーズを可視化してくれる手法です。特別なセンサーや深度計などが不要で簡単にCPU環境でも実装することができる便利なライブラリです。
アーキテクチャ
処理の流れ
- 動画を画像に分解
- フレームごとに関節の座標データを取得
- 動画間の類似度を判定
- 類似度が一定値以上の範囲を自衛隊体操であると特定
教師用動画（自衛隊体操）を用意して、フレームごとのOpenPoseで関節を取得して、LSTMで特徴量を抽出してDBに格納します。類似度を検証したい動画も同様にフレームごとの関節データを計測、特徴抽出して、DBに格納した教師用動画と比較します。
アーキ図
実装
Python3.6系で実装していきます。OpenPoseやOpenCVなどのライブラリ関連のインストールは省略します。
フレームごとの座標データの取得
OpenCVで動画を１フレームずつ読み込みこんで、OpenPoseで関節データを保存していきます。
# ビデオの読み込み
videos = glob.glob("/data/jieitai-taiso/*.mp4")
for video_path in videos:
cap,count,fps = load_video(video_path)
print(f"filename:{video_path},count:{count}, fps:{fps}")
# モデルの読み込み
model = load_model()
frames = []
for num in range(0, int(count)):
try:
# 動画からフレームを取り出し
cap.set(cv2.CAP_PROP_POS_FRAMES, num)
image = cap.read()[1]
if image is None:
count = num -1
break
humans,best_human,pos = get_pose_data(image,model)
# 関節の描写
# pone_image = describe_parts(image,[best_human])
# plt.imshow(cv2.cvtColor(pone_image, cv2.COLOR_BGR2RGB))
frame = {'num':num,'pos':pos}
frames.append(frame)
except:
frame = {'num':num,'pos':{}}
cap.release()
video_data = {'file_name':video_path,'count':count,'fps':fps,'frames':frames}
フレームに２名以上の人物を検知した場合、1名の関節のみ取得します。
def get_best_human(humans):
best_human_idx = 0
best_human_size = 0
if len(humans) > 1:
for idx, human in enumerate(humans):
d = 0
if 0 not in human.body_parts.keys():
continue
for i in human.body_parts.keys():
if i == 0:
continue
d += abs(human.body_parts[i].x - human.body_parts[0].x)\
+ abs(human.body_parts[i].y - human.body_parts[0].y)
if d > best_human_size:
best_human_idx = idx
best_human_size = d
elif len(humans) == 0:
return None
best_human = humans[best_human_idx]
return best_human
def pos_list(human):
human_pos = {}
for i in human.body_parts.keys():
human_pos[str(i)] = (human.body_parts[i].x , human.body_parts[i].y)
return human_pos
def get_pose_data(image,model):
# 推論
humans = model.inference(image, resize_to_default=True, upsample_size=4.0)
if len(humans) != 0:
# 人物特定
best_human = get_best_human(humans)
# 座標取得
pos = pos_list(best_human)
else:
human = None
pos = {}
return humans,best_human,pos
動画の類似度を判定
２つの動画の類似度をフレームごとに比較していきます。類似度の閾値を上回るフレームをマークします。
マークが集中した箇所が検知した場所になります。
def calc_sim(best_human_pos,target):
d = 0
for i in best_human_pos.keys() & target.keys():
d += (abs(target[str(i)][0] - best_human_pos[str(i)][0]) \
+ abs(target[str(i)][1] - best_human_pos[str(i)][1]))
d += len(set(best_human_pos.keys()).symmetric_difference(target.keys()))
return d/18
field1 = collection.find_one({"file_name" : "/data/jieitai.mp4"})
field2 = collection.find_one({"file_name" : "/data/jieitai2.mp4"})
video_data_1 = field1['frames']
video_data_2 = field2['frames']
from_ = 0
fig = plt.figure(figsize=(20, 20))
ax = fig.add_subplot(4,1,1)
ax2 = fig.add_subplot(4,1,2)
ax3 = fig.add_subplot(4,1,3)
ax.set_title('video_1')
ax.set_xlabel('time')
ax.set_ylabel('score')
ax2.set_title('video_2')
ax2.set_xlabel('video_2 time')
ax2.set_ylabel('video_1 time')
ax3.set_xlabel('video_1 time')
ax3.set_ylabel('video_2 time')
pixels = []
thureshhold = 1.0
for v1 in video_data_1:
high_score_1 = 1
target_1 = None
for v2 in video_data_2:
score = calc_sim(v2['pos'],v1['pos'])
if high_score_1 >= score:
high_score_1 = score
target_1 = v2
if high_score_1 > thureshhold :
continue
video1_time = v1['num']/field1['fps']
video2_time = target_1['num']/field2['fps']
ax.scatter(video1_time,-high_score_1,c='red')
ax2.scatter(video1_time,video2_time,c='red')
ax3.scatter(video2_time,video1_time,c='red')
pixels.append(target_1['num'])
SEC = 3
video2_time_length = field2['count']/field2['fps']
bins = np.linspace(0, int(video2_time_length), int(video2_time_length/SEC))
index = np.digitize(np.array(pixels)/field2['fps'],bins)
c = Counter(index)
l = len(list(set(index.tolist())))
mode = c.most_common()
for i in range(l):
most = mode[i][0]
print(datetime.timedelta(seconds=int((most-1)*SEC)),'-',datetime.timedelta(seconds=SEC+int((most-1)*SEC)))
plt.hist(np.array(pixels)/field2['fps'],range(0,int(video2_time_length)))# ヒストグラムの出力
plt.show()
結果
本家の自衛隊体操動画の統制運動部分を切り取って、教師用動画として使用。
自撮りで自衛隊体操をやっている動画を撮影。
検証用動画で統制運動をしている場面を推測できるかどうかを検証しました。
教師用動画（2:57~3:10）
2:57~3:10部分の統制運動を切り取って学習させました。
検証用動画
体の前後屈 → 体の前屈前倒 → 体の斜前屈 → その場跳躍 → 統制運動 の順で１動作5秒ずつ25秒の検証動画を撮影。
統制運動の類似度が高い場面に効果音と字幕を入れるアルゴリズムを追加しています。
うまい具合に統制運動部分に自動でエフェクトが入っているのがわかります。
類似度グラフ
散布図とヒストグラムで類似度が高く出ている場所をプロット。検証用動画の20~25秒付近に類似度が集中しています。
統制運動のポーズ
やっぱりこのポーズが特徴的なので類似度が高く出るようです。
おまけ
効果音と字幕を挿入するコード。
pydubとpillowを使っています。
output_file = "/data/jieitai.mp4"
video_path = "/data/jieitai-taiso/jieitai-trim.mp4"
output_video = "/data/edited.mp4"
output_sound = "/data/sound.mp4"
effect = '/data/jieitai-taiso/統制運動.m4a'
title = '統制運動'
font_size_rate = 0.0002
font_path = '/System/Library/Fonts/AppleGothic.ttf'
def add_subtitles(input_path,output_path,titles):
'''
字幕を追加
'''
cap = cv2.VideoCapture(input_path)
fps = cap.get(cv2.CAP_PROP_FPS)
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
fourcc = cv2.VideoWriter_fourcc('m', 'p', '4', 'v')
out = cv2.VideoWriter(output_path, fourcc, fps, (width, height))
font_size = int(font_size_rate * width * height)
font_x = width/2-font_size*2
font_y = height-font_size
num = 0
while True:
ret, frame = cap.read()
if ret:
for t in titles:
if fps*t['from_sec']<num<fps*(t['to_sec']+t['from_sec']):
# 枠の描写
frame = cv2.rectangle(frame, (0, 0), (width, height), (0, 0, 255,), 50)
# 字幕の描写
frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
pil_image = Image.fromarray(frame)
draw = ImageDraw.Draw(pil_image)
font = ImageFont.truetype(font_path, font_size)
draw.text((font_x, font_y), t['title'], font=font)
frame = cv2.cvtColor(np.array(pil_image), cv2.COLOR_RGB2BGR)
break
out.write(frame)
num += 1
else:
break
cap.release()
out.release()
def split_sound(video_path,output_sound):
'''
動画と音声を分離
'''
cmd = "ffmpeg -y -i {} -vn -acodec copy {}".format(video_path,output_sound)
resp = subprocess.check_output(cmd, shell=True)
def add_effects(sound,output_sound,effect,from_sec):
'''
効果音を追加
'''
base_sound = AudioSegment.from_file(sound, format="mp4") # 音声を読み込み
effect_sound = AudioSegment.from_file(effect, format="m4a") # 効果音を読み込み
start_time_ms = from_sec * 1000 # 効果音をX秒時点から鳴らす
result_sound = base_sound.overlay(effect_sound, start_time_ms)
result_sound .export(output_sound, format="mp4") # 保存する
def mix_video_sound(video,sound,output_file):
'''
動画と音声を結合
'''
cmd = f"ffmpeg -y -i {video} -i {sound} -c:v copy -c:a aac {output_file}"
resp = subprocess.check_output(cmd, shell=True)
結論・感想
人物の動きで動画の類似度を測定し、動画の類似度＋秒数を特定するプログラムを書くことができました。
今回は統制運動でしか試してないですが、他の動きでも同様に検知できるか試してみたいと思います。
参考
こちらの記事を元に類似度のロジックを作成しました。
