Merge pull request #60 from AlzhraaIbrahim/event_detection_for_ic_stride

Event detection for ic stride

Author: AKuederle

CHANGELOG.md

@@ -10,6 +10,22 @@ project.
 
 ## [2.6.0] - Unreleased
 
+### Scientific Changes
+
+- The TC detection within FilteredRampp event detection now uses the filtered gyro signal instead of the raw gyro 
+  signal.
+  We saw in a couple of patients that they had a lot of high frequency noise in the gyro signal, which affected the
+  TC detection.
+  For existing cases, this should change the position of the TC event slightly (approximately 1-2 samples).
+  (https://github.com/mad-lab-fau/gaitmap/pull/60)
+
+### Added
+
+- Both variants of the Rampp event detection now support either IC-IC strides or "segmented" strides as inputs.
+  This allows the use of strides that are pre-segmented by other algorithms or by a reference system as input to the
+  rest of the pipeline.
+  The algorithm still searches for a proper IC in the input data within a search window around the provided IC.
+  (https://github.com/mad-lab-fau/gaitmap/pull/60)
 - Dataset checks can now optionally check magnetometer data (https://github.com/mad-lab-fau/gaitmap/pull/73)
 - Dataset rotations can now rotate magnetometer data (https://github.com/mad-lab-fau/gaitmap/pull/73)
 - Added an option to the madgwick algorithm to use the algorithm version that also uses the magnetometer for correction.

example_data/imu_sample_ic_stride.csv

@@ -0,0 +1,15 @@
+s_id,foot,start,end,gsd_id
+0,left,696,812,1
+1,left,812,926,1
+2,left,1531,1647,1
+3,left,1647,1765,1
+4,left,2510,2627,1
+5,left,2627,2746,1
+6,left,3361,3479,1
+7,right,757,870,1
+8,right,870,986,1
+9,right,1591,1709,1
+10,right,2451,2570,1
+11,right,2570,2686,1
+12,right,3306,3424,1
+13,right,3424,3544,1

example_data/stride_borders_sample_ic_stride.csv

@@ -66,10 +66,10 @@
 min_vel_events_left.head()
 
 # %%
-# As a secondary output we get the `segmented_event_list_`, which holds the same event information than the
+# As a secondary output we get the `annotated_original_event_list_`, which holds the same event information than the
 # `min_vel_event_list_`, but the start and the end of each stride are unchanged compared to the input.
 # This also means that no strides are removed due to the conversion step explained below.
-segmented_events_left = ed.segmented_event_list_["left_sensor"]
+segmented_events_left = ed.annotated_original_event_list_["left_sensor"]
 print(f"Gait events for {len(segmented_events_left)} segmented strides were detected.")
 segmented_events_left.head()
 
@@ -202,7 +202,7 @@
 # This is required due to the shift of stride borders between the `stride_list` and the `min_vel_event_list_`.
 # Thus, the dropped first segmented stride of a continuous sequence only provides a pre_ic and a min_vel sample for
 # the first stride in the `min_vel_event_list_`.
-# Therefore, the `min_vel_event_list_` list has one stride less than the `segmented_event_list_`.
+# Therefore, the `min_vel_event_list_` list has one stride less than the `annotated_original_event_list_`.
 
 ed2 = HerzerEventDetection()
 segmented_stride_list = stride_list["left_sensor"].iloc[[11, 12, 13, 14, 15, 16]]

examples/event_detection/herzer_event_detection.py

@@ -84,10 +84,10 @@
 min_vel_events_left.head()
 
 # %%
-# As a secondary output we get the `segmented_event_list_`, which holds the same event information than the
+# As a secondary output we get the `annotated_original_event_list_`, which holds the same event information than the
 # `min_vel_event_list_`, but the start and the end of each stride are unchanged compared to the input.
 # This also means that no strides are removed due to the conversion step explained below.
-segmented_events_left = ed.segmented_event_list_["left_sensor"]
+segmented_events_left = ed.annotated_original_event_list_["left_sensor"]
 print(f"Gait events for {len(segmented_events_left)} segmented strides were detected.")
 segmented_events_left.head()
 
@@ -231,7 +231,7 @@
 # This is required due to the shift of stride borders between the `stride_list` and the `min_vel_event_list_`.
 # Thus, the dropped first segmented stride of a continuous sequence only provides a pre_ic and a min_vel sample for
 # the first stride in the `min_vel_event_list_`.
-# Therefore, the `min_vel_event_list_` list has one stride less than the `segmented_event_list_`.
+# Therefore, the `min_vel_event_list_` list has one stride less than the `annotated_original_event_list_`.
 from gaitmap.event_detection import RamppEventDetection
 
 ed2 = RamppEventDetection()
@@ -334,7 +334,7 @@
 min_vel_events_left = edfilt.min_vel_event_list_["left_sensor"]
 print(f"Gait events for {len(min_vel_events_left)} min_vel strides using the filtered version were detected.")
 min_vel_events_left.head()
-segmented_events_left = edfilt.segmented_event_list_["left_sensor"]
+segmented_events_left = edfilt.annotated_original_event_list_["left_sensor"]
 print(f"Gait events for {len(segmented_events_left)} segmented strides using the filtered version were detected.")
 segmented_events_left.head()
 fig, (ax1, ax2) = plt.subplots(2, sharex=True, figsize=(10, 5))

examples/event_detection/rampp_event_detection.py

@@ -1,12 +1,13 @@
 """Mixin for event detection algorithms that work similar to Rampp et al."""
 
+import warnings
 from typing import Any, Callable, Optional, Union
 
 import numpy as np
 import pandas as pd
 from joblib import Memory
 from numpy.linalg import norm
-from typing_extensions import Self
+from typing_extensions import Literal, Self
 
 from gaitmap.utils._algo_helper import invert_result_dictionary, set_params_from_dict
 from gaitmap.utils._types import _Hashable
@@ -22,7 +23,7 @@
 )
 from gaitmap.utils.exceptions import ValidationError
 from gaitmap.utils.stride_list_conversion import (
-    _segmented_stride_list_to_min_vel_single_sensor,
+    _stride_list_to_min_vel_single_sensor,
     enforce_stride_list_consistency,
 )
 
@@ -33,21 +34,24 @@ class _EventDetectionMixin:
     detect_only: Optional[tuple[str, ...]]
 
     min_vel_event_list_: Optional[Union[pd.DataFrame, dict[str, pd.DataFrame]]]
-    segmented_event_list_: Optional[Union[pd.DataFrame, dict[str, pd.DataFrame]]]
+    annotated_original_event_list_: Optional[Union[pd.DataFrame, dict[str, pd.DataFrame]]]
 
     data: SensorData
     sampling_rate_hz: float
     stride_list: pd.DataFrame
+    input_stride_type: Literal["segmented", "ic"]
 
     def __init__(
         self,
         memory: Optional[Memory] = None,
         enforce_consistency: bool = True,
         detect_only: Optional[tuple[str, ...]] = None,
-    ) -> None:
+        input_stride_type: Literal["segmented", "ic"] = "segmented",
+    ):
         self.memory = memory
         self.enforce_consistency = enforce_consistency
         self.detect_only = detect_only
+        self.input_stride_type = input_stride_type
 
     def detect(self, data: SensorData, stride_list: StrideList, *, sampling_rate_hz: float) -> Self:
         """Find gait events in data within strides provided by stride_list.
@@ -121,50 +125,62 @@ def _detect_single_dataset(
         # find events in all segments
         event_detection_func = self._select_all_event_detection_method()
         event_detection_func = memory.cache(event_detection_func)
-        ic, tc, min_vel = event_detection_func(gyr, acc, stride_list, events=events, **detect_kwargs)
+        ic, tc, min_vel = event_detection_func(
+            gyr, acc, stride_list, events=events, input_stride_type=self.input_stride_type, **detect_kwargs
+        )
 
         # build first dict / df based on segment start and end
-        segmented_event_list = {
+        annotated_original_event_list = {
             "s_id": stride_list.index,
             "start": stride_list["start"],
             "end": stride_list["end"],
         }
         for event, event_list in zip(("ic", "tc", "min_vel"), (ic, tc, min_vel)):
             if event in events:
-                segmented_event_list[event] = event_list
-
-        segmented_event_list = pd.DataFrame(segmented_event_list).set_index("s_id")
-
+                annotated_original_event_list[event] = event_list
+        annotated_original_event_list = pd.DataFrame(annotated_original_event_list).set_index("s_id")
         if self.enforce_consistency:
             # check for consistency, remove inconsistent strides
-            segmented_event_list, _ = enforce_stride_list_consistency(
-                segmented_event_list, stride_type="segmented", check_stride_list=False
+            annotated_original_event_list, _ = enforce_stride_list_consistency(
+                annotated_original_event_list, input_stride_type=self.input_stride_type, check_stride_list=False
             )
 
         if "min_vel" not in events or self.enforce_consistency is False:
             # do not set min_vel_event_list_ if consistency is not enforced as it would be completely scrambled
             # and can not be used for anything anyway
-            return {"segmented_event_list": segmented_event_list}
+            return {"annotated_original_event_list": annotated_original_event_list}
 
         # convert to min_vel event list
-        min_vel_event_list, _ = _segmented_stride_list_to_min_vel_single_sensor(
-            segmented_event_list, target_stride_type="min_vel"
+        min_vel_event_list, _ = _stride_list_to_min_vel_single_sensor(
+            annotated_original_event_list, source_stride_type=self.input_stride_type, target_stride_type="min_vel"
         )
 
         output_order = [c for c in ["start", "end", "ic", "tc", "min_vel", "pre_ic"] if c in min_vel_event_list.columns]
 
         # We enforce consistency again here, as a valid segmented stride list does not necessarily result in a valid
         # min_vel stride list
         min_vel_event_list, _ = enforce_stride_list_consistency(
-            min_vel_event_list[output_order], stride_type="min_vel", check_stride_list=False
+            min_vel_event_list[output_order], input_stride_type="min_vel", check_stride_list=False
         )
 
-        return {"min_vel_event_list": min_vel_event_list, "segmented_event_list": segmented_event_list}
+        return {
+            "min_vel_event_list": min_vel_event_list,
+            "annotated_original_event_list": annotated_original_event_list,
+        }
+
+    @property
+    def segmented_event_list_(self) -> Optional[Union[pd.DataFrame, dict[str, pd.DataFrame]]]:
+        warnings.warn(
+            "`segmented_event_list_` is deprecated and will be removed in a future version. "
+            "Use `annotated_original_event_list_` instead.",
+            DeprecationWarning,
+        )
+        return self.annotated_original_event_list_
 
     def _select_all_event_detection_method(self) -> Callable:
         """Select the function to calculate the all events.
 
-        This is separate method to make it easy to overwrite by a subclass.
+        This is a separate method to make it easy to overwrite by a subclass.
         """
         raise NotImplementedError()
 

gaitmap/_event_detection_common/_event_detection_mixin.py

@@ -7,6 +7,7 @@
 from joblib import Memory
 from scipy import signal
 from tpcp import cf
+from typing_extensions import Literal
 
 from gaitmap._event_detection_common._event_detection_mixin import _detect_min_vel_gyr_energy, _EventDetectionMixin
 from gaitmap.base import BaseEventDetection
@@ -54,7 +55,9 @@ class HerzerEventDetection(_EventDetectionMixin, BaseEventDetection):
         By default, all events ("ic", "tc", "min_vel") are detected.
         If `min_vel` is not detected, the `min_vel_event_list_` output will not be available.
         If "ic" is not detected, the `pre_ic` will also not be available in the output.
-
+    input_stride_type
+        Only segmented strides are supported by this method.
+        Hence, this parameter is set to "segmented" by default and changing it will raise an error.
 
     Attributes
     ----------
@@ -65,11 +68,13 @@ class HerzerEventDetection(_EventDetectionMixin, BaseEventDetection):
         corresponds to the min_vel sample of the subsequent stride.
         Strides for which no valid events could be found are removed.
         Additional strides might have been removed due to the conversion from segmented to min_vel strides.
-    segmented_event_list_ : A stride list or dictionary with such values
+    annotated_original_event_list_ : A stride list or dictionary with such values
         The result of the `detect` method holding all temporal gait events and start / end of all strides.
         This version of the results has the same stride borders than the input `stride_list` and has additional columns
         for all the detected events.
         Strides for which no valid events could be found are removed.
+    segmented_event_list_ :
+        Deprecated, use `annotated_original_event_list_` instead.
 
 
     Other Parameters
@@ -89,7 +94,7 @@ class HerzerEventDetection(_EventDetectionMixin, BaseEventDetection):
 
     >>> event_detection = HerzerEventDetection()
     >>> event_detection.detect(data=data, stride_list=stride_list, sampling_rate_hz=204.8)
-    >>> event_detection.segmented_event_list_
+    >>> event_detection.annotated_original_event_list_
           start    end       ic       tc  min_vel
     s_id
     0     48304  48558  48382.0  48304.0  48479.0
@@ -132,6 +137,9 @@ class HerzerEventDetection(_EventDetectionMixin, BaseEventDetection):
         The window size can be adjusted via the `min_vel_search_win_size_ms` parameter.
         This approach is identical to [1]_.
 
+    The :func:`~gaitmap.event_detection.HerzerEventDetection.detect` method is implemented only for "segmented" stride
+    type
+
     The :func:`~gaitmap.event_detection.HerzerEventDetection.detect` method provides a stride list `min_vel_event_list`
     with the gait events mentioned above and additionally `start` and `end` of each stride, which are aligned to the
     `min_vel` samples.
@@ -141,15 +149,15 @@ class HerzerEventDetection(_EventDetectionMixin, BaseEventDetection):
     the stride list.
     This function should NOT be used, since the detection of the min_vel gait events has not been validated
     for stair ambulation.
-    Please refer to the `segmented_event_list_` instead.
+    Please refer to the `annotated_original_event_list_` instead.
 
     The :class:`~gaitmap.event_detection.HerzerEventDetection` includes a consistency check that is enabled by default.
     The gait events within one stride provided by the `stride_list` must occur in the expected order.
     Any stride where the gait events are detected in a different order or are not detected at all is dropped!
     For more infos on this see :func:`~gaitmap.utils.stride_list_conversion.enforce_stride_list_consistency`.
     If you wish to disable this consistency check, set `enforce_consistency` to False.
-    In this case, the attribute `min_vel_event_list_` will not be set, but you can use `segmented_event_list_` to get
-    all detected events for the exact stride list that was used as input.
+    In this case, the attribute `min_vel_event_list_` will not be set, but you can use `annotated_original_event_list_`
+    to get all detected events for the exact stride list that was used as input.
     Note, that this list might contain NaN for some events.
 
     Furthermore, during the conversion from the segmented stride list to the "min_vel" stride list, breaks in
@@ -188,6 +196,7 @@ class HerzerEventDetection(_EventDetectionMixin, BaseEventDetection):
     ic_lowpass_filter: BaseFilter
     memory: Optional[Memory]
     enforce_consistency: bool
+    input_stride_type: Literal["segmented"]
 
     def __init__(
         self,
@@ -198,12 +207,18 @@ def __init__(
         memory: Optional[Memory] = None,
         enforce_consistency: bool = True,
         detect_only: Optional[tuple[str, ...]] = None,
-    ) -> None:
+        input_stride_type: Literal["segmented"] = "segmented",
+    ):
         self.min_vel_search_win_size_ms = min_vel_search_win_size_ms
         self.mid_swing_peak_prominence = mid_swing_peak_prominence
         self.mid_swing_n_considered_peaks = mid_swing_n_considered_peaks
         self.ic_lowpass_filter = ic_lowpass_filter
-        super().__init__(memory=memory, enforce_consistency=enforce_consistency, detect_only=detect_only)
+        super().__init__(
+            memory=memory,
+            enforce_consistency=enforce_consistency,
+            detect_only=detect_only,
+            input_stride_type=input_stride_type,
+        )
 
     def _get_detect_kwargs(self) -> dict[str, int]:
         min_vel_search_win_size = int(self.min_vel_search_win_size_ms / 1000 * self.sampling_rate_hz)
@@ -234,8 +249,11 @@ def _find_all_events(
     mid_swing_n_considered_peaks: int,
     ic_lowpass_filter: BaseFilter,
     sampling_rate_hz: float,
+    input_stride_type: Literal["segmented"],
 ) -> tuple[Optional[np.ndarray], Optional[np.ndarray], Optional[np.ndarray]]:
     """Find events in provided data by looping over single strides."""
+    if input_stride_type != "segmented":
+        raise NotImplementedError("This method support only segmented stride type")
     gyr_ml = gyr["gyr_ml"].to_numpy()
     gyr = gyr.to_numpy()
     # inverting acc, as this algorithm was developed assuming a flipped axis like the original Rampp algorithm

gaitmap/event_detection/_herzer_event_detection.py

@@ -83,6 +83,19 @@ def get_healthy_example_imu_data():
     return data
 
 
+def get_healthy_example_imu_data_ic_stride():
+    """Get example IMU data from a healthy subject doing a 4x10 gait test.
+
+    The sampling rate is 102.4 Hz
+    """
+    test_data_path = _get_data("imu_sample_ic_stride.csv")
+    data = pd.read_csv(test_data_path, header=[0, 1], index_col=0)
+
+    # Get index in seconds
+    data.index /= 102.4
+    return data
+
+
 def get_ms_example_imu_data():
     """Get example IMU data from a MS subject performing a longer uninterrupted walking sequence.
 
@@ -130,6 +143,24 @@ def get_healthy_example_stride_borders():
     return data
 
 
+def get_healthy_example_stride_borders_ic_stride():
+    """Get hand labeled stride borders for :func:`get_healthy_example_imu_data_ic_stride`.
+
+    The stride borders are obtained from mocap where each stride starts with initial contact.
+    """
+    test_data_path = _get_data("stride_borders_sample_ic_stride.csv")
+    data = pd.read_csv(test_data_path, header=0)
+
+    # Convert to dict with sensor name as key.
+    # Sensor name here is derived from the foot. In the real pipeline that would be provided to the algo.
+    data["sensor"] = data["foot"] + "_sensor"
+    data = data.set_index("sensor")
+    data = data.groupby(level=0)
+    data = {k: v.reset_index(drop=True) for k, v in data}
+
+    return data
+
+
 def get_healthy_example_mocap_data():
     """Get 3D Mocap information of the foot synchronised with :func:`get_healthy_example_imu_data`.