Motion tracking II: 2D pose estimation via OpenPose

Overview

In this script, we take the videos prepared in the previous script and perform OpenPose 2D pose estimation (Cao et al., 2019) on them.

We are using the BODY_135 model with 135 keypoints.

Demo of this pipeline has been published on EnvisionBOX

See OpenPose documentation for more information.

Code to prepare the environment

import os
import subprocess
import glob
import tempfile
from IPython.display import Video
import random

curfolder = os.getcwd()

# Openpose demo.exe location
openposefol = os.path.join(curfolder, 'openpose')
openpose_demo_loc = os.path.join(openposefol, 'bin', 'OpenPoseDemo.exe')

# This is the model to employ
model_to_employ = 'BODY_135'

# List folders in a main folder
folderstotrack = glob.glob(os.path.join(curfolder, 'projectdata', '*'))

# Get all folders per participant, per session
pcnfolders = []

for i in folderstotrack:
    pcnfolders_in_session = glob.glob(os.path.join(i, '*'))
    pcnfolders = pcnfolders + pcnfolders_in_session

# There might be some other things we don't want now
pcnfolders = [x for x in pcnfolders if 'Config' not in x]
pcnfolders = [x for x in pcnfolders if 'opensim' not in x]
pcnfolders = [x for x in pcnfolders if 'xml' not in x]
pcnfolders = [x for x in pcnfolders if 'ResultsInverseDynamics' not in x]
pcnfolders = [x for x in pcnfolders if 'ResultsInverseKinematics' not in x]
pcnfolders = [x for x in pcnfolders if 'sto' not in x]
pcnfolders = [x for x in pcnfolders if 'txt' not in x]
pcnfolders = [x for x in pcnfolders if 'calibration' not in x]

print(pcnfolders[0:10])

['f:\\FLESH_ContinuousBodilyEffort\\02_MotionTracking_processing\\projectdata\\Session_10_1\\10_1_11_p1', 'f:\\FLESH_ContinuousBodilyEffort\\02_MotionTracking_processing\\projectdata\\Session_10_1\\10_1_12_p1', 'f:\\FLESH_ContinuousBodilyEffort\\02_MotionTracking_processing\\projectdata\\Session_10_1\\10_1_13_p1', 'f:\\FLESH_ContinuousBodilyEffort\\02_MotionTracking_processing\\projectdata\\Session_10_1\\10_1_14_p1', 'f:\\FLESH_ContinuousBodilyEffort\\02_MotionTracking_processing\\projectdata\\Session_10_1\\10_1_15_p1', 'f:\\FLESH_ContinuousBodilyEffort\\02_MotionTracking_processing\\projectdata\\Session_10_1\\10_1_16_p1', 'f:\\FLESH_ContinuousBodilyEffort\\02_MotionTracking_processing\\projectdata\\Session_10_1\\10_1_17_p1', 'f:\\FLESH_ContinuousBodilyEffort\\02_MotionTracking_processing\\projectdata\\Session_10_1\\10_1_20_p0', 'f:\\FLESH_ContinuousBodilyEffort\\02_MotionTracking_processing\\projectdata\\Session_10_1\\10_1_21_p0', 'f:\\FLESH_ContinuousBodilyEffort\\02_MotionTracking_processing\\projectdata\\Session_10_1\\10_1_22_p0']

Running OpenPose as a subprocess

Instead of running OpenPose in the notebook, we will send commands to the OpenPose executable as a subprocess. This is more efficient and allows us to run OpenPose in parallel for multiple videos.

Code to run OpenPose as a subprocess

def runcommand(command):
    '''
    Run a command in the command line.
    '''
    import subprocess
    # Run the command using subprocess for OPENPOSE TRACKING
    try:
        subprocess.run(command, shell=True, check=True)
    except subprocess.CalledProcessError as e:
        print(f"Command execution failed with error code {e.returncode}")
    except FileNotFoundError:
        print("The OpenPoseDemo.exe executable was not found.")

for i in pcnfolders:
    os.chdir(openposefol)
    
    print('working on ' + i)

    # Identify all avi files in folder
    direc = glob.glob(os.path.join(i, 'raw-2d', '*.avi'))

    # 3 cameras
    video0 = direc[0]
    video1 = direc[1]
    video2 = direc[2]

    videolist = [video0, video1, video2]
    
    # Make a new directory if it doesn't exist
    if not os.path.exists(os.path.join(i, 'pose')):
        os.makedirs(os.path.join(i, 'pose'))
    if not os.path.exists(os.path.join(i, 'pose', 'pose_cam1_json')):
        os.makedirs(os.path.join(i, 'pose', 'pose_cam1_json'))
    if not os.path.exists(os.path.join(i, 'pose', 'pose_cam2_json')):
        os.makedirs(os.path.join(i, 'pose', 'pose_cam2_json'))
    if not os.path.exists(os.path.join(i, 'pose', 'pose_cam3_json')):
        os.makedirs(os.path.join(i, 'pose', 'pose_cam3_json'))
    

    # Also make directory for openpose videos (pose-2d-trackingvideos)
    if not os.path.exists(os.path.join(i, 'pose-2d-trackingvideos')):
        os.makedirs(os.path.join(i, 'pose-2d-trackingvideos'))

    # if in trackingvideos are three videos, skip
    existingtrackingvideos = glob.glob(os.path.join(i, 'pose-2d-trackingvideos', '*.avi'))
    if len(existingtrackingvideos) == 3:
        print('Skipping ' + i + ' as tracking already exist')
        continue

    # Initialize the pose2 folder
    outputfol1 = os.path.join(i, 'pose', 'pose_cam1_json')
    outputfol2 = os.path.join(i, 'pose', 'pose_cam2_json')
    outputfol3 = os.path.join(i, 'pose', 'pose_cam3_json')

    outputfollist = [outputfol1, outputfol2, outputfol3]

    for it, j in enumerate(outputfollist):
        # Prepare the command
        openposelocation = ' ' + openpose_demo_loc + ' '
        model = '--model_pose' + ' ' + model_to_employ + ' '
        video = '--video ' + videolist[it] + ' '
        todo = '--write_json '
        outputfol = j + ' '
        videoadd = '--write_video '
        videopath = os.path.join(i, 'pose-2d-trackingvideos', 'video'+str(it)+'.avi') + ' '
        # Send the command via subprocess
        command = r' '+openposelocation+model+video+todo+outputfol+videoadd+videopath
        print('were going to send this to command prompt: ' + command)
        runcommand(command)

Now we check that all trials have been processed by checking the number of output videos (original videos with overlayed skeleton)

pcnfolders = []

for i in folderstotrack:
    pcnfolders_in_session = glob.glob(os.path.join(i, '*'))

    # Append to the list
    pcnfolders = pcnfolders + pcnfolders_in_session

# There might be some other things we don't want now
pcnfolders = [x for x in pcnfolders if 'Config' not in x]
pcnfolders = [x for x in pcnfolders if 'opensim' not in x]
pcnfolders = [x for x in pcnfolders if 'xml' not in x]
pcnfolders = [x for x in pcnfolders if 'ResultsInverseDynamics' not in x]
pcnfolders = [x for x in pcnfolders if 'ResultsInverseKinematics' not in x]
pcnfolders = [x for x in pcnfolders if 'sto' not in x]
pcnfolders = [x for x in pcnfolders if 'txt' not in x]
pcnfolders = [x for x in pcnfolders if 'calibration' not in x]

for i in pcnfolders:
    existingtrackingvideos = glob.glob(os.path.join(i, 'pose-2d-trackingvideos', '*.avi'))
    if len(existingtrackingvideos) != 3:
        print('Warning: ' + i + ' has ' + str(len(existingtrackingvideos)) + ' tracking videos.')

The results are saved in /pose folder in json format, and are used for triangulation via Pose2sim (Pagnon et al., 2022) in the next script.

The videos with the estimated poses are saved in the /pose-2d-trackingvideos folder

Here is an example of the output:

References

Cao, Z., Hidalgo, G., Simon, T., Wei, S.-E., & Sheikh, Y. (2019). OpenPose: Realtime Multi-Person 2D Pose Estimation using Part Affinity Fields. https://doi.org/10.48550/arXiv.1812.08008

Pagnon, D., Domalain, M., & Reveret, L. (2022). Pose2Sim: An End-to-End Workflow for 3D Markerless Sports Kinematics—Part 2: Accuracy. Sensors, 22(7, 7), 2712. https://doi.org/10.3390/s22072712