fulyvv/pose2sim
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

implementing OpensimProcessing (#130)

Browse Source 
* implementing opensimProcessing

* edited in cooperating with opensimProcessing

* Update scaling2IK.py

* implementing opensimProcessing

* Update and rename scaling2IK.py to kinematics.py

* Add files via upload

* Add files via upload

* Update Config.toml

* Update Config.toml

* Update Config.toml

* Update Config.toml

* Update Config.toml

* Update Config.toml

* implementing opensim processing

* OpenSim part addition

* OpenSim Processing function signature adjustment

* code logic and layout adjustments

* opensimProcessing test enabled

* Update Pose2Sim.py

* Update Pose2Sim.py

* docstring for opensimProcessing updated

* saved folder name changed to opensim

* opensim processing tests enabled

* Deleted a repetitive line at opensim kinematics section
This commit is contained in:
peterlololsss 2024-09-18 07:04:52 +08:00 committed by GitHub
parent 7af1584e01
commit 215c8a6e6c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 675 additions and 145 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
Pose2Sim/Demo_Batch/Config.toml
View File

@ -186,6 +186,9 @@ static_trial = ['S00_P00_Participant/S00_P00_T00_StaticTrial']
# # At the Participant level, specify the name of the static trial folder name, e.g. ['S00_P00_T00_StaticTrial']; # # At the Participant level, specify the name of the static trial folder name, e.g. ['S00_P00_T00_StaticTrial'];
# # At the Session level, add participant subdirectory, e.g. ['S00_P00_Participant/S00_P00_T00_StaticTrial', 'S00_P01_Participant/S00_P00_T00_StaticTrial'] # # At the Session level, add participant subdirectory, e.g. ['S00_P00_Participant/S00_P00_T00_StaticTrial', 'S00_P01_Participant/S00_P00_T00_StaticTrial']
opensim_bin_path = 'C:\OpenSim 4.4\bin' opensim_bin_path = 'C:\OpenSim 4.4\bin'
use_augmentation = false # If using augmented measurements then set it true
load_trc_name = 'filtered' # 'default' or 'filtered', if use_augmentation = true, this line will be ignored instead using __LSTM.trc
IK_timeRange = [] #left empty to IK full range or eg.[0.5,1.0]

13
Pose2Sim/Demo_Batch/Trial_1/Config.toml
View File

@ -181,11 +181,14 @@
# [opensim] # [opensim]
# static_trial = ['S00_P00_Participant/S00_P00_T00_StaticTrial'] #static_trial = ['S00_P00_Participant/S00_P00_T00_StaticTrial']
# # # If this Config.toml file is at the Trial level, set to true or false (lowercase); # # If this Config.toml file is at the Trial level, set to true or false (lowercase);
# # # At the Participant level, specify the name of the static trial folder name, e.g. ['S00_P00_T00_StaticTrial']; # # At the Participant level, specify the name of the static trial folder name, e.g. ['S00_P00_T00_StaticTrial'];
# # # At the Session level, add participant subdirectory, e.g. ['S00_P00_Participant/S00_P00_T00_StaticTrial', 'S00_P01_Participant/S00_P00_T00_StaticTrial'] # # At the Session level, add participant subdirectory, e.g. ['S00_P00_Participant/S00_P00_T00_StaticTrial', 'S00_P01_Participant/S00_P00_T00_StaticTrial']
# opensim_bin_path = 'C:\OpenSim 4.4\bin' #opensim_bin_path = 'C:\OpenSim 4.4\bin'
#use_augmentation = false # If using augmented measurements then set it true
#load_trc_name = 'filtered' # 'default' or 'filtered', if use_augmentation = true, this line will be ignored instead using __LSTM.trc
#IK_timeRange = [] #left empty to IK full range or eg.[0.5,1.0]

17
Pose2Sim/Demo_Batch/Trial_2/Config.toml
View File

@ -19,8 +19,8 @@
[project] [project]
multi_person = true # true for trials with multiple participants. If false, only the main person in scene is analyzed (and it run much faster). multi_person = true # true for trials with multiple participants. If false, only the main person in scene is analyzed (and it run much faster).
participant_height = [1.72, 1.40] # m # float if single person, list of float if multi-person (same order as the Static trials) # Only used for marker augmentation participant_height = [1.72, 1.40, 1.86] # m # float if single person, list of float if multi-person (same order as the Static trials) # Only used for marker augmentation
participant_mass = [70.0, 63.5] # kg # Only used for marker augmentation and scaling participant_mass = [70.0, 63.5, 88.0] # kg # Only used for marker augmentation and scaling
# frame_rate = 'auto' # fps # int or 'auto'. If 'auto', finds from video (or defaults to 60 fps if you work with images) # frame_rate = 'auto' # fps # int or 'auto'. If 'auto', finds from video (or defaults to 60 fps if you work with images)
# frame_range = [] # For example [10,300], or [] for all frames. # frame_range = [] # For example [10,300], or [] for all frames.
@ -181,11 +181,14 @@ keypoints_to_consider = 'all' # 'all' if all points should be considered, for ex
# [opensim] # [opensim]
# static_trial = ['S00_P00_Participant/S00_P00_T00_StaticTrial'] #static_trial = ['S00_P00_Participant/S00_P00_T00_StaticTrial']
# # # If this Config.toml file is at the Trial level, set to true or false (lowercase); # # If this Config.toml file is at the Trial level, set to true or false (lowercase);
# # # At the Participant level, specify the name of the static trial folder name, e.g. ['S00_P00_T00_StaticTrial']; # # At the Participant level, specify the name of the static trial folder name, e.g. ['S00_P00_T00_StaticTrial'];
# # # At the Session level, add participant subdirectory, e.g. ['S00_P00_Participant/S00_P00_T00_StaticTrial', 'S00_P01_Participant/S00_P00_T00_StaticTrial'] # # At the Session level, add participant subdirectory, e.g. ['S00_P00_Participant/S00_P00_T00_StaticTrial', 'S00_P01_Participant/S00_P00_T00_StaticTrial']
# opensim_bin_path = 'C:\OpenSim 4.4\bin' #opensim_bin_path = 'C:\OpenSim 4.4\bin'
#use_augmentation = false # If using augmented measurements then set it true
#load_trc_name = 'filtered' # 'default' or 'filtered', if use_augmentation = true, this line will be ignored instead using __LSTM.trc
#IK_timeRange = [] #left empty to IK full range or eg.[0.5,1.0]

7
Pose2Sim/Demo_MultiPerson/Config.toml
View File

@ -19,8 +19,8 @@
[project] [project]
multi_person = true # true for trials with multiple participants. If false, only the main person in scene is analyzed (and it run much faster). multi_person = true # true for trials with multiple participants. If false, only the main person in scene is analyzed (and it run much faster).
participant_height = [1.72, 1.40] # m # float if single person, list of float if multi-person (same order as the Static trials) # Only used for marker augmentation participant_height = [1.72, 1.40, 1.90] # m # float if single person, list of float if multi-person (same order as the Static trials) # Only used for marker augmentation
participant_mass = [70.0, 63.5] # kg # Only used for marker augmentation and scaling participant_mass = [70.0, 63.5, 90.0] # kg # Only used for marker augmentation and scaling
frame_rate = 'auto' # fps # int or 'auto'. If 'auto', finds from video (or defaults to 60 fps if you work with images) frame_rate = 'auto' # fps # int or 'auto'. If 'auto', finds from video (or defaults to 60 fps if you work with images)
frame_range = [] # For example [10,300], or [] for all frames. frame_range = [] # For example [10,300], or [] for all frames.
@ -186,6 +186,9 @@ static_trial = ['S00_P00_Participant/S00_P00_T00_StaticTrial']
# # At the Participant level, specify the name of the static trial folder name, e.g. ['S00_P00_T00_StaticTrial']; # # At the Participant level, specify the name of the static trial folder name, e.g. ['S00_P00_T00_StaticTrial'];
# # At the Session level, add participant subdirectory, e.g. ['S00_P00_Participant/S00_P00_T00_StaticTrial', 'S00_P01_Participant/S00_P00_T00_StaticTrial'] # # At the Session level, add participant subdirectory, e.g. ['S00_P00_Participant/S00_P00_T00_StaticTrial', 'S00_P01_Participant/S00_P00_T00_StaticTrial']
opensim_bin_path = 'C:\OpenSim 4.4\bin' opensim_bin_path = 'C:\OpenSim 4.4\bin'
use_augmentation = false # If using augmented measurements then set it true
load_trc_name = 'filtered' # 'default' or 'filtered', if use_augmentation = true, this line will be ignored instead using __LSTM.trc
IK_timeRange = [] #left empty to IK full range or eg.[0.5,1.0]

9
Pose2Sim/Demo_SinglePerson/Config.toml
View File

@ -53,9 +53,9 @@ output_format = 'openpose' # 'openpose', 'mmpose', 'deeplabcut', 'none' or a lis
[synchronization] [synchronization]
display_sync_plots = true # true or false (lowercase) display_sync_plots = true # true or false (lowercase)
keypoints_to_consider = ['RWrist'] # 'all' if all points should be considered, for example if the participant did not perform any particicular sharp movement. In this case, the capture needs to be 5-10 seconds long at least keypoints_to_consider = ['RWrist'] # 'all' if all points should be considered, for example if the participant did not perform any particicular sharp movement. In this case, the capture needs to be 5-10 seconds long at least
# ['RWrist', 'RElbow'] list of keypoint names if you want to specify keypoints with a sharp vertical motion. # ['RWrist', 'RElbow'] list of keypoint names if you want to specify the keypoints to consider.
approx_time_maxspeed = 'auto' # 'auto' if you want to consider the whole capture (default, slower if long sequences) approx_time_maxspeed = 'auto' # 'auto' if you want to consider the whole capture (default, slower if long sequences)
# [10.0, 2.0, 8.0, 11.0] list of times (seconds) if you want to specify the approximate time of a clear vertical event for each camera # [10.0, 2.0, 8.0, 11.0] list of times in seconds, one value per camera if you want to specify the approximate time of a clear vertical event by one person standing alone in the scene
time_range_around_maxspeed = 2.0 # Search for best correlation in the range [approx_time_maxspeed - time_range_around_maxspeed, approx_time_maxspeed + time_range_around_maxspeed] time_range_around_maxspeed = 2.0 # Search for best correlation in the range [approx_time_maxspeed - time_range_around_maxspeed, approx_time_maxspeed + time_range_around_maxspeed]
likelihood_threshold = 0.4 # Keypoints whose likelihood is below likelihood_threshold are filtered out likelihood_threshold = 0.4 # Keypoints whose likelihood is below likelihood_threshold are filtered out
filter_cutoff = 6 # time series are smoothed to get coherent time-lagged correlation filter_cutoff = 6 # time series are smoothed to get coherent time-lagged correlation
@ -186,6 +186,11 @@ static_trial = ['S00_P00_Participant/S00_P00_T00_StaticTrial']
# # At the Participant level, specify the name of the static trial folder name, e.g. ['S00_P00_T00_StaticTrial']; # # At the Participant level, specify the name of the static trial folder name, e.g. ['S00_P00_T00_StaticTrial'];
# # At the Session level, add participant subdirectory, e.g. ['S00_P00_Participant/S00_P00_T00_StaticTrial', 'S00_P01_Participant/S00_P00_T00_StaticTrial'] # # At the Session level, add participant subdirectory, e.g. ['S00_P00_Participant/S00_P00_T00_StaticTrial', 'S00_P01_Participant/S00_P00_T00_StaticTrial']
opensim_bin_path = 'C:\OpenSim 4.4\bin' opensim_bin_path = 'C:\OpenSim 4.4\bin'
use_augmentation = false # If using augmented measurements then set it true
load_trc_name = 'filtered' # 'default' or 'filtered', if use_augmentation = true, this line will be ignored instead using __LSTM.trc
IK_timeRange = [] #left empty to IK full range or eg.[0.5,1.0]

110
Pose2Sim/Pose2Sim.py
View File

@ -39,6 +39,7 @@ Pose2Sim.personAssociation()
Pose2Sim.triangulation() Pose2Sim.triangulation()
Pose2Sim.filtering() Pose2Sim.filtering()
Pose2Sim.markerAugmentation() Pose2Sim.markerAugmentation()
Pose2Sim.opensimProcessing()
# Then run OpenSim (see Readme.md) # Then run OpenSim (see Readme.md)
''' '''
@ -157,6 +158,7 @@ def read_config_files(config):
return level, config_dicts return level, config_dicts
def calibration(config=None): def calibration(config=None):
''' '''
Cameras calibration from checkerboards or from qualisys files. Cameras calibration from checkerboards or from qualisys files.
@ -194,7 +196,6 @@ def calibration(config=None):
end = time.time() end = time.time()
logging.info(f'\nCalibration took {end-start:.2f} s.\n') logging.info(f'\nCalibration took {end-start:.2f} s.\n')
logging.shutdown()
def poseEstimation(config=None): def poseEstimation(config=None):
@ -241,8 +242,6 @@ def poseEstimation(config=None):
elapsed = end - start elapsed = end - start
logging.info(f'\nPose estimation took {time.strftime("%Hh%Mm%Ss", time.gmtime(elapsed))}.\n') logging.info(f'\nPose estimation took {time.strftime("%Hh%Mm%Ss", time.gmtime(elapsed))}.\n')
logging.shutdown()
def synchronization(config=None): def synchronization(config=None):
''' '''
@ -287,8 +286,6 @@ def synchronization(config=None):
elapsed = end-start elapsed = end-start
logging.info(f'\nSynchronization took {time.strftime("%Hh%Mm%Ss", time.gmtime(elapsed))}.\n') logging.info(f'\nSynchronization took {time.strftime("%Hh%Mm%Ss", time.gmtime(elapsed))}.\n')
logging.shutdown()
def personAssociation(config=None): def personAssociation(config=None):
''' '''
@ -336,8 +333,6 @@ def personAssociation(config=None):
elapsed = end-start elapsed = end-start
logging.info(f'\nAssociating persons took {time.strftime("%Hh%Mm%Ss", time.gmtime(elapsed))}.\n') logging.info(f'\nAssociating persons took {time.strftime("%Hh%Mm%Ss", time.gmtime(elapsed))}.\n')
logging.shutdown()
def triangulation(config=None): def triangulation(config=None):
''' '''
@ -384,8 +379,6 @@ def triangulation(config=None):
elapsed = end-start elapsed = end-start
logging.info(f'\nTriangulation took {time.strftime("%Hh%Mm%Ss", time.gmtime(elapsed))}.\n') logging.info(f'\nTriangulation took {time.strftime("%Hh%Mm%Ss", time.gmtime(elapsed))}.\n')
logging.shutdown()
def filtering(config=None): def filtering(config=None):
''' '''
@ -429,8 +422,6 @@ def filtering(config=None):
logging.info('\n') logging.info('\n')
logging.shutdown()
def markerAugmentation(config=None): def markerAugmentation(config=None):
''' '''
@ -474,65 +465,58 @@ def markerAugmentation(config=None):
elapsed = end-start elapsed = end-start
logging.info(f'\nMarker augmentation took {time.strftime("%Hh%Mm%Ss", time.gmtime(elapsed))}.\n') logging.info(f'\nMarker augmentation took {time.strftime("%Hh%Mm%Ss", time.gmtime(elapsed))}.\n')
logging.shutdown()
def opensimProcessing(config=None): def opensimProcessing(config=None):
''' '''
Uses OpenSim to run scaling based on a static trc pose Performing OpenSim scaling and inverse kinematics.
and inverse kinematics based on a trc motion file. Selected the 10% slowest frames from trc for scaling
Saved as .osim and .mot
config can be a dictionary, config can be a dictionary,
or a the directory path of a trial, participant, or session, or a the directory path of a trial, participant, or session,
or the function can be called without an argument, in which case it the config directory is the current one. or the function can be called without an argument, in which case it the config directory is the current one.
''' '''
raise NotImplementedError('This has not been implemented yet. \nPlease see README.md for further explanation') from Pose2Sim.kinematics import opensimProcessing
# # TODO # Read the configuration files
# from Pose2Sim.opensimProcessing import opensim_processing_all level, config_dicts = read_config_files(config)
# # Determine the level at which the function is called (root:2, trial:1) # Ensure the configuration is properly structured
# level, config_dicts = read_config_files(config) if isinstance(config, dict):
config_dict = config_dicts[0]
if config_dict.get('project').get('project_dir') is None:
raise ValueError('Please specify the project directory in config_dict:\n \
config_dict.get("project").update({"project_dir":"<YOUR_TRIAL_DIRECTORY>"})')
# if type(config)==dict: session_dir = os.path.realpath(os.path.join(config_dicts[0].get('project').get('project_dir'), '..'))
# config_dict = config_dicts[0] setup_logging(session_dir)
# if config_dict.get('project').get('project_dir') == None:
# raise ValueError('Please specify the project directory in config_dict:\n \
# config_dict.get("project").update({"project_dir":"<YOUR_TRIAL_DIRECTORY>"})')
# # Set up logging # Process each configuration dictionary
# session_dir = os.path.realpath(os.path.join(config_dicts[0].get('project').get('project_dir'), '..')) for config_dict in config_dicts:
# setup_logging(session_dir) start = time.time()
currentDateAndTime = datetime.now()
project_dir = os.path.realpath(config_dict.get('project').get('project_dir'))
seq_name = os.path.basename(project_dir)
frame_range = config_dict.get('project').get('frame_range')
frames = ["all frames" if frame_range == [] else f"frames {frame_range[0]} to {frame_range[1]}"][0]
# # Batch process all trials logging.info("\n---------------------------------------------------------------------")
# for config_dict in config_dicts: logging.info(f"OpenSim processing for {seq_name}, for {frames}.")
# currentDateAndTime = datetime.now() logging.info(f"On {currentDateAndTime.strftime('%A %d. %B %Y, %H:%M:%S')}")
# start = time.time() logging.info(f"Project directory: {project_dir}")
# project_dir = os.path.realpath(config_dict.get('project').get('project_dir')) logging.info("---------------------------------------------------------------------\n")
# seq_name = os.path.basename(project_dir)
# frame_range = config_dict.get('project').get('frame_range')
# frames = ["all frames" if frame_range == [] else f"frames {frame_range[0]} to {frame_range[1]}"][0]
# logging.info("\n---------------------------------------------------------------------") try:
# # if static_file in project_dir: opensimProcessing(config_dict)
# # logging.info(f"Scaling model with <STATIC TRC FILE>.")
# # else:
# # logging.info(f"Running inverse kinematics <MOTION TRC FILE>.")
# logging.info(f"On {currentDateAndTime.strftime('%A %d. %B %Y, %H:%M:%S')}")
# logging.info(f"OpenSim output directory: {project_dir}")
# logging.info("---------------------------------------------------------------------\n")
# opensim_processing_all(config_dict) except Exception as e:
logging.error(f"Error during OpenSim processing: {e}")
continue
# end = time.time() end = time.time()
# elapsed = end-start elapsed = end - start
# # if static_file in project_dir: logging.info(f'\nOpenSim processing took {time.strftime("%Hh%Mm%Ss", time.gmtime(elapsed))}.\n')
# # logging.info(f'Model scaling took {time.strftime("%Hh%Mm%Ss", time.gmtime(elapsed))}.\n')
# # else:
# # logging.info(f'Inverse kinematics took {time.strftime("%Hh%Mm%Ss", time.gmtime(elapsed))}.\n')
# logging.shutdown()
def runAll(config=None, do_calibration=True, do_poseEstimation=True, do_synchronization=True, do_personAssociation=True, do_triangulation=True, do_filtering=True, do_markerAugmentation=True, do_opensimProcessing=True): def runAll(config=None, do_calibration=True, do_poseEstimation=True, do_synchronization=True, do_personAssociation=True, do_triangulation=True, do_filtering=True, do_markerAugmentation=True, do_opensimProcessing=True):
@ -626,17 +610,17 @@ def runAll(config=None, do_calibration=True, do_poseEstimation=True, do_synchron
logging.info('Skipping marker augmentation.') logging.info('Skipping marker augmentation.')
logging.info("\n\n=====================================================================") logging.info("\n\n=====================================================================")
# if do_opensimProcessing: if do_opensimProcessing:
# logging.info("\n\n=====================================================================") logging.info("\n\n=====================================================================")
# logging.info('Running opensim processing.') logging.info("Running OpenSim processing.")
# logging.info("=====================================================================") logging.info("=====================================================================")
# opensimProcessing(config) opensimProcessing(config)
# else: else:
# logging.info("\n\n=====================================================================") logging.info("\n\n=====================================================================")
# logging.info('Skipping opensim processing.') logging.info('Skipping OpenSim processing.')
# logging.info("=====================================================================") logging.info("\n\n=====================================================================")
logging.info("Pose2Sim pipeline completed.")
end = time.time() end = time.time()
elapsed = end-start elapsed = end-start
logging.info(f'\nRUNNING ALL FUNCTIONS TOOK {time.strftime("%Hh%Mm%Ss", time.gmtime(elapsed))}.\n') logging.info(f'\nRUNNING ALL FUNCTIONS TOOK {time.strftime("%Hh%Mm%Ss", time.gmtime(elapsed))}.\n')
logging.shutdown()

8
Pose2Sim/Utilities/tests.py
View File

@ -15,12 +15,14 @@
- triangulation - triangulation
- filtering - filtering
- marker augmentation - marker augmentation
- opensim processing
- Multi-person: - Multi-person:
- synchronization - synchronization
- person association - person association
- triangulation - triangulation
- filtering - filtering
- marker augmentation - marker augmentation
- opensim processing
- SINGLE TRIAL: - SINGLE TRIAL:
- calibration - calibration
@ -29,6 +31,7 @@
- triangulation - triangulation
- filtering - filtering
- marker augmentation - marker augmentation
- opensim processing
N.B.: N.B.:
1. Calibration from scene dimensions is not tested, as it requires the 1. Calibration from scene dimensions is not tested, as it requires the
@ -74,6 +77,7 @@ class TestWorkflow(unittest.TestCase):
- triangulation - triangulation
- filtering - filtering
- marker augmentation - marker augmentation
- OpenSim processing
- run all - run all
N.B.: Calibration from scene dimensions is not tested, as it requires the N.B.: Calibration from scene dimensions is not tested, as it requires the
@ -109,7 +113,7 @@ class TestWorkflow(unittest.TestCase):
Pose2Sim.triangulation(config_dict) Pose2Sim.triangulation(config_dict)
Pose2Sim.filtering(config_dict) Pose2Sim.filtering(config_dict)
Pose2Sim.markerAugmentation(config_dict) Pose2Sim.markerAugmentation(config_dict)
# Pose2Sim.kinematics(config_dict) Pose2Sim.opensimProcessing(config_dict)
config_dict.get("pose").update({"overwrite_pose":False}) config_dict.get("pose").update({"overwrite_pose":False})
Pose2Sim.runAll(config_dict) Pose2Sim.runAll(config_dict)
@ -137,7 +141,7 @@ class TestWorkflow(unittest.TestCase):
Pose2Sim.triangulation(config_dict) Pose2Sim.triangulation(config_dict)
Pose2Sim.filtering(config_dict) Pose2Sim.filtering(config_dict)
Pose2Sim.markerAugmentation(config_dict) Pose2Sim.markerAugmentation(config_dict)
# Pose2Sim.kinematics(config_dict) Pose2Sim.opensimProcessing(config_dict)
# Run all # Run all
config_dict.get("pose").update({"overwrite_pose":False}) config_dict.get("pose").update({"overwrite_pose":False})

511
Pose2Sim/kinematics.py Normal file
View File

@ -0,0 +1,511 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
'''
###########################################################################
## KINEMATICS PROCESSING ##
###########################################################################
Process kinematic data using OpenSim tools.
This script performs scaling, inverse kinematics, and related processing
on 3D motion capture data (TRC files). The scaling process adjusts the
generic model to match the subject's physical dimensions, while inverse
kinematics computes the joint angles based on the motion data.
Set your parameters in Config.toml.
INPUTS:
- a directory containing TRC files
- kinematic processing parameters in Config.toml
OUTPUT:
- scaled OpenSim model files (.osim)
- joint angle data files (.mot)
'''
import os
import sys
from collections import defaultdict
from pathlib import Path
import numpy as np
import pandas as pd
from lxml import etree
import logging
import opensim
## FUNCTIONS
def find_config_and_pose3d(project_dir):
"""
Find configuration files and associated pose-3d directories in the project directory.
Args:
project_dir (str): The root directory of the project.
Returns:
list: A list of tuples containing the config path and the corresponding pose-3d directory.
"""
config_paths = []
for root, dirs, files in os.walk(project_dir):
if 'Config.toml' in files:
config_path = Path(root) / 'Config.toml'
possible_pose3d_dir = Path(root) / 'pose-3d'
if not possible_pose3d_dir.exists():
possible_pose3d_dir = Path(root).parent / 'pose-3d'
if possible_pose3d_dir.exists():
config_paths.append((config_path, possible_pose3d_dir))
else:
logging.warning(f"No pose-3d directory found for config: {config_path}")
return config_paths
def get_grouped_files(directory, pattern='*.trc'):
"""
Group TRC files by person ID or treat them as single-person if no ID is found.
Args:
directory (str): The directory containing TRC files.
pattern (str): The file pattern to search for.
Returns:
dict: A dictionary grouping TRC files by person ID.
"""
files = list(Path(directory).glob(pattern))
grouped_files = defaultdict(list)
for file in files:
parts = file.stem.split('_')
if len(parts) > 2 and 'P' in parts[2]: # Multi-person file naming convention
person_id = parts[2]
else:
person_id = "SinglePerson"
grouped_files[person_id].append(file)
return grouped_files
def process_all_groups(config_dict):
"""
Process all groups (single or multi-person) based on the configuration.
Args:
config_dict (dict): The configuration dictionary containing project details.
"""
logging.info("Processing all groups in the project.")
project_dir = config_dict.get('project', {}).get('project_dir')
config_and_pose3d_paths = find_config_and_pose3d(project_dir)
for config_path, pose3d_dir in config_and_pose3d_paths:
logging.info(f"Processing setup with config: {config_path}")
trc_groups = get_grouped_files(pose3d_dir)
trial_name = Path(pose3d_dir).parent.name # Use the parent directory name as the trial name
for person_id, trc_files in trc_groups.items():
filtered_trc_files = load_trc(config_dict, trc_files)
# Ensure output directory includes the trial name
trial_output_dir = get_output_dir(Path(config_dict['project']['project_dir']).parent / trial_name, person_id)
perform_scaling(config_dict, person_id, filtered_trc_files, trial_output_dir)
perform_inverse_kinematics(config_dict, person_id, filtered_trc_files, trial_output_dir)
def load_trc(config_dict, trc_files):
"""
Load and filter TRC files according to the configuration.
Args:
config_dict (dict): The configuration dictionary.
trc_files (list): A list of TRC file paths.
Returns:
list: A list of filtered TRC files based on the criteria specified in the configuration.
"""
opensim_config = config_dict.get('opensim', {})
use_lstm = opensim_config.get('use_augmentation', False)
load_trc_name = opensim_config.get('load_trc_name', 'default')
# Filter out any scaled TRC files
unscaled_trc_files = [file for file in trc_files if '_scaling' not in str(file)]
logging.info(f"Starting TRC file filtering with criteria: use_lstm = {use_lstm}, load_trc_name = {load_trc_name}")
logging.info(f"Initial list of TRC files: {unscaled_trc_files}")
# Initialize the list to store filtered TRC files
trc_files = []
# Check for LSTM files if LSTM is being used
if use_lstm:
lstm_files = [file for file in unscaled_trc_files if '_LSTM.trc' in str(file)]
if not lstm_files:
raise FileNotFoundError("No LSTM TRC file found in the provided list.")
trc_files.extend(lstm_files)
# Check for default or filtered TRC files
if load_trc_name == 'default':
default_files = [file for file in unscaled_trc_files if '_LSTM' not in str(file) and '_filt_butterworth' not in str(file)]
trc_files.extend(default_files)
elif load_trc_name == 'filtered':
filtered_files = [file for file in unscaled_trc_files if '_filt_butterworth' in str(file) and '_LSTM' not in str(file)]
trc_files.extend(filtered_files)
# If no TRC files are found after filtering, raise an error
if not trc_files:
logging.error(f"No suitable TRC files found with the specified criteria: use_lstm = {use_lstm}, load_trc_name = {load_trc_name}")
raise FileNotFoundError(f"No suitable TRC files found in the provided list with the specified criteria: use_lstm = {use_lstm}, load_trc_name = {load_trc_name}")
logging.info(f"Filtered TRC files: {trc_files}")
return trc_files
def read_trc(trc_path):
"""
Read a TRC file and extract its contents.
Args:
trc_path (str): The path to the TRC file.
Returns:
tuple: A tuple containing the Q coordinates, frames column, time column, and header.
"""
try:
logging.info(f"Attempting to read TRC file: {trc_path}")
with open(trc_path, 'r') as trc_file:
header = [next(trc_file) for _ in range(5)]
trc_df = pd.read_csv(trc_path, sep="\t", skiprows=4, encoding='utf-8')
frames_col, time_col = trc_df.iloc[:, 0], trc_df.iloc[:, 1]
Q_coords = trc_df.drop(trc_df.columns[[0, 1]], axis=1)
return Q_coords, frames_col, time_col, header
except Exception as e:
logging.error(f"Error reading TRC file at {trc_path}: {e}")
raise
def make_trc_with_Q(Q, header, trc_path):
"""
Write the processed Q coordinates back to a TRC file.
Args:
Q (pd.DataFrame): The Q coordinates data.
header (list): The header of the original TRC file.
trc_path (str): Path to save the new TRC file.
"""
header_2_split = header[2].split('\t')
header_2_split[2] = str(len(Q))
header_2_split[-1] = str(len(Q))
header[2] = '\t'.join(header_2_split) + '\n'
time = pd.Series(np.arange(len(Q)) / float(header_2_split[0]), name='t')
Q.insert(0, 't', time)
with open(trc_path, 'w') as trc_o:
[trc_o.write(line) for line in header]
Q.to_csv(trc_o, sep='\t', index=True, header=None, lineterminator='\n')
def get_key(config_dict):
"""
Determine the key for the OpenSim model and setup files based on the configuration.
Args:
config_dict (dict): The configuration dictionary.
Returns:
str: The key used to select the model and setup files.
"""
use_augmentation = config_dict.get('opensim', {}).get('use_augmentation', False)
if use_augmentation:
return 'LSTM'
pose_model = config_dict.get('pose', {}).get('pose_model', '').upper()
if not pose_model:
raise ValueError(f"Invalid or missing 'pose_model' in config: {pose_model}")
return pose_model
def get_OpenSim_Setup():
"""
Locate the OpenSim setup directory within the Pose2Sim package.
Returns:
Path: The path to the OpenSim setup directory.
"""
pose2sim_path = Path(sys.modules['Pose2Sim'].__file__).resolve().parent
setup_dir = pose2sim_path / 'OpenSim_Setup'
return setup_dir
def get_Model(config_dict):
"""
Retrieve the OpenSim model file path based on the configuration.
Args:
config_dict (dict): The configuration dictionary.
Returns:
str: The path to the OpenSim model file.
"""
setup_key = get_key(config_dict)
setup_dir = get_OpenSim_Setup()
if setup_key == 'LSTM':
pose_model_file = 'Model_Pose2Sim_LSTM.osim'
elif setup_key == 'BLAZEPOSE':
pose_model_file = 'Model_Pose2Sim_Blazepose.osim'
elif setup_key == 'BODY_25':
pose_model_file = 'Model_Pose2Sim_Body25.osim'
elif setup_key == 'BODY_25B':
pose_model_file = 'Model_Setup_Pose2Sim_Body25b.osim'
elif setup_key == 'BODY_135':
pose_model_file = 'Model_Pose2Sim_Body135.osim'
elif setup_key == 'COCO_17':
pose_model_file = 'Model_Pose2Sim_Coco17.osim'
elif setup_key == 'COCO_133':
pose_model_file = 'Model_Pose2Sim_Coco133.osim'
elif setup_key == 'HALPE_26':
pose_model_file = 'Model_Pose2Sim_Halpe26.osim'
elif setup_key == 'HALPE_68':
pose_model_file = 'Model_Pose2Sim_Halpe68_136.osim'
else:
raise ValueError(f"pose_model '{setup_key}' not found.")
pose_model_path = os.path.join(setup_dir, pose_model_file)
return pose_model_path
def get_Scale_Setup(config_dict):
"""
Retrieve the OpenSim scaling setup file path based on the configuration.
Args:
config_dict (dict): The configuration dictionary.
Returns:
str: The path to the OpenSim scaling setup file.
"""
setup_key = get_key(config_dict)
setup_dir = get_OpenSim_Setup()
if setup_key == 'LSTM':
scale_setup_file = 'Scaling_Setup_Pose2Sim_LSTM.xml'
elif setup_key == 'BLAZEPOSE':
scale_setup_file = 'Scaling_Setup_Pose2Sim_Blazepose.xml'
elif setup_key == 'BODY_25':
scale_setup_file = 'Scaling_Setup_Pose2Sim_Body25.xml'
elif setup_key == 'BODY_25B':
scale_setup_file = 'Scaling_Setup_Pose2Sim_Body25b.xml'
elif setup_key == 'BODY_135':
scale_setup_file = 'Scaling_Setup_Pose2Sim_Body135.xml'
elif setup_key == 'COCO_17':
scale_setup_file = 'Scaling_Setup_Pose2Sim_Coco17.xml'
elif setup_key == 'COCO_133':
scale_setup_file = 'Scaling_Setup_Pose2Sim_Coco133.xml'
elif setup_key == 'HALPE_26':
scale_setup_file = 'Scaling_Setup_Pose2Sim_Halpe26.xml'
elif setup_key == 'HALPE_68':
scale_setup_file = 'Scaling_Setup_Pose2Sim_Halpe68_136.xml'
else:
raise ValueError(f"pose_model '{setup_key}' not found.")
scale_setup_path = os.path.join(setup_dir, scale_setup_file)
return scale_setup_path
def get_IK_Setup(config_dict):
"""
Retrieve the OpenSim inverse kinematics setup file path based on the configuration.
Args:
config_dict (dict): The configuration dictionary.
Returns:
str: The path to the OpenSim inverse kinematics setup file.
"""
setup_key = get_key(config_dict)
setup_dir = get_OpenSim_Setup()
if setup_key == 'LSTM':
ik_setup_file = 'IK_Setup_Pose2Sim_LSTM.xml'
elif setup_key == 'BLAZEPOSE':
ik_setup_file = 'IK_Setup_Pose2Sim_Blazepose.xml'
elif setup_key == 'BODY_25':
ik_setup_file = 'IK_Setup_Pose2Sim_Body25.xml'
elif setup_key == 'BODY_25B':
ik_setup_file = 'IK_Setup_Pose2Sim_Body25b.xml'
elif setup_key == 'BODY_135':
ik_setup_file = 'IK_Setup_Pose2Sim_Body135.xml'
elif setup_key == 'COCO_17':
ik_setup_file = 'IK_Setup_Pose2Sim_Coco17.xml'
elif setup_key == 'COCO_133':
ik_setup_file = 'IK_Setup_Pose2Sim_Coco133.xml'
elif setup_key == 'HALPE_26':
ik_setup_file = 'IK_Setup_Pose2Sim_Halpe26.xml'
elif setup_key == 'HALPE_68':
ik_setup_file = 'IK_Setup_Pose2Sim_Halpe68_136.xml'
else:
raise ValueError(f"pose_model '{setup_key}' not found.")
ik_setup_path = os.path.join(setup_dir, ik_setup_file)
return ik_setup_path
def get_output_dir(config_dir, person_id):
"""
Determines the correct output directory based on the configuration and the person identifier.
Args:
config_dir (Path): The root directory where the configuration file is located.
person_id (str): Identifier for the person (e.g., 'SinglePerson', 'P1').
Returns:
Path: The path where the output files should be stored.
"""
output_dir = config_dir / 'opensim' # Assuming 'opensim' as the default output subdirectory
# Append the person_id to the output directory if it's a multi-person setup
if person_id != "SinglePerson":
output_dir = output_dir / person_id
logging.debug(f"Output directory determined as: {output_dir}")
# Create the directory if it does not exist
if not output_dir.exists():
output_dir.mkdir(parents=True, exist_ok=True)
return output_dir
def perform_scaling(config_dict, person_id, trc_files, output_dir):
"""
Perform scaling on the TRC files according to the OpenSim configuration.
Args:
config_dict (dict): The configuration dictionary.
person_id (str): The person identifier (e.g., 'SinglePerson', 'P1').
trc_files (list): List of TRC files to be processed.
output_dir (Path): The directory where the output files should be saved.
"""
geometry_path = Path(get_OpenSim_Setup()) / 'Geometry'
geometry_path_str = str(geometry_path)
opensim.ModelVisualizer.addDirToGeometrySearchPaths(geometry_path_str)
try:
athlete_config = config_dict.get('project', {})
athlete_height = athlete_config.get('participant_height', -1)
athlete_weight = athlete_config.get('participant_mass', -1)
if person_id == "SinglePerson":
if not isinstance(athlete_height, float) or not isinstance(athlete_weight, float):
raise ValueError("For a single person configuration, 'participant_height' and 'participant_mass' must be floats.")
else:
if person_id.startswith("P"):
try:
person_idx = int(person_id.replace('P', '')) - 1
athlete_height = athlete_height[person_idx]
athlete_weight = athlete_weight[person_idx]
except (ValueError, IndexError) as e:
raise ValueError(f"Error processing multi-person data for '{person_id}': {e}")
else:
raise ValueError(f"Unexpected person_id format: '{person_id}'")
logging.debug(f"Performing scaling. Output directory: {output_dir}")
pose_model = get_Model(config_dict)
if not pose_model:
raise ValueError(f"Model path not found for pose_model: {pose_model}")
for trc_file in trc_files:
trc_file = Path(trc_file)
scaling_path = get_Scale_Setup(config_dict)
Q_coords, _, _, header = read_trc(trc_file)
Q_diff = Q_coords.diff(axis=0).sum(axis=1)
Q_diff = Q_diff[Q_diff != 0]
min_speed_indices = Q_diff.abs().nsmallest(int(len(Q_diff) * 0.1)).index
Q_coords_scaling = Q_coords.iloc[min_speed_indices].reset_index(drop=True)
trc_scaling_path = trc_file.parent / (trc_file.stem + '_scaling.trc')
make_trc_with_Q(Q_coords_scaling, header, str(trc_scaling_path))
scaling_file_path = str(trc_file.parent / (trc_file.stem + '_' + Path(scaling_path).name))
scaled_model_path = (output_dir / (trc_file.stem + '_scaled.osim')).resolve()
scaling_tree = etree.parse(str(scaling_path))
scaling_root = scaling_tree.getroot()
scaling_root[0].find('mass').text = str(athlete_weight)
scaling_root[0].find('height').text = str(athlete_height)
scaling_root[0].find('GenericModelMaker').find('model_file').text = str(pose_model)
scaling_root[0].find('ModelScaler').find('marker_file').text = trc_scaling_path.name
scaling_root[0].find('ModelScaler').find('time_range').text = '0 ' + str(Q_coords_scaling['t'].iloc[-1])
scaling_root[0].find('ModelScaler').find('output_model_file').text = str(scaled_model_path)
scaling_root[0].find('MarkerPlacer').find('marker_file').text = trc_scaling_path.name
scaling_root[0].find('MarkerPlacer').find('time_range').text = '0 ' + str(Q_coords_scaling['t'].iloc[-1])
scaling_root[0].find('MarkerPlacer').find('output_model_file').text = str(scaled_model_path)
scaling_tree.write(scaling_file_path)
logging.debug(f"Running ScaleTool with scaling file: {scaling_file_path}")
opensim.ScaleTool(scaling_file_path).run()
except Exception as e:
logging.error(f"Error during scaling for {person_id}: {e}")
raise
def perform_inverse_kinematics(config_dict, person_id, trc_files, output_dir):
"""
Perform inverse kinematics on the TRC files according to the OpenSim configuration.
Args:
config_dict (dict): The configuration dictionary.
person_id (str): The person identifier (e.g., 'SinglePerson', 'P1').
trc_files (list): List of TRC files to be processed.
output_dir (Path): The directory where the output files should be saved.
"""
try:
logging.debug(f"Performing inverse kinematics. Output directory: {output_dir}")
for trc_file in trc_files:
trc_file_path = Path(trc_file).resolve()
scaled_model_path = Path(output_dir) / (trc_file_path.stem + '_scaled.osim')
ik_setup_path = get_IK_Setup(config_dict)
Q_coords, frames_col, time_col, header = read_trc(trc_file_path)
ik_time_range = config_dict.get('opensim', {}).get('IK_timeRange', [])
if not ik_time_range:
start_time = time_col.iloc[0]
end_time = time_col.iloc[-1]
else:
start_time, end_time = ik_time_range[0], ik_time_range[1]
ik_file_path = Path(trc_file_path.parent / (trc_file_path.stem + '_' + Path(ik_setup_path).name)).resolve()
scaled_model_path = scaled_model_path.resolve()
output_motion_file = Path(output_dir, trc_file_path.stem + '.mot').resolve()
ik_tree = etree.parse(ik_setup_path)
ik_root = ik_tree.getroot()
ik_root.find('.//model_file').text = str(scaled_model_path)
ik_root.find('.//time_range').text = f'{start_time} {end_time}'
ik_root.find('.//output_motion_file').text = str(output_motion_file)
ik_root.find('.//marker_file').text = str(trc_file_path)
ik_tree.write(ik_file_path)
logging.info(f"Running InverseKinematicsTool with TRC file: {trc_file_path}")
if not trc_file_path.exists():
raise FileNotFoundError(f"TRC file does not exist: {trc_file_path}")
logging.debug(f"Running InverseKinematicsTool with IK setup file: {ik_file_path}")
opensim.InverseKinematicsTool(str(ik_file_path)).run()
except Exception as e:
logging.error(f"Error during IK for {person_id}: {e}")
raise
def opensimProcessing(config_dict):
logging.info("Starting OpenSim processing...")
process_all_groups(config_dict)
logging.info("OpenSim processing completed successfully.")

14
README.md
View File

@ -175,9 +175,13 @@ Pose2Sim.personAssociation()
Pose2Sim.triangulation() Pose2Sim.triangulation()
Pose2Sim.filtering() Pose2Sim.filtering()
Pose2Sim.markerAugmentation() Pose2Sim.markerAugmentation()
Pose2Sim.opensimProcessing()
``` ```
3D results are stored as .trc files in each trial folder in the `pose-3d` directory. 3D results are stored as .trc files in each trial folder in the `pose-3d` directory.
OpenSim results are stored as scaled model .osim and .mot in each trial folder in the `opensim` directory.
</br> </br>
**Note:** **Note:**
@ -188,6 +192,7 @@ Pose2Sim.markerAugmentation()
Pose2Sim.runAll(do_calibration=True, do_poseEstimation=True, do_synchronization=True, do_personAssociation=True, do_triangulation=True, do_filtering=True, do_markerAugmentation=True, do_opensimProcessing=True) Pose2Sim.runAll(do_calibration=True, do_poseEstimation=True, do_synchronization=True, do_personAssociation=True, do_triangulation=True, do_filtering=True, do_markerAugmentation=True, do_opensimProcessing=True)
``` ```
- Try the calibration tool by changing `calibration_type` to `calculate` instead of `convert` in [Config.toml](https://github.com/perfanalytics/pose2sim/blob/main/Pose2Sim/Demo_SinglePerson/Config.toml) (more info [there](#calculate-from-scratch)). - Try the calibration tool by changing `calibration_type` to `calculate` instead of `convert` in [Config.toml](https://github.com/perfanalytics/pose2sim/blob/main/Pose2Sim/Demo_SinglePerson/Config.toml) (more info [there](#calculate-from-scratch)).
- If the results are not convincing, refer to Section [OpenSim-kinematics](#OpenSim-kinematics) in the document.
</br> </br>
@ -652,6 +657,15 @@ Pose2Sim.markerAugmentation()
</br> </br>
### If kinematics results are not convicing:
> _***Explanation on choosing the best frames for scaling (L437-448):***_
>
> On difficult trials, some points are not well triangulated, which can lead to bad scaling. For example, if a point of the foot is very far from the rest of the body on some frames, scaling will consider that the foot is very large. Consequently, we need to scale only on the frames that are best triangulated. Now, how to find these best frames?
>
>
> My reasoning was that the points of badly triangulated frames would go all over the place, and thus that their speeds would be fast. So I only selected the 10% slowest frames for scaling. I think that in addition, we should take the median scale factor for these frames, because we might have slow frames that are still bad. -> This last step has not been done.
### Command line ### Command line
Alternatively, you can use command-line tools: Alternatively, you can use command-line tools: