dispel.providers.generic.tasks.gait.core module#

All generic transformation steps related to the gait.

class dispel.providers.generic.tasks.gait.core.AverageRollingVerticalAcceleration[source]#

Bases: TransformStep

Extract and smooth vertical acceleration (in unit G).

The vertical acceleration is smoothed by using a centered moving average method.

Parameters:

data_set_id – The id corresponding to the preprocessed accelerometer signal.
axis – The axis on which the centered moving average is computed.
rolling_window – The window size used in the centered moving average operation. Default value 5 has been determined empirically in order to provide the best results.

__init__(data_set_id, axis='userAccelerationX', rolling_window=5, change_sign=False)[source]#

Parameters:

data_set_id (str) –
axis (str) –
rolling_window (int) –
change_sign (bool) –

class dispel.providers.generic.tasks.gait.core.DetectBouts[source]#

Bases: TransformStep

Detect walking bouts.

definitions: List[RawDataValueDefinition] = [<RawDataValueDefinition: start_time (start time)>, <RawDataValueDefinition: end_time (end time)>, <RawDataValueDefinition: detected_walking (detected walking)>, <RawDataValueDefinition: duration (duration)>, <RawDataValueDefinition: bout_id (bout id)>]#

class dispel.providers.generic.tasks.gait.core.DetectBoutsBeltPlacement[source]#

Bases: DetectBouts

Detect walking bouts filtered for placement.

data_set_ids: str | Iterable[str] = ['movement_bouts', 'placement_bouts']#: An iterable of data sets to be being processed

new_data_set_id: str = 'walking_placement_bouts'#

thres_min_bout = <dispel.processing.core.Parameter object>#

class dispel.providers.generic.tasks.gait.core.DetectBoutsBeltPlacementNoTurns[source]#

Bases: DetectBouts

Detect walking bouts filtered for placement and no turns.

data_set_ids: str | Iterable[str] = ['movement_bouts', 'placement_bouts', 'gyroscope_ts_rotated_resampled_butterworth_low_pass_filter_x_turns']#: An iterable of data sets to be being processed

false_final_thres_min_bout = <dispel.processing.core.Parameter object>#

new_data_set_id: str = 'walking_placement_no_turn_bouts'#

skip_first_seconds = <dispel.processing.core.Parameter object>#

true_final_thres_min_bout = <dispel.processing.core.Parameter object>#

turn_max_angle = <dispel.processing.core.Parameter object>#

class dispel.providers.generic.tasks.gait.core.DetectBoutsHarmonic[source]#

Bases: DetectBouts

Detect walking bouts with CWT.

data_set_ids: str | Iterable[str] = 'vertical_acceleration'#: An iterable of data sets to be being processed

new_data_set_id: str = 'movement_bouts'#

class dispel.providers.generic.tasks.gait.core.DetectMovementBouts[source]#

Bases: DetectBouts

Detect walking bouts based on dynamics algorithm.

cutoff = <dispel.processing.core.Parameter object>#

data_set_ids: str | Iterable[str] = 'acc_ts_rotated_resampled_detrend'#: An iterable of data sets to be being processed

new_data_set_id: str = 'movement_bouts'#

rolling_step_size = <dispel.processing.core.Parameter object>#

rolling_window_size = <dispel.processing.core.Parameter object>#

thres_min_bout = <dispel.processing.core.Parameter object>#

thres_moving = <dispel.processing.core.Parameter object>#

thres_upright = <dispel.processing.core.Parameter object>#

class dispel.providers.generic.tasks.gait.core.DetectStepsProcessingBase[source]#

Bases: TransformStep

An abstract base class for step detection.

Given the step detection algorithm specified through the method argument, e.g.: dispel.providers.generic.tasks.gait.lee.detect_steps(), the transform step run the step detection on each of the walking bouts and create a generic pandas.DataFrame with annotated events as in Bourke et. al.

definitions: List[RawDataValueDefinition] = [<RawDataValueDefinition: foot (foot)>, <RawDataValueDefinition: event (event)>, <RawDataValueDefinition: bout_id (bout id)>]#

class dispel.providers.generic.tasks.gait.core.DetectStepsWithoutBoutsBase[source]#

Bases: TransformStep

Generic detect steps transform without walking bouts.

definitions: List[RawDataValueDefinition] = [<RawDataValueDefinition: foot (foot)>, <RawDataValueDefinition: event (event)>]#

class dispel.providers.generic.tasks.gait.core.DetectWalking[source]#

Bases: TransformStep

Identify walking in three seconds fixed temporal windows.

__init__()[source]#

class dispel.providers.generic.tasks.gait.core.ExtractPowerBoutDivSteps[source]#

Bases: GaitBoutExtractStep

Extract step to compute the Step Power.

Parameters:

data_set_ids – The data set ids corresponding to first the walking bouts then the signal to take as the magnitude of acceleration and the step dataset.
level_filter – An optional LevelFilter to determine the levels for extraction.

__init__(data_set_ids, **kwargs)[source]#

Parameters:: data_set_ids (str) –

class dispel.providers.generic.tasks.gait.core.ExtractStepCount[source]#

Bases: GaitBoutExtractStep

Extract step count.

Parameters:

data_set_ids – A list of two elements with the id of the walking bout data set and the id of the step_dataset containing the detected steps formatted as the output of TransformStepDetection.
level_filter – An optional LevelFilter to determine the levels for extraction.

definition: ValueDefinition | ValueDefinitionPrototype | None = <dispel.data.measures.MeasureValueDefinitionPrototype object>#: The specification of the measure definition

get_definition(**kwargs)[source]#

Get the measure definition.

Return type:: ValueDefinition

transform_function()#

Count the number of steps.

Parameters:: data (DataFrame) – A pandas data frame containing the detected steps.
Returns:: The number of steps.
Return type:: int

class dispel.providers.generic.tasks.gait.core.ExtractStepDurationAll[source]#

Bases: GaitBoutAggregateStep

Extract step duration related measures.

Parameters:: data_set_ids – The data set ids that will be considered to extract step duration measures.

__init__(data_set_ids, **kwargs)[source]#

Parameters:: data_set_ids (List[str]) –

class dispel.providers.generic.tasks.gait.core.ExtractStepIntensityAll[source]#

Bases: GaitBoutAggregateStep

Extract step intensity related measures.

Parameters:: data_set_ids – The data set ids that will be considered to extract step intensity measures.

__init__(data_set_ids, **kwargs)[source]#

Parameters:: data_set_ids (List[str]) –

class dispel.providers.generic.tasks.gait.core.ExtractStepPowerAll[source]#

Bases: GaitBoutAggregateStep

Extract step power related measures.

Parameters:: data_set_ids – The data set ids that will be considered to extract step power measures.

__init__(data_set_ids, **kwargs)[source]#

Parameters:: data_set_ids (List[str]) –

class dispel.providers.generic.tasks.gait.core.ExtractStepRegularity[source]#

Bases: GaitBoutAggregateStep

Extract step regularity aggregate measures.

Parameters:: data_set_ids – The data set ids that will be considered to extract step regularity.

__init__(data_set_ids, **kwargs)[source]#

Parameters:: data_set_ids (List[str]) –

class dispel.providers.generic.tasks.gait.core.ExtractStrideRegularity[source]#

Bases: GaitBoutAggregateStep

Extract stride regularity aggregate measures.

Parameters:: data_set_ids – The data set ids that will be considered to extract stride regularity.

__init__(data_set_ids, **kwargs)[source]#

Parameters:: data_set_ids (List[str]) –

class dispel.providers.generic.tasks.gait.core.FootUsed[source]#

Bases: IntEnum

Information on the foot being used for step annotation.

LEFT = 1#

RIGHT = 2#

UNKNOWN = 0#

class dispel.providers.generic.tasks.gait.core.FormatWalkingBouts[source]#

Bases: TransformStep

Format fixed three seconds walking bouts into bouts of any size.

__init__()[source]#

class dispel.providers.generic.tasks.gait.core.GaitBoutAggregateStep[source]#

Bases: AggregateRawDataSetColumn

Base class for gait bouts agg raw data set measure extraction.

__init__(*args, **kwargs)[source]#

bout_strategy: BoutStrategy#

process_level(level, reading, **kwargs)[source]#

Overwrite process_level.

Parameters:

level (Level) –
reading (Reading) –

Return type:

Generator[ProcessingResult | ProcessingControlResult, None, None]

class dispel.providers.generic.tasks.gait.core.GaitBoutExtractStep[source]#

Bases: ExtractStep

Base class for gait bouts measure extraction.

__init__(*args, **kwargs)[source]#

bout_strategy: BoutStrategy#

process_level(level, reading, **kwargs)[source]#

Overwrite process level.

Parameters:

level (Level) –
reading (Reading) –

Return type:

Generator[ProcessingResult | ProcessingControlResult, None, None]

class dispel.providers.generic.tasks.gait.core.StepEvent[source]#

Bases: IntEnum

Generic events for step annotation.

FOOT_OFF = 3#

INITIAL_CONTACT = 1#

MID_SWING = 2#

UNKNOWN = 0#

class dispel.providers.generic.tasks.gait.core.TransformGaitRegularity[source]#

Bases: _TransformGaitRegularityBase

Transform gait regularity.

Create a data set with columns step_regularity, stride_regularity, and bout_id indicating the step and stride regularity per step and corresponding bout identifier.

definitions: List[RawDataValueDefinition] = [<RawDataValueDefinition: step_regularity (step regularity)>, <RawDataValueDefinition: stride_regularity (stride regularity)>, <RawDataValueDefinition: lag_step_regularity (lag step regularity)>, <RawDataValueDefinition: lag_stride_regularity (lag stride regularity)>, <RawDataValueDefinition: bout_id (bout id)>]#

class dispel.providers.generic.tasks.gait.core.TransformGaitRegularityWithoutBout[source]#

Bases: _TransformGaitRegularityBase

Transform gait regularity without bout.

Create a data set with two columns step_regularity and stride_regularity per step.

definitions: List[RawDataValueDefinition] = [<RawDataValueDefinition: step_regularity (step regularity)>, <RawDataValueDefinition: stride_regularity (stride regularity)>, <RawDataValueDefinition: lag_step_regularity (lag step regularity)>, <RawDataValueDefinition: lag_stride_regularity (lag stride regularity)>]#

class dispel.providers.generic.tasks.gait.core.TransformStepDetection[source]#

Bases: TransformStep

A transform step that creates a generic data set for step analysis.

Given the step detection algorithm specified through the method argument, e.g.: dispel.providers.generic.tasks.gait.lee.detect_steps(), the transform step create a generic data frame with annotated events as in Bourke et. al.

definitions: List[RawDataValueDefinition] = [<RawDataValueDefinition: foot (foot)>, <RawDataValueDefinition: event (event)>]#

class dispel.providers.generic.tasks.gait.core.TransformStepDuration[source]#

Bases: TransformStep

Transform step duration.

Create a data set with one column step_duration indicating the amount of time to perform each steps.

definitions: List[RawDataValueDefinition] = [<RawDataValueDefinition: step_duration (step duration, s)>, <RawDataValueDefinition: bout_id (bout id)>]#

get_new_data_set_id()[source]#

Overwrite new data set id.

Return type:: str

class dispel.providers.generic.tasks.gait.core.TransformStepDurationWithoutBout[source]#

Bases: TransformStep

Transform step duration without using walking bout.

Create a data set with one column step_duration indicating the amount of time to perform each steps.

definitions: List[RawDataValueDefinition] = [<RawDataValueDefinition: step_duration (step duration, s)>]#

get_new_data_set_id()[source]#

Overwrite new data set id.

Return type:: str

class dispel.providers.generic.tasks.gait.core.TransformStepVigor[source]#

Bases: TransformStep

Transform step vigor.

Create a data set with columns step_power, step_intensity, and bout_id indicating the power, intensity and respective bout identifier of each step.

__init__(component, *args, **kwargs)[source]#

Parameters:: component (str) –

definitions: List[RawDataValueDefinition] = [<RawDataValueDefinition: step_power (step power, m^2/s^3)>, <RawDataValueDefinition: step_intensity (step intensity, m/s^2)>, <RawDataValueDefinition: bout_id (bout id)>]#

get_new_data_set_id()[source]#

Overwrite new data set id.

Return type:: str

class dispel.providers.generic.tasks.gait.core.TransformStepVigorWithoutBout[source]#

Bases: TransformStep

Transform step vigor without bout.

Create a data set with two columns step_power and step_intensity indicating the step power and intensity, respectively for each step.

__init__(component, *args, **kwargs)[source]#

Parameters:: component (str) –

definitions: List[RawDataValueDefinition] = [<RawDataValueDefinition: step_power (step power, m^2/s^3)>, <RawDataValueDefinition: step_intensity (step intensity, m/s^2)>]#

get_new_data_set_id()[source]#

Overwrite new data set id.

Return type:: str

dispel.providers.generic.tasks.gait.core.compute_gait_regularity_multiple(acc, step_dataset, window_size_in_steps=14, window_step_in_steps=7, component='norm', bout_filter=False)[source]#

Compute the step and stride regularity.

The step and stride regularity is defined as the first and second peak of the unbiased normalized autocorrelation.

Parameters:

acc (DataFrame) –

A pandas DataFrame containing the at least one column with the intended
acceleration signal component to be used for step & stride regularity.
step_dataset (DataFrame) – A pandas series containing at the minimum the the times of the initial contacts (‘IC’ column).
window_size_in_steps (int) – The interval expressed in number of succeeding steps to be used to trim the signal to compute the regularity.
window_step_in_steps (int) – The step of the sliding window expressed in number of steps.
component (str) – A str indicating the column name of acc to be used.
bout_filter (bool) – A boolean indicating whether to take into account bouts in the transformation of the dataset.

Returns:

The step and stride regularity values.

Return type:

pandas.DataFrame

dispel.providers.generic.tasks.gait.core.compute_gait_regularity_single(data, find_peak_height_centile=80, find_peak_distance=0.3)[source]#

Compute gait regularity based on filtered signal.

The step and stride regularity is defined as the first and second peak of the unbiased normalized autocorrelation. In order to minimize the negative effect of local maxima we filter the accelerometer signal and detect the lags based on that, which we then apply in the autocorrelation of the non-filtered signal.

Parameters:

data (Series) – A pandas Series containing the accelerometer signal.
find_peak_height_centile (float) – Required height of peak as a percentile of the autocorrelation signal.
find_peak_distance (float) – Required minimal horizontal distance (>0) in seconds between neighbouring peaks.

Returns:

A Tuple containing the step and stride regularity and respective lags.

Return type:

Tuple[float, float, float, float]

References

dispel.providers.generic.tasks.gait.core.compute_step_intensity_single(acc_norm)[source]#

Compute the Step Intensity for a single step.

The step intensity is defined as the RMS of the magnitude of acceleration for the period of a single step.

Parameters:: acc_norm (Series) – A pandas series containing the magnitude of the acceleration.
Returns:: The step intensity.
Return type:: float

References

dispel.providers.generic.tasks.gait.core.compute_step_power_single(acc_norm)[source]#

Compute the Step Power for a single step.

The step power is defined as the integral of centered magnitude acceleration for the period of a single step.

Parameters:: acc_norm (Series) – A pandas series containing the magnitude of the acceleration.
Returns:: The step power.
Return type:: float

References

dispel.providers.generic.tasks.gait.core.compute_step_vigor_multiple(acc, step_dataset, component, bout_filter)[source]#

Compute the step vigor for multiple steps.

The step vigor is defined as the step power and step intensity properties.

Parameters:

acc (DataFrame) –

A pandas DataFrame containing the at least one column with the intended
acceleration signal component to be used for step power.
step_dataset (DataFrame) – A pandas series containing at the minimum the the times of the initial contacts.
component (str) – A str indicating the column name of acc to be used.
bout_filter (bool) – A boolean indicating whether to take into account bouts in the transformation of the dataset.

Returns:

The step power and intensity values.

Return type:

pandas.DataFrame

dispel.providers.generic.tasks.gait.core.find_peaks_in_local_slice(initial_peak_idx, autocorr_values, lag_values, slice_size=5)[source]#

Find peaks in local slice.

Parameters:

initial_peak_idx (ndarray) – A numpy.ndarray containing the initial guesses for the peak indices.
autocorr_values (ndarray) – The autocorrelation values.
lag_values (ndarray) – The value of the lags.
slice_size (int) – The size of the slice to look for peaks forward and backward in number of samples.

Returns:

Tuple[np.ndarray, np.ndarray]
A Tuple containing the refined peaks and lags.

Return type:

Tuple[ndarray, ndarray]

dispel.providers.generic.tasks.gait.core.format_walking_bouts(walking_segments, start_time_col='ts', end_time_col='ts', bool_col='detected_walking')[source]#

Group consecutive flags of same type in a single bin.

Parameters:

walking_segments (DataFrame) – a pd.DataFrame containing the flags
start_time_col (str) – the start time column name
end_time_col (str) – the end time column name
bool_col (str) – the name of the column to be looked for activity

Returns:

The dataframe with grouped consecutive flags of same type

Return type:

pandas.DataFrame

dispel.providers.generic.tasks.gait.core.get_mask_from_intervals(intervals, ts_index, bool_column='detected_walking', start_column='start_time', end_column='end_time')[source]#

Get a boolean mask indicating walking or not walking in a signal.

Parameters:

intervals (DataFrame) – pd.DataFrame: contains start_time and end_time for each interval/bout
ts_index (DatetimeIndex) – pd.DatetimeIndex: a datetime index of the signal to be masked
bool_column (str) – str indicating the column of boolean type to be used for the mask
start_column (str) – str indicating the column of the start time of the interval
end_column (str) – str indicating the column of the end time of the interval

Returns:

contains the boolean mask of len(timestamp_index)

Return type:

pandas.Series

dispel.providers.generic.tasks.gait.core.get_step_detector_from_data_set_ids(data_set_ids, position)[source]#

Get step detector from data_set_ids.

Parameters:

data_set_ids (str | List[str]) –
position (int) –

Return type:

str

dispel.providers.generic.tasks.gait.core.movement_detection(acc_ts_rotated_resampled_detrend)[source]#

Detect movement bouts.

Parameters:: acc_ts_rotated_resampled_detrend (DataFrame) – A pd.DataFrame containing the accelerometer signal rotated in a gravity based coordinate frame with x pointing towards gravity
Returns:: A pd.DataFrame containing the start/end timestamp, duration of each detected movement or non-movement bout
Return type:: pandas.DataFrame

dispel.providers.generic.tasks.gait.core.power_bout_div_steps(acc_magnitude, step_dataset, walking_bouts=None)[source]#

Compute the Step Power based on bout power divided by number of steps.

The step power is defined as the integral of centered magnitude acceleration divided by the number of steps. For more information please see the section 2.3 of [1]_.

Parameters:

acc_magnitude (Series) – A pandas series containing the magnitude of the acceleration.
step_dataset (DataFrame) – A pandas data frame containing the detected steps formatted as the output of TransformStepDetection.
walking_bouts (DataFrame | None) – A dataframe with the concatenated walking bouts.

Returns:

The step power.

Return type:

float

References

dispel.providers.generic.tasks.gait.core.remove_short_bouts(data, true_thres=3, false_thres=0.3, bool_col='detected_walking')[source]#

Remove short walking and non-walking bouts.

Parameters:

data (DataFrame) – A pd.DataFrame containing the intervals of the bouts with including at least bool_col and duration as columns
true_thres (float) – A float indicating the minimum duration of a true bout
false_thres (float) – A float indicating the minimum duration of a false bout
bool_col (str) – A ‘str’ containing the column name of the boolean

Returns:

A pd.DataFrame containing the filtered bout data satisfying the bout type and duration criteria.

Return type:

pandas.DataFrame

dispel.providers.generic.tasks.gait.core.step_count(data)[source]#

Count the number of steps.

Parameters:: data (DataFrame) – A pandas data frame containing the detected steps.
Returns:: The number of steps.
Return type:: int

dispel.providers.generic.tasks.gait.core.walking_detection_harmonic(vertical_acc, abs_thresh=0.01)[source]#

Detect walking bouts.

A sliding window of three seconds is moved through the time series of vertical acceleration to detect walking bouts. Walking is identified based on spectral measures computed on the temporal window. More specifically, walking is seen if the maximum amplitude of the power spectral density in the walking frequency band [0.6 Hz - 2.0 Hz] is more significant (and more distant than a given threshold) than the maximum amplitude of the power spectral density of the non-walking frequency band [0 - 0.6 Hz].

Parameters:

vertical_acc (DataFrame) – A smoothed version of the vertical acceleration typically obtained using the transform step ~AverageRollingVerticalAcceleration.
abs_thresh (float) – A threshold defining the minimum distance that should separate the amplitude of the walking harmonic from the amplitude of the non-walking harmonic.

Returns:

A data frame with the maximum walking and non-walking harmonic as well as the detected walking.

Return type:

pandas.DataFrame

Raises:

ValueError – Raises a value error if vertical_acc is not sampled at a constant sampling rate.

dispel.providers.generic.tasks.gait.core.walking_placement_detection(movement_bouts, placement_bouts, placement='belt')[source]#

Detect intersection of walking and intended placement.

Parameters:

movement_bouts (DataFrame) – A pd.DataFrame containing the intervals of the walking bouts
placement_bouts (DataFrame) – A pd.DataFrame containing the intervals of the placement bouts
placement (str) – A ‘str’ containing the label of the intended placement

Returns:

A pd.DataFrame containing the start/end timestamp, duration of each detected walking or no walking bout considering intended placement

Return type:

pandas.DataFrame

dispel.providers.generic.tasks.gait.core.walking_placement_no_turn_detection(movement_bouts, placement_bouts, turn_bouts, placement_label='belt')[source]#

Intersect walking bouts, placement and no turn constraints.

This function finds all the walking bout and apply a logical and with the mask of the walking bout, the placement detection that respects the label e.g.: ‘belt’ and the non-turn detection mask. A postprocessing step to remove short bouts and fill in gaps at the beginning and end of the signal is applied.

Parameters:

movement_bouts (DataFrame) – A pd.DataFrame containing the intervals of the walking bouts
placement_bouts (DataFrame) – A pd.DataFrame containing the intervals of the placement bouts
turn_bouts (DataFrame) – A pd.DataFrame containing the intervals of the turns
placement_label (str) – A ‘str’ containing the label of the intended placement

Returns:

A pd.DataFrame containing the start/end timestamp, duration of each detected walking or no walking bout considering intended placement

Return type:

pandas.DataFrame