src.pipeline package¶

src.pipeline.pipeline module¶

This module contains the pipeline class.

class src.pipeline.pipeline.Pipeline(pipeline_config)¶

Bases: object

This class is the skeleton for all possible pipeline instantiations.

It is instantiated with a config dictionary that defines the different variable components of the pipeline, the data source and auxillary variables.

add_to_evaluator(subject_num, run_num, reference_data, gait_parameters)¶

Add estimated gait parameters and reference data for one subject and run to the evaluator.

Parameters

subject_num (int) – subject index
run_num (int) – run index
reference_data (dict[str, DataFrame]) – DataFrames with reference gait parameters for the left and right foot
gait_parameters (dict[str, DataFrame]) – DataFrame with the estimated gait parameters for the left and right foot

Returns

None

calculate_gait_parameters(gait_events, trajectories, imu_ic)¶

Calculate gait parameters.

Parameters

gait_events (dict[str, dict]) – IC and FO samples and timestamps for the right and left foot
trajectories (dict[str, DataFrame]) – DataFrames with trajectory information for the right and left foot
imu_ic (float) – Inital contact timestamp

Returns

DataFrame with the estimated gait parameters for the left and right foot

Return type

dict[str, DataFrame]

detect_gait_events(imus)¶

Perform gait event detection with the gait event detector specified in the config.

Parameters: imus (dict[str, IMU]) – IMU objects for each sensor location
Returns: IC and FO samples and timestamps for the right and left foot
Return type: dict[str, dict]

estimate_trajectories(subject_num, run_num, imus, imu_ic)¶

Estimate left and right foot trajectories using the specified trajectory estimation algorithm. Subject and run need to be identified since the trajectory estimator uses caching.

Parameters

subject_num (int) – subject index
run_num (int) – run index
imus (dict[str, IMU]) – IMU objects for each sensor location
imu_ic (float) – Inital contact timestamp

Returns

DataFrames with trajectory information for the right and left foot

Return type

dict[str, DataFrame]

execute(subject_runs)¶

Core function of the pipeline. For each subject and run: Load IMU data, estimate trajectories, detect gait events, calculate estimated gait parameters, add them together with the reference_data to the evaluator. Exectue the evaluator with the results of all runs altogether.

Parameters: subject_runs (tuple[int, int]) – Index of the subject and run whose data should be loaded
Returns: None

load_data(subject_num, run_num)¶

Load all IMU data for the given subject_num and run_num from the IMU folder.

Parameters

subject_num (int) – Index of the subject whose data should be loaded
run_num (int) – Index of the run/trial that should be loaded

Returns

Tuple of a dictionary of IMU objects and the timestamp of initial contact in seconds

Return type

tuple[dict[str, IMU], float)

load_reference_data(subject_num, run_num)¶

Load reference data with the specified reference loader. Note: subject and run need to be identified since the reference loader uses caching.

Parameters

subject_num (int) – subject index
run_num (int) – run index

Returns

DataFrames with gait parameters for the left and right foot

Return type

dict[str, DataFrame]

src.pipeline.abstract_pipeline_components module¶

This module implements the abstract classes/interfaces of all variable pipeline components.

class src.pipeline.abstract_pipeline_components.AbstractDataLoader(raw_base_path, dataset, subject, run)¶

Bases: object

The AbstractDataLoader defines the interface for any kind of IMU data loader.

get_data()¶

Get the loaded data.

Returns: IMU data from all sensors
Return type: dict[str, IMU]

load(raw_base_path, dataset, subject, run)¶

This method is called at instantiation and needs to be implemented by the derived class. It is expected to find the actual data files based on the given parameters and store them as a dictionary of IMU objects into self.data

Parameters

raw_base_path (str) – Base folder for every dataset
dataset (str) – Folder containing the dataset
subject (str) – Subject identifier
run (str) – Run identifier

Returns

None

class src.pipeline.abstract_pipeline_components.AbstractEventDetector(imus)¶

Bases: object

The AbstractEventDetector defines the interface for any kind of event detection algorithm.

detect()¶

This method is expected to be implemented by each event detection algorithm.

Returns: dictionaries containing gait event information for each foot.
Return type: (dict[str, dict])

class src.pipeline.abstract_pipeline_components.AbstractReferenceLoader(name, raw_base_path, interim_base_path, dataset, subject, run, overwrite)¶

Bases: object

The AbstractReferenceLoader defines the interface for any kind of reference data loader.

get_data()¶

Get the loaded reference data.

Returns: DataFrames with gait parameters for the left and right foot
Return type: dict[str, DataFrame]

load()¶

This method is called at instantiation and needs to be implemented by the derived class. It is expected to find the actual data files based on the given parameters and store reference data for the left and right foot in self.data

Returns: None

class src.pipeline.abstract_pipeline_components.AbstractTrajectoryEstimator(imus)¶

Bases: object

The AbstractTrajectoryEstimator defines the interface for each trajectory estimation algorithm.

estimate(interim_base_path, dataset, subject_num, run_num)¶

This method is expected to be implemented by each trajectory estimation algorithm.

Parameters

interim_base_path (str) – Base folder where interim data can be stored
dataset (str) – Folder containing the dataset
subject_num (int) – Subject index
run_num (int) – Run index

Returns

DataFrames containing trajectory information for each foot

Return type

dict[str, DataFrame]

src.pipeline.data_loader module¶

This module contains the implementations of actual IMU data loaders A data loader is expected to implement functionality that builds a number of IMU objects from data specified by the given parameters.

class src.pipeline.data_loader.MatlabDataLoader(raw_base_path, dataset, subject, run)¶

Bases: pipeline.abstract_pipeline_components.AbstractDataLoader

This class reads data from a CSV file as exported from a MATLAB script.

load(raw_base_path, dataset, subject, run)¶

The load function is called at instantiation (see AbstractDataLoader).

It loads the data specified by the given parameters and stores it into self.data.

Parameters

raw_base_path (str) – Base folder for every dataset
dataset (str) – Folder containing the dataset
subject (str) – Subject identifier
run (str) – Run identifier

Returns

None

class src.pipeline.data_loader.PhysilogDataLoader(raw_base_path, dataset, subject, run)¶

Bases: pipeline.abstract_pipeline_components.AbstractDataLoader

This class reads data from CSV files as generated by the PhysilogRTK.

load(raw_base_path, dataset, subject, run)¶

The load function is called at instantiation (see AbstractDataLoader).

It loads the data specified by the given parameters and stores it into self.data.

Parameters

raw_base_path (str) – Base folder for every dataset
dataset (str) – Folder containing the dataset
subject (str) – Subject identifier
run (str) – Run identifier

Returns

None

src.pipeline.reference_loader module¶

This module contains the data loaders for different reference systems.

class src.pipeline.reference_loader.OptogaitReferenceLoader(name, raw_base_path, interim_base_path, dataset, subject, run, overwrite)¶

Bases: pipeline.abstract_pipeline_components.AbstractReferenceLoader

This class loads reference data created by the OptoGait system.

load()¶

Load the data based on parameters provided in the constructor.

Returns: None

class src.pipeline.reference_loader.ZebrisReferenceLoader(name, raw_base_path, interim_base_path, dataset, subject, run, overwrite)¶

Bases: pipeline.abstract_pipeline_components.AbstractReferenceLoader

This class loads reference data created by the Zebris FDM-THQ system. It uses caching, since the extraction of foot positions from the Zebris raw data requires some computation and only needs to be done once. This computation is outsourced to the ZebrisJsonReader that can be also used to generate visualizations and inspect the data.

load()¶

Load the data based on parameters provided in the constructor.

Returns: None

load_interim_data()¶

Load data from cached files.

Returns: None

load_raw_data()¶

Load data from raw data files. Zebris has two types of files (_raw and _steps). _raw files contain raw sensor readings. _steps files contain aggregated data per roll-off cycle.

Returns: None

src.pipeline.trajectory_estimator module¶

This module contains the implementation of a trajectory estimator.

class src.pipeline.trajectory_estimator.TuncaTrajectoryEstimator(imus)¶

Bases: pipeline.abstract_pipeline_components.AbstractTrajectoryEstimator

Trajectory estimator based on Tunca et al. (https://doi.org/10.3390/s17040825). The actual error-state Kalman filter is implemented in trajectory_estimation/filter.py

estimate(name, interim_base_path, dataset, subject, run, imu_ic, stance_thresholds, overwrite)¶

Estimate trajectories from IMU data.

Parameters

name (str) – Name of the configuration (used to identify cache files)
interim_base_path (str) – Folder where caching data can be stored
dataset (str) – Folder containing the dataset
subject (str) – Identifier of the subject
run (str) – Identifier of the run
imu_ic (float) – timestamp of initial contact in the IMU data
stance_thresholds (dict[str, float]) – Gyroscope magnitude and stance count thresholds for stance detection for the right and left foot
overwrite (bool) – Flag if cached files should be overwritten

Returns

DataFrames with trajectory information for the right and left foot

Return type

dict[str, DataFrame]

src.pipeline.event_detector module¶

This module contains implementation of different gait event detectors.

class src.pipeline.event_detector.HundzaEventDetector(imus)¶

Bases: pipeline.abstract_pipeline_components.AbstractEventDetector

Gait event detection by Hundza et al. (https://doi.org/10.1109/TNSRE.2013.2282080) An example of another gait event detection algorithm. This EventDetector is for demonstration only and has not been evaluated.

detect()¶

Detect gait events.

Returns: TOFS, IOFS and TO events
Return type: dict[str, dict]

class src.pipeline.event_detector.TuncaEventDetector(imus)¶

Bases: pipeline.abstract_pipeline_components.AbstractEventDetector

Gait event detection as presented by Tunca et al. (https://doi.org/10.3390/s17040825). The actual algorithm is implemented in event_datection/imu_event_detection.py

detect(stance_thresholds)¶

Detect gait events.

Parameters: stance_thresholds (dict[str, float]) – Gyroscope magnitude and stance count thresholds for stance detection
Returns: IC and FO samples and timestamps for the right and left foot.
Return type: dict[str, dict]

src.pipeline.gait_parameters module¶

This module contains the GaitParameters class.

class src.pipeline.gait_parameters.GaitParameters(trajectories, gait_events, initial_contact)¶

Bases: object

This class merges trajectories and gait events together and calculates the gait parameters stride length, stride time, swing time and stance time.

adjust_data()¶

Adjust gait events so that the data starts and ends with a start event.

Returns: None

stance_time()¶

Calculate stance time.

Returns: List of stance times for the right and left foot
Return type: dict[str, list[float]]

stride_length()¶

Calculate stride length.

Returns: List of stride length for the right and left foot
Return type: dict[str, list[float]]

stride_time()¶

Calculate stride time.

Returns: List of stride times for the right and left foot
Return type: dict[str, list[float]]

summary()¶

Calculate the actual gait parameters.

Returns: DataFrame with the gait parameters
Return type: DataFrame

swing_time()¶

Calculate swing time.

Returns: List of swing times for the right and left foot
Return type: dict[str, list[float]]

src.pipeline.evaluator module¶

This module contains the implementation of the Evaluator.

class src.pipeline.evaluator.Evaluator¶

Bases: object

The Evaluator merges temporal and spatial gait parameters from the trajectory estimation algorithm, the gait event detection algorithm and the reference system and creates plots and metrics in order to assess the quality of the used algorithms and data.

The pipeline is meant to be executed for each trial individually. The results of the pipeline are then added to the Evaluator using add_data().

add_data(subject_num, run_num, data, reference_data)¶

Adds a new pair of IMU and reference data for one trial.

Parameters

subject_num (int) – Subject index
run_num (int) – Run index
data (dict[str, DataFrame]) – Gait parameters produced by the pipeline for each foot
reference_data (dict[str, DataFrame]) – Reference gait parameters for each foot

Returns

None

detect_outliers(column)¶

Detect outliers via z-score and maximum deviation for one column. Marks the outliers in a special boolean column.

Parameters: column (str) – Name of the column under consideration (stride_length, stride_time, swing_time, stance_time).
Returns: None

filter_outliers(data)¶

Select all datapoints that have not been labeled previously as outliers by detect_outliers().

Parameters: data (DataFrame) – DataFrame with outliers marked in column “outlier”
Returns: Filtered DataFrame without outliers
Return type: DataFrame

load(filename)¶

Load the evaluation results from a json file.

Parameters: filename (str) – filename of the saved data
Returns: None

match_timestamps()¶

Match timestamps from reference system and IMUs. All timestamps are zero based to the timestamp of initial contact. In order to match gait parameters form the reference system and IMUs stride by stride, they are merged based on minimal time difference of the timestamps within a certain tolerance. Strides that cannot be matched are dropped.

Returns: None

merge_sides(merged_subject_runs)¶

Merge data from left and right foot.

Parameters: merged_subject_runs (dict[str, DataFrame]) – DataFrames for left and right foot
Returns: Merged DataFrame for both, left and right foot
Return type: DataFrame

merge_subject_runs(subject_run_nums)¶

Merge data from all subjects and runs into one DataFrame and add special columns for subject and run.

Parameters: subject_run_num (list[tuple[int, int]]) – list of subject and run numbers that should be merged
Returns: merged DataFrame for each foot
Return type: dict[str, DataFrame]

plot_bland_altmann(column, subject_run_nums, reference_name)¶

Create Bland-Altman plots of IMU data and reference data for the selected subjects and runs.

Parameters

column (str) – Name of the data column that should be plotted
subject_run_nums (list[tuple[int, int]]) – List of subject and run indices
reference_name (str) – Name of the reference system (used as axis label)

Returns

None

plot_correlation(title, column, subject_run_nums, reference_name)¶

Create correlation plot of IMU data and reference data for the selected subjects and runs.

Parameters

title (str) – Title of the plot
column (str) – Name of the data column that should be plotted
subject_run_nums (list[tuple[int, int]]) – List of subject and run indices
reference_name (str) – Name of the reference system (used as axis label)

Returns

None

reg_line(x, y)¶

Calculate linear regression on two dimensional data.

Parameters

x (list[float]) – x values of the data
y (list[float]) – y values of the data

Returns

x values of the regression line, y values of the regression line, calculated regression line parameters (gradient, intercept, r_value, p_value, std_err, rmse, mae), p-values of the statistical model, confidence interval

Return type

tuple(list[float], ..)

save(filename)¶

Save the evaluation results to a json file. This can be handy if different plots should be produced without rerunning the whole pipeline.

Parameters: filename (str) – filename for the saved data
Returns: None

src.pipeline package¶

src.pipeline.pipeline module¶

src.pipeline.abstract_pipeline_components module¶

src.pipeline.data_loader module¶

src.pipeline.reference_loader module¶

src.pipeline.trajectory_estimator module¶

src.pipeline.event_detector module¶

src.pipeline.gait_parameters module¶

src.pipeline.evaluator module¶

Table of Contents

Previous topic

Next topic

This Page