Skip to content
Merged
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
18 changes: 11 additions & 7 deletions src/jabs/feature_extraction/window_operations/window_stats.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,29 +14,33 @@ def pad_sliding_window(arr: np.ndarray, window: int, pad_const: float | None = N
Returns:
an unmodifiable 2d view of the input array where the first axis is time and the second axis is the window. Note that typical usage will use summary stats with axis=1.
"""
if pad_const:
if pad_const is not None:
arr_ext = np.concatenate([np.full(window, pad_const), arr, np.full(window, pad_const)])
else:
arr_ext = np.concatenate([np.full(window, arr[0]), arr, np.full(window, arr[-1])])
return np.lib.stride_tricks.sliding_window_view(arr_ext, window_shape=window * 2 + 1)


def get_window_masks(sliding_window_view: np.ndarray, const: float) -> np.ndarray:
"""Creates a mask for invalid values in a sliding window matrix.
"""Creates a mask identifying the usable values in a sliding window matrix.

Args:
sliding_window_view: sliding window matrix from `pad_sliding_window`
const: constant pad value

Returns:
matrix describing valid (0) and invalid (1) window values
boolean matrix that is True for valid window values and False for values
equal to the pad constant. Rows where every value is the pad constant are
returned as all True, since masking them entirely would leave the
downstream reduction with no data to operate on.
"""
if np.isnan(const):
window_masks = ~np.isnan(sliding_window_view)
else:
window_masks = sliding_window_view != const
for no_data_row in np.where(np.all(~window_masks, axis=1)):
window_masks[no_data_row] = True

no_data_rows = np.all(~window_masks, axis=1)
window_masks[no_data_rows] = True

return window_masks

Expand Down Expand Up @@ -165,14 +169,14 @@ def window_skew(values: np.ndarray, window: int) -> np.ndarray:


def window_min(values: np.ndarray, window: int) -> np.ndarray:
"""Calculates a masked maximum of a window
"""Calculates a masked minimum of a window

Args:
values: 1d np.ndarray of values
window: window size

Returns:
sliding window maximum values
sliding window minimum values
"""
window_values = pad_sliding_window(values, window, pad_const=np.nan)
window_masks = get_window_masks(window_values, np.nan)
Expand Down
104 changes: 104 additions & 0 deletions tests/feature_extraction/window_operations/test_window_stats.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,104 @@
"""Unit tests for the sliding window statistics helpers."""

import numpy as np
import pytest

from jabs.feature_extraction.window_operations import window_stats


def test_pad_sliding_window_shape() -> None:
"""The view has one row per input frame and 2 * window + 1 columns."""
values = np.arange(5, dtype=np.float64)
view = window_stats.pad_sliding_window(values, window=2, pad_const=np.nan)
assert view.shape == (values.size, 2 * 2 + 1)


def test_pad_sliding_window_edge_padding() -> None:
"""A None pad constant repeats the first and last values at the edges."""
values = np.array([3.0, 4.0, 5.0])
view = window_stats.pad_sliding_window(values, window=2, pad_const=None)
np.testing.assert_array_equal(view[0], [3.0, 3.0, 3.0, 4.0, 5.0])
np.testing.assert_array_equal(view[-1], [3.0, 4.0, 5.0, 5.0, 5.0])


@pytest.mark.parametrize("pad_const", [0.0, -1.0, np.nan], ids=["zero", "negative", "nan"])
def test_pad_sliding_window_uses_constant_padding(pad_const: float) -> None:
"""Any non-None pad constant is used verbatim, including a falsy 0.0."""
values = np.array([3.0, 4.0, 5.0])
view = window_stats.pad_sliding_window(values, window=1, pad_const=pad_const)
np.testing.assert_array_equal(view[0], [pad_const, 3.0, 4.0])
np.testing.assert_array_equal(view[-1], [4.0, 5.0, pad_const])


def test_get_window_masks_marks_pad_values_invalid() -> None:
"""Values equal to the pad constant are False, real values are True."""
view = np.array([[np.nan, 1.0, 2.0], [1.0, 2.0, 3.0]])
masks = window_stats.get_window_masks(view, np.nan)
np.testing.assert_array_equal(masks, [[False, True, True], [True, True, True]])


def test_get_window_masks_non_nan_constant() -> None:
"""A non-nan pad constant is compared by equality."""
view = np.array([[0.0, 1.0, 2.0], [1.0, 2.0, 3.0]])
masks = window_stats.get_window_masks(view, 0.0)
np.testing.assert_array_equal(masks, [[False, True, True], [True, True, True]])


def test_get_window_masks_all_pad_row_is_all_valid() -> None:
"""A row consisting entirely of pad values is left fully unmasked."""
view = np.array([[np.nan, np.nan, np.nan], [1.0, np.nan, 3.0]])
masks = window_stats.get_window_masks(view, np.nan)
np.testing.assert_array_equal(masks, [[True, True, True], [True, False, True]])


def test_window_mean_ignores_padding() -> None:
"""The mean at each frame only averages real values within the window."""
values = np.array([1.0, 2.0, 3.0])
result = window_stats.window_mean(values, window=1)
np.testing.assert_allclose(result, [1.5, 2.0, 2.5])


def test_window_median_ignores_padding() -> None:
"""The median at each frame only considers real values within the window."""
values = np.array([1.0, 2.0, 10.0])
result = window_stats.window_median(values, window=1)
np.testing.assert_allclose(result, [1.5, 2.0, 6.0])


def test_window_std_dev_ignores_padding() -> None:
"""The standard deviation at each frame only considers real values."""
values = np.array([1.0, 3.0, 5.0])
result = window_stats.window_std_dev(values, window=1)
np.testing.assert_allclose(result, [1.0, np.std([1.0, 3.0, 5.0]), 1.0])


def test_window_min_and_max() -> None:
"""window_min and window_max reduce over the valid window values."""
values = np.array([4.0, 1.0, 7.0, 2.0])
np.testing.assert_allclose(window_stats.window_min(values, window=1), [1.0, 1.0, 1.0, 2.0])
np.testing.assert_allclose(window_stats.window_max(values, window=1), [4.0, 7.0, 7.0, 7.0])


def test_window_min_and_max_all_nan_input() -> None:
"""An all-nan input yields an all-nan result rather than raising."""
values = np.full(4, np.nan)
assert np.all(np.isnan(window_stats.window_min(values, window=1)))
assert np.all(np.isnan(window_stats.window_max(values, window=1)))


def test_np_skew_matches_window_skew() -> None:
"""window_skew is np_skew applied to the padded sliding window view."""
values = np.array([1.0, 2.0, 8.0, 3.0, 5.0])
view = window_stats.pad_sliding_window(values, window=2, pad_const=np.nan)
np.testing.assert_allclose(
window_stats.window_skew(values, window=2), window_stats.np_skew(view)
)


def test_np_kurtosis_matches_window_kurtosis() -> None:
"""window_kurtosis is np_kurtosis applied to the padded sliding window view."""
values = np.array([1.0, 2.0, 8.0, 3.0, 5.0])
view = window_stats.pad_sliding_window(values, window=2, pad_const=np.nan)
np.testing.assert_allclose(
window_stats.window_kurtosis(values, window=2), window_stats.np_kurtosis(view)
)
Loading