WbAssembly#

class mobgap.wba.WbAssembly( rules: list[tuple[str, BaseWbCriteria]] | None = cf([('min_strides', NStridesCriteria(min_strides=4, min_strides_left=3, min_strides_right=3)), ('max_break', MaxBreakCriteria(consider_end_as_break=True, max_break_s=3, remove_last_ic=False))]), )[source]#

Assembles strides into walking bouts based on a set of criteria.

This method uses a two-step approach. First, it iterates through the list of strides and adds them to a preliminary walking bout until a termination is reached. These preliminary walking bouts are then checked against a set of inclusion rules. See Notes sections for some more details.

Parameters:

rules: The rules that are used to assemble the walking bouts. They need to be provided as a list of tuples, where the first element is the name of the rule and the second is an instance of a subclass of BaseWBCriteria. If None, no rules are applied and all strides are returned as a single WB.

Other Parameters:

filtered_stride_list: The stride list provided to the assemble method.
sampling_rate_hz: The sampling rate provided to the assemble method.

Attributes:

annotated_stride_list_: A dataframe containing all strides that are considered that are part of a WB. The dataframe has a multi-index where the first level is the wb_id and the second level is the stride id. This dataframe can be used to calculate statistics per WB by grouping by the wb_id.
wbs_: A dictionary containing all walking bouts that were assembled. The keys are the ids of the walking bouts. The values are dataframes containing the strides that are part of the respective walking bout.
wb_meta_parameters_: A dataframe containing the general parameters per WB. This includes the start, the end, the number of strides and the duration of the WB.
excluded_stride_list_: A dataframe containing all strides that are considered invalid, as they are not part of a WB. This can happen, because strides were part of a preliminary WB that was later discarded or because they were never part of a WB, which can happen when a termination rule moves the start or end of a WB.
excluded_wbs_: A dictionary containing all walking bouts that were discarded.
termination_reasons_: A dataframe containing the reason why a WB was terminated. The dataframe has two columns: rule_name and rule_obj corresponding to the values in the rules parameter. These rules are match to the rule tuples provided in the rules parameter. The only exception is the end_of_list rule, which is used when the end of the stride list is reached and no other rule terminated the WB. This dataframe contains all WBs, including the ones that were discarded.
exclusion_reasons_: A dataframe containing the reason why a WB was terminated. The dataframe has two columns: rule_name and rule_obj corresponding to the values in the rules parameter. These rules are match to the rule tuples provided in the rules parameter. The dataframe only contains WBs that were discarded.
stride_exclusion_reasons_: A dataframe containing the reason why a WB was terminated. The dataframe has the stride id as index and two columns: rule_name and rule_obj corresponding to the values in the rules parameter. Note, that this only contains strides that were never part of any WB. Strides that were part of a WB that was later discarded by an Inclusion Rule are not listed here.

Notes

Each rule can act as a termination rule and an inclusion rule using the check_wb_start_end and check_include methods respectively. Inclusion rules are rather simple, as they only need to return a boolean value and can perform a simple check on the stride list or a preliminary WB. Termination rules are more complex, as they need to be able to not just terminate a WB, but also adjust the start and end of the WB. For example, the break rule does not just stop a WB, when there is a gap between strides, it also removes the last stride of the WB, as this is not considered a valid stride based on the Mobilise-D definition. This means, it must also signal that the last stride should be removed. This is done by providing all strides to the check_wb_start_end method and the current start and end of the WB as defined by all previous rules. The method can then adjust the start and end of the WB to implement arbitrary complex rules. At each step, the rule that would result in the shortest WB is chosen (i.e. the latest start and the earliest end).

For the currently implemented rules, a lot of this complexity is not needed. However, the current approach was developed as a general framework to solve these types of grouping issues. More complex rules where implemented, but ultimately not used, as they were not needed for the Mobilise-D pipeline. However, we still left the basic framework in place to allow for more complex rules to be implemented in the future or on the user side.

Methods

`assemble`(filtered_stride_list, *, ...)	Assemble the walking bouts based on the pre-filtered stride list.
`clone`()	Create a new instance of the class with all parameters copied over.
`get_params`([deep])	Get parameters for this algorithm.
`set_params`(**params)	Set the parameters of this Algorithm.

PredefinedParameters

__init__( rules: list[tuple[str, BaseWbCriteria]] | None = cf([('min_strides', NStridesCriteria(min_strides=4, min_strides_left=3, min_strides_right=3)), ('max_break', MaxBreakCriteria(consider_end_as_break=True, max_break_s=3, remove_last_ic=False))]), ) → None[source]#

assemble( filtered_stride_list: DataFrame, *, sampling_rate_hz: float, ) → Self[source]#

Assemble the walking bouts based on the pre-filtered stride list.

Parameters:

filtered_stride_list: The list of valid strides that should be used to assemble final WBs. Usually this is the output of the StrideSelection. We expect at least to have a start and end column. We assume that the start and end values are in samples and can be converted to seconds using the sampling_rate_hz parameter. Additional required columns depend on the rules that are used for aggregation.
sampling_rate_hz: The sampling rate of the data in Hz. This is used to potentially convert the start and end values of a stride to seconds, assuming that they are in samples. If this is not the case and the value are already in seconds, sampling_rate_hz should set to 1.

Returns:

self: The instance itself with the result parameters set.

clone() → Self[source]#

Create a new instance of the class with all parameters copied over.

This will create a new instance of the class itself and all nested objects

get_params(deep: bool = True) → dict[str, Any][source]#

Get parameters for this algorithm.

Parameters:

deep: Only relevant if object contains nested algorithm objects. If this is the case and deep is True, the params of these nested objects are included in the output using a prefix like nested_object_name__ (Note the two “_” at the end)

Returns: