Skip to content
Merged

Dev #85

Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .github/workflows/gh-pages.yml
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ jobs:
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install .[dev] mike
pip install .[dev] "mike<2.2"

# 4. Deploy MkDocs site with mike
- name: Deploy with mike 🚀
Expand Down
36 changes: 18 additions & 18 deletions examples/06_multimodal_capstone.ipynb

Large diffs are not rendered by default.

5 changes: 4 additions & 1 deletion examples/requirements.txt
Original file line number Diff line number Diff line change
@@ -1,2 +1,5 @@
ipywidgets
tqdm
tqdm
matplotlib
opencv-python
pandas
9 changes: 5 additions & 4 deletions pyeyesweb/low_level/direction_change.py
Original file line number Diff line number Diff line change
Expand Up @@ -125,14 +125,15 @@ def _polygon_area(self, pos: np.ndarray) -> float:
# Close the loop
closed_polygon = np.vstack([subset, subset[0]])

# NumPy 2 only supports 3D cross products: lift 2D points to z=0
if closed_polygon.shape[1] == 2:
closed_polygon = np.pad(closed_polygon, ((0, 0), (0, 1)))

# Shoelace-style cross product area computation
cross_products = np.cross(closed_polygon[:-1], closed_polygon[1:])
area_vector = np.sum(cross_products, axis=0) / 2.0

# Handle both 2D (scalar return) and 3D (vector return) area magnitudes
area = np.linalg.norm(area_vector) if np.ndim(area_vector) > 0 else np.abs(area_vector)

return float(area)
return float(np.linalg.norm(area_vector))

def compute(self, window_data: np.ndarray) -> DirectionChangeResult:
"""Compute the Direction Change over a temporal window.
Expand Down
122 changes: 63 additions & 59 deletions pyeyesweb/low_level/smoothness.py
Original file line number Diff line number Diff line change
@@ -1,72 +1,60 @@
from dataclasses import dataclass
from typing import Literal, List, Optional
import numpy as np
from scipy.signal import butter, filtfilt

from pyeyesweb.data_models.base import DynamicFeature
from pyeyesweb.data_models.results import FeatureResult
from pyeyesweb.utils.signal_processing import apply_savgol_filter
from pyeyesweb.utils.math_utils import compute_sparc, compute_jerk_rms
from pyeyesweb.utils.math_utils import (
compute_sparc,
compute_jerk_rms,
compute_ldlj,
compute_sample_entropy,
compute_harmonicity_index,
compute_submovement_count
)
from pyeyesweb.utils.validators import validate_numeric, validate_boolean, validate_string


@dataclass(slots=True)
class SmoothnessResult(FeatureResult):
"""Output contract for Smoothness metrics.

Attributes
----------
sparc : float, optional
The computed SPARC metric representing spectral arc length.
jerk_rms : float, optional
The compute root mean square of jerk.
"""
"""Output contract for Smoothness metrics."""
sparc: Optional[float] = None
jerk_rms: Optional[float] = None
ldlj_v: Optional[float] = None
ldlj_a: Optional[float] = None
samp_en: Optional[float] = None
harmonicity: Optional[float] = None
n_submovements: Optional[int] = None


class Smoothness(DynamicFeature):
"""Calculates movement smoothness metrics from a 1D speed profile.

!!! tip
You can calculate smoothness via Spectral Arc Length (SPARC) or Jerk RMS.

Read more in the [User Guide](../../user_guide/theoretical_framework/low_level/smoothness.md).

Parameters
----------
rate_hz : float, optional
Sampling rate in Hz. Defaults to `50.0`.
use_filter : bool, optional
Whether to apply Savitzky-Golay filtering. Defaults to `True`.
metrics : list of {'sparc', 'jerk_rms'}, optional
Metrics to calculate. Defaults to all allowed metrics.
sparc_amplitude_threshold : float, optional
Amplitude threshold for SPARC. Defaults to `0.05`.
sparc_min_fc : float, optional
Minimum cutoff frequency for SPARC. Defaults to `2.0`.
sparc_max_fc : float, optional
Maximum cutoff frequency for SPARC. Defaults to `20.0`.
"""

_ALLOWED_METRICS = ["sparc", "jerk_rms"]
"""Calculates movement smoothness and complexity metrics from a 1D speed profile."""

_ALLOWED_METRICS = [
"sparc", "jerk_rms", "ldlj_v", "ldlj_a",
"samp_en", "harmonicity", "n_submovements"
]

def __init__(
self,
rate_hz: float = 50.0,
use_filter: bool = True,
metrics: List[Literal["sparc", "jerk_rms"]] = None,
metrics: List[Literal["sparc", "jerk_rms", "ldlj_v", "ldlj_a", "samp_en", "harmonicity", "n_submovements"]] = None,
sparc_amplitude_threshold: float = 0.05,
sparc_min_fc: float = 2.0,
sparc_max_fc: float = 20.0
sparc_max_fc: float = 20.0,
min_speed_threshold: float = 30.0
):
super().__init__()
# Initializing through setters natively routes the values to the validators
self.rate_hz = rate_hz
self.use_filter = use_filter

self.sparc_min_fc = sparc_min_fc
self.sparc_max_fc = sparc_max_fc
self.sparc_threshold = sparc_amplitude_threshold
self.min_speed_threshold = min_speed_threshold

self.metrics = metrics

Expand Down Expand Up @@ -108,7 +96,6 @@ def sparc_max_fc(self) -> float:

@sparc_max_fc.setter
def sparc_max_fc(self, value: float):
# We ensure that this validator uses the dynamic minimum
min_fc = getattr(self, '_sparc_min_fc', 0.1)
self._sparc_max_fc = validate_numeric(value, 'sparc_max_fc', min_val=min_fc)

Expand All @@ -122,42 +109,37 @@ def metrics(self, value: Optional[List[str]]):
self._metrics = [validate_string(m, self._ALLOWED_METRICS) for m in target_metrics]

def _filter_signal(self, signal: np.ndarray) -> np.ndarray:
"""Applies Savitzky-Golay filter if enabled."""
if not self.use_filter:
if not self.use_filter or len(signal) < 15:
return signal
return apply_savgol_filter(signal, self.rate_hz)

cutoff = min(10.0, (self.rate_hz / 2.0) - 1.0)
b, a = butter(4, cutoff / (self.rate_hz / 2.0), btype='low')
return filtfilt(b, a, signal)

def compute(self, window_data: np.ndarray) -> SmoothnessResult:
"""Executes smoothness calculation on the speed profile.

Parameters
----------
window_data : numpy.ndarray
A 1D array representing the speed profile within the window.

Returns
-------
SmoothnessResult
The computed smoothness metrics.
"""
if window_data.size != window_data.shape[0]:
raise ValueError("Smoothness expects a 1D speed profile.")

speed_profile = window_data.ravel()

# Minimum sample threshold for FFT
if len(speed_profile) < 10:
return SmoothnessResult(is_valid=False)

# 1. Preprocessing (Filtering)
# Zero-velocity check
# if np.mean(speed_profile) < self.min_speed_threshold or np.max(speed_profile) < (self.min_speed_threshold * 1.5):
# return SmoothnessResult(is_valid=False)

filtered_speed = self._filter_signal(speed_profile)

# 2. Metrics calculation
sparc_val = None
jerk_val = None
ldlj_v_val = None
ldlj_a_val = None
samp_en_val = None
harmonicity_val = None
n_sub_val = None

if "sparc" in self.metrics:
# Pass custom parameters to the function in math_utils
sparc_val = float(compute_sparc(
filtered_speed,
rate_hz=self.rate_hz,
Expand All @@ -167,7 +149,29 @@ def compute(self, window_data: np.ndarray) -> SmoothnessResult:
))

if "jerk_rms" in self.metrics:
# Standard Jerk RMS calculation from velocity
jerk_val = float(compute_jerk_rms(filtered_speed, self.rate_hz, signal_type='velocity'))

return SmoothnessResult(sparc=sparc_val, jerk_rms=jerk_val)
if "ldlj_v" in self.metrics:
ldlj_v_val = float(compute_ldlj(filtered_speed, self.rate_hz, signal_type='velocity'))

if "ldlj_a" in self.metrics:
ldlj_a_val = float(compute_ldlj(filtered_speed, self.rate_hz, signal_type='acceleration'))

if "samp_en" in self.metrics:
samp_en_val = float(compute_sample_entropy(filtered_speed))

if "harmonicity" in self.metrics:
harmonicity_val = float(compute_harmonicity_index(filtered_speed, self.rate_hz))

if "n_submovements" in self.metrics:
n_sub_val = int(compute_submovement_count(filtered_speed, self.rate_hz))

return SmoothnessResult(
sparc=sparc_val,
jerk_rms=jerk_val,
ldlj_v=ldlj_v_val,
ldlj_a=ldlj_a_val,
samp_en=samp_en_val,
harmonicity=harmonicity_val,
n_submovements=n_sub_val
)
Loading
Loading