From 3c505e3623bdff787fbbeb8c66f3c21a1e5aa329 Mon Sep 17 00:00:00 2001 From: Shing Chan Date: Thu, 8 Oct 2026 21:09:57 +0000 Subject: [PATCH] Calculate cadence from inter-step timing Derive cadence metrics from step timestamps so peak and percentile summaries reflect instantaneous walking cadence. Expose validated cadence bounds through the CLI and apply them consistently to adjusted and unadjusted outputs. src/stepcount/stepcount.py - Calculate rolling minutely cadence from inter-step intervals. - Filter cadence observations with configurable bounds and handle empty inputs. - Pass validated CLI cadence settings into summary generation. tests/conftest.py - Add reusable step-timestamp test data. tests/test_numerical_regression.py - Update cadence regressions for timestamp-based calculations. tests/test_stepcount.py - Cover cadence thresholds, cohort summaries, CLI validation, and end-to-end option propagation. --- src/stepcount/stepcount.py | 79 ++++++++++---- tests/conftest.py | 13 +++ tests/test_numerical_regression.py | 48 ++++---- tests/test_stepcount.py | 169 ++++++++++++++++++++++------- 4 files changed, 227 insertions(+), 82 deletions(-) diff --git a/src/stepcount/stepcount.py b/src/stepcount/stepcount.py index 95601ea..ace94fc 100644 --- a/src/stepcount/stepcount.py +++ b/src/stepcount/stepcount.py @@ -1,6 +1,7 @@ from __future__ import annotations import warnings +import math import os import sys import pathlib @@ -83,6 +84,14 @@ def main() -> None: parser.add_argument("--min-walk-per-day", help="The minimum required walking time (in minutes) in a day for metrics calculation.", type=float, default=5) + parser.add_argument("--min-cadence", + help=("Minimum instantaneous inter-step cadence retained for cadence metrics, " + "in steps/min. Default: 50."), + type=float, default=50.0) + parser.add_argument("--max-cadence", + help=("Maximum instantaneous inter-step cadence retained for cadence metrics, " + "in steps/min. Default: 150."), + type=float, default=150.0) parser.add_argument("--bouts-min-walk", help="Minimum percentage of walking for a bout to be considered valid.", type=float, default=0.8) parser.add_argument("--bouts-max-idle", help="Maximum idle (in windows) before a bout is considered to have ended.", @@ -116,6 +125,13 @@ def main() -> None: parser.add_argument('--quiet', '-q', action='store_true', help='Suppress output') args = parser.parse_args() + if not ( + math.isfinite(args.min_cadence) + and math.isfinite(args.max_cadence) + and 0 < args.min_cadence < args.max_cadence + ): + parser.error("--min-cadence must be positive and lower than --max-cadence") + # Arm the certificate fallback before any model download can run. _ensure_download_ssl_context(verbose=not args.quiet) @@ -335,8 +351,19 @@ def main() -> None: info.update({f'WalkingAdjusted(mins)_Hour{h:02}_Weekend': steps_summary_adj['weekend_hour_walks'].loc[h] for h in range(24)}) info.update({f'WalkingAdjusted(mins)_Hour{h:02}_Weekday': steps_summary_adj['weekday_hour_walks'].loc[h] for h in range(24)}) - cadence_summary = summarize_cadence(Y, model.steptol, min_walk_per_day=args.min_walk_per_day) - cadence_summary_adj = summarize_cadence(Y, model.steptol, min_walk_per_day=args.min_walk_per_day, adjust_estimates=True) + cadence_summary = summarize_cadence( + T_steps, + min_walk_per_day=args.min_walk_per_day, + min_cadence=args.min_cadence, + max_cadence=args.max_cadence, + ) + cadence_summary_adj = summarize_cadence( + T_steps, + min_walk_per_day=args.min_walk_per_day, + min_cadence=args.min_cadence, + max_cadence=args.max_cadence, + adjust_estimates=True, + ) for adjusted, summary in (('', cadence_summary), ('Adjusted', cadence_summary_adj)): for cohort, key_prefix in (('', ''), ('_Weekend', 'weekend_'), ('_Weekday', 'weekday_')): for peak_minutes in _CADENCE_PEAK_MINUTES: @@ -1018,25 +1045,29 @@ def _tdelta_to_str(tdelta: Any) -> Any: def summarize_cadence( - Y: _Series, - steptol: int = 3, + T_steps: _Series, min_walk_per_day: int = 5, + min_cadence: float = 50.0, + max_cadence: float = 150.0, + rolling_min_periods: int = 10, adjust_estimates: bool = False ) -> Dict[str, Any]: """ - Summarize cadence information from a series of step counts. + Summarize cadence information from a series of step timestamps. Parameters: - - Y (pd.Series): A pandas Series of step counts. - - steptol (int, optional): The minimum number of steps per window for the window to be considered valid for calculation. Defaults to 3 steps per window. + - T_steps (pd.Series): A pandas Series of timestamps of the steps. - min_walk_per_day (int, optional): The minimum number of walking minutes per day for cadence calculation. Defaults to 5 minutes. + - min_cadence (float, optional): The minimum cadence to consider. Defaults to 50 steps per minute. + - max_cadence (float, optional): The maximum cadence to consider. Defaults to 150 steps per minute. + - rolling_min_periods (int, optional): The minimum number of observations needed for the rolling mean cadence. Defaults to 10 observations. - adjust_estimates (bool, optional): Whether to adjust estimates to account for missing data. Defaults to False. Returns: - dict: A dictionary containing various summary cadence statistics. Example: - summary = summarize_cadence(Y, steptol=3, adjust_estimates=True) + summary = summarize_cadence(T_steps, adjust_estimates=True) """ import numpy as np @@ -1045,33 +1076,43 @@ def summarize_cadence( # TODO: split walking and running cadence? - dt = utils.infer_freq(Y.index).total_seconds() - min_steps_per_min = steptol * 60 / dt # rescale steptol to steps/min - cadence_columns = [ *(f'CadencePeak{n}(steps/min)' for n in _CADENCE_PEAK_MINUTES), 'Cadence95th(steps/min)', ] + if len(T_steps) == 0: + summary: Dict[str, Any] = {'daily': pd.DataFrame(columns=cadence_columns)} + for key_prefix in ('', 'weekend_', 'weekday_'): + summary.update({f'{key_prefix}cadence_peak{n}': np.nan for n in _CADENCE_PEAK_MINUTES}) + summary[f'{key_prefix}cadence_p95'] = np.nan + return summary + def _summarize_day( x: Any, - min_steps_per_min: float = min_steps_per_min, min_walk_per_day: int = min_walk_per_day, ) -> Tuple[Any, ...]: - y = x[x >= min_steps_per_min] # if not enough walking time, return NA. # note: min_walk_per_day in minutes, x must be minutely - if len(y) < min_walk_per_day: + if x.count() < min_walk_per_day: return (np.nan,) * len(cadence_columns) - descending = y.sort_values(ascending=False) + descending = x.dropna().sort_values(ascending=False) cadence_peaks = tuple( descending[descending >= descending.iloc[min(n, len(descending)) - 1]].mean() for n in _CADENCE_PEAK_MINUTES ) - return (*cadence_peaks, y.quantile(.95)) - - minutely = cast(Any, Y).resample('T').sum().rename('Steps') # steps/min + return (*cadence_peaks, x.quantile(.95)) + + step_times = cast(Any, T_steps.copy()) + step_times.index = step_times + cad = 60 / step_times.diff().dt.total_seconds() # steps/min + # set very low and very high cadences to NA + cad = cad.where(cad.between(min_cadence, max_cadence)) + # 1min moving average + cad = cad.rolling('T', min_periods=rolling_min_periods).mean() + # resample to minutely + minutely = cad.resample('T').mean().rename('Cadence(steps/min)') # cadence https://jamanetwork.com/journals/jama/fullarticle/2763292 @@ -1126,7 +1167,7 @@ def _summarize_day( daily_cadence_p95.round().astype(pd.Int64Dtype()), ], axis=1) - summary: Dict[str, Any] = {'daily': daily} + summary = {'daily': daily} for key_prefix, peaks, p95 in ( ('', cadence_peaks, cadence_p95), ('weekend_', weekend_cadence_peaks, weekend_cadence_p95), diff --git a/tests/conftest.py b/tests/conftest.py index 5ae1515..fc354a3 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -329,6 +329,19 @@ def step_counts_series(accel_data_1_5_days): return pd.Series(step_counts, index=window_times, name='Steps') +@pytest.fixture(scope="session") +def step_times_series(step_counts_series): + """Expand window-level counts into evenly spaced step timestamps.""" + step_times = [] + window_seconds = 10 + for window_start, count in step_counts_series.items(): + if count > 0: + offsets = np.arange(1, int(count) + 1) * (window_seconds / count) + step_times.extend(window_start + pd.to_timedelta(offsets, unit="s")) + + return pd.Series(step_times, name="time") + + @pytest.fixture(scope="function") def temp_dir(): """Create a temporary directory for test outputs, cleaned up after test.""" diff --git a/tests/test_numerical_regression.py b/tests/test_numerical_regression.py index e19fab1..83e2e66 100644 --- a/tests/test_numerical_regression.py +++ b/tests/test_numerical_regression.py @@ -130,43 +130,47 @@ def test_adjusted_step_summary_imputes_a_known_gap(): def test_cadence_summary_matches_hand_calculated_values(): - times = pd.date_range("2024-01-15", periods=30, freq="1min") - steps = pd.Series(range(3, 33), index=times, name="Steps") + step_times = pd.Series( + pd.date_range("2024-01-15", periods=241, freq="500ms"), + name="time", + ) result = stepcount.summarize_cadence( - steps, - steptol=3, + step_times, min_walk_per_day=1, + rolling_min_periods=1, ) - assert result["cadence_peak1"] == 32 - assert result["cadence_peak5"] == 30 - assert result["cadence_peak10"] == 28 - assert result["cadence_peak30"] == 18 - assert result["cadence_p95"] == 31 + assert result["cadence_peak1"] == 120 + assert result["cadence_peak5"] == 120 + assert result["cadence_peak10"] == 120 + assert result["cadence_peak30"] == 120 + assert result["cadence_p95"] == 120 assert result["daily"].iloc[0].to_dict() == { - "CadencePeak1(steps/min)": 32, - "CadencePeak5(steps/min)": 30, - "CadencePeak10(steps/min)": 28, - "CadencePeak30(steps/min)": 18, - "Cadence95th(steps/min)": 31, + "CadencePeak1(steps/min)": 120, + "CadencePeak5(steps/min)": 120, + "CadencePeak10(steps/min)": 120, + "CadencePeak30(steps/min)": 120, + "Cadence95th(steps/min)": 120, } def test_cadence_peaks_include_every_epoch_tied_at_the_cutoff(): - times = pd.date_range("2024-01-15", periods=21, freq="1min") - steps = pd.Series([200] + [10] * 20, index=times, name="Steps") + step_times = pd.Series( + pd.date_range("2024-01-15", periods=2401, freq="500ms"), + name="time", + ) result = stepcount.summarize_cadence( - steps, - steptol=3, + step_times, min_walk_per_day=1, + rolling_min_periods=1, ) - assert result["cadence_peak1"] == 200 - assert result["cadence_peak5"] == 19 - assert result["cadence_peak10"] == 19 - assert result["cadence_peak30"] == 19 + assert result["cadence_peak1"] == 120 + assert result["cadence_peak5"] == 120 + assert result["cadence_peak10"] == 120 + assert result["cadence_peak30"] == 120 def test_bout_summary_matches_hand_calculated_values(): diff --git a/tests/test_stepcount.py b/tests/test_stepcount.py index c7bc1d7..6a575c7 100644 --- a/tests/test_stepcount.py +++ b/tests/test_stepcount.py @@ -182,17 +182,17 @@ def test_summarize_steps_hour_profile(self, step_counts_series): class TestSummarizeCadence: """Tests for cadence summarization.""" - def test_summarize_cadence_basic(self, step_counts_series): + def test_summarize_cadence_basic(self, step_times_series): """Test basic cadence summarization.""" - summary = stepcount.summarize_cadence(step_counts_series, steptol=3) + summary = stepcount.summarize_cadence(step_times_series) for peak_minutes in CADENCE_PEAK_MINUTES: assert f'cadence_peak{peak_minutes}' in summary assert 'cadence_p95' in summary - def test_summarize_cadence_daily(self, step_counts_series): + def test_summarize_cadence_daily(self, step_times_series): """Test daily cadence values.""" - summary = stepcount.summarize_cadence(step_counts_series, steptol=3) + summary = stepcount.summarize_cadence(step_times_series) assert 'daily' in summary daily = summary['daily'] @@ -205,9 +205,9 @@ def test_summarize_cadence_daily(self, step_counts_series): 'Cadence95th(steps/min)', ] - def test_summarize_cadence_peaks_are_monotonic(self, step_counts_series): + def test_summarize_cadence_peaks_are_monotonic(self, step_times_series): """Test that averages cannot increase as more peak minutes are included.""" - summary = stepcount.summarize_cadence(step_counts_series, steptol=3) + summary = stepcount.summarize_cadence(step_times_series) peaks = [summary[f'cadence_peak{minutes}'] for minutes in CADENCE_PEAK_MINUTES] if not any(np.isnan(peak) for peak in peaks): @@ -215,66 +215,97 @@ def test_summarize_cadence_peaks_are_monotonic(self, step_counts_series): def test_summarize_cadence_longer_peaks_use_available_minutes(self): """Test that eligible short days retain the established peak30 behavior.""" - times = pd.date_range('2024-01-15', periods=5, freq='1min') - steps = pd.Series([0, 30, 60, 90, 120], index=times, name='Steps') + step_times = pd.Series( + pd.date_range('2024-01-15', periods=601, freq='500ms'), + name='time', + ) - summary = stepcount.summarize_cadence(steps, steptol=3, min_walk_per_day=1) + summary = stepcount.summarize_cadence( + step_times, + min_walk_per_day=1, + rolling_min_periods=1, + ) assert summary['cadence_peak1'] == 120 - assert summary['cadence_peak5'] == 75 - assert summary['cadence_peak10'] == 75 - assert summary['cadence_peak30'] == 75 + assert summary['cadence_peak5'] == 120 + assert summary['cadence_peak10'] == 120 + assert summary['cadence_peak30'] == 120 + + def test_summarize_cadence_default_min_walk_is_five(self): + """The public API uses the same five-minute threshold as the CLI.""" + step_times = pd.Series( + pd.date_range('2024-01-15', periods=721, freq='500ms'), + name='time', + ) + + summary = stepcount.summarize_cadence(step_times) + + for peak_minutes in CADENCE_PEAK_MINUTES: + assert summary[f'cadence_peak{peak_minutes}'] == 120 + assert summary['cadence_p95'] == 120 def test_summarize_cadence_min_walk_filter(self): """Test exact behavior immediately below and at the walking threshold.""" - times = pd.date_range('2024-01-15', periods=5, freq='1min') - below_threshold = pd.Series([30, 30, 30, 30, 0], index=times, name='Steps') - at_threshold = pd.Series([30] * 5, index=times, name='Steps') + below_threshold = pd.Series( + pd.date_range('2024-01-15', periods=480, freq='500ms'), + name='time', + ) + at_threshold = pd.Series( + pd.date_range('2024-01-15', periods=600, freq='500ms'), + name='time', + ) below_summary = stepcount.summarize_cadence( - below_threshold, steptol=3, min_walk_per_day=5 + below_threshold, min_walk_per_day=5, rolling_min_periods=1 ) at_summary = stepcount.summarize_cadence( - at_threshold, steptol=3, min_walk_per_day=5 + at_threshold, min_walk_per_day=5, rolling_min_periods=1 ) for peak_minutes in CADENCE_PEAK_MINUTES: assert np.isnan(below_summary[f'cadence_peak{peak_minutes}']) - assert at_summary[f'cadence_peak{peak_minutes}'] == 30 + assert at_summary[f'cadence_peak{peak_minutes}'] == 120 assert np.isnan(below_summary['cadence_p95']) - assert at_summary['cadence_p95'] == 30 + assert at_summary['cadence_p95'] == 120 def test_summarize_cadence_weekend_weekday_and_adjusted_values(self): - """Test exact split values and weekend imputation for adjusted estimates.""" + """Test exact overall, weekend, weekday, and adjusted values.""" daily_cadences = { - '2024-01-15': 10, - '2024-01-16': 20, - '2024-01-17': 30, - '2024-01-18': 40, - '2024-01-19': 50, - # Saturday is deliberately missing and is imputed from Sunday. - '2024-01-21': 70, + '2024-01-15': 60, + '2024-01-16': 70, + '2024-01-17': 80, + '2024-01-18': 90, + '2024-01-19': 100, + '2024-01-20': 110, + '2024-01-21': 120, } - steps = pd.concat([ + step_times = pd.concat([ pd.Series( - [cadence] * 5, - index=pd.date_range(f'{date} 12:00', periods=5, freq='1min'), + pd.date_range( + f'{date} 12:00', + periods=5 * cadence + 1, + freq=pd.Timedelta(seconds=60 / cadence), + ), + name='time', ) for date, cadence in daily_cadences.items() - ]).rename('Steps') + ], ignore_index=True) summary = stepcount.summarize_cadence( - steps, steptol=3, min_walk_per_day=1 + step_times, min_walk_per_day=1, rolling_min_periods=1 ) adjusted = stepcount.summarize_cadence( - steps, steptol=3, min_walk_per_day=1, adjust_estimates=True + step_times, + min_walk_per_day=1, + rolling_min_periods=1, + adjust_estimates=True, ) - for cadence_summary, expected_overall in ((summary, 35), (adjusted, 40)): + for cadence_summary in (summary, adjusted): for peak_minutes in CADENCE_PEAK_MINUTES: - assert cadence_summary[f'cadence_peak{peak_minutes}'] == expected_overall - assert cadence_summary[f'weekday_cadence_peak{peak_minutes}'] == 30 - assert cadence_summary[f'weekend_cadence_peak{peak_minutes}'] == 70 + assert cadence_summary[f'cadence_peak{peak_minutes}'] == 90 + assert cadence_summary[f'weekday_cadence_peak{peak_minutes}'] == 80 + assert cadence_summary[f'weekend_cadence_peak{peak_minutes}'] == 115 class TestNumbaDetectBouts: @@ -712,7 +743,7 @@ def test_enmo_movement(self): class TestIntegration: """Integration tests for summary functions.""" - def test_full_summary_pipeline(self, step_counts_series, accel_data_1_5_days): + def test_full_summary_pipeline(self, step_counts_series, step_times_series, accel_data_1_5_days): """Test running all summary functions together.""" enmo_summary = stepcount.summarize_enmo(accel_data_1_5_days) assert 'avg' in enmo_summary @@ -720,7 +751,7 @@ def test_full_summary_pipeline(self, step_counts_series, accel_data_1_5_days): steps_summary = stepcount.summarize_steps(step_counts_series, steptol=3) assert 'total_steps' in steps_summary - cadence_summary = stepcount.summarize_cadence(step_counts_series, steptol=3) + cadence_summary = stepcount.summarize_cadence(step_times_series) assert 'cadence_peak1' in cadence_summary bouts_summary = stepcount.summarize_bouts( @@ -835,6 +866,49 @@ def test_cli_ssl_repo_path_option(self): assert result.returncode == 0 assert '--ssl-repo-path' in result.stdout + def test_cli_cadence_bounds_in_help(self): + """Test that cadence filtering options are documented.""" + result = subprocess.run( + [sys.executable, '-m', 'stepcount.stepcount', '--help'], + capture_output=True, + text=True, + timeout=30 + ) + + assert result.returncode == 0 + assert '--min-cadence' in result.stdout + assert '--max-cadence' in result.stdout + + @pytest.mark.parametrize( + ("min_cadence", "max_cadence"), + [ + ("0", "150"), + ("150", "150"), + ("nan", "150"), + ("50", "inf"), + ], + ) + def test_cli_rejects_invalid_cadence_bounds(self, min_cadence, max_cadence): + """Cadence bounds must be finite, positive, and ordered.""" + result = subprocess.run( + [ + sys.executable, + '-m', + 'stepcount.stepcount', + 'recording.cwa', + '--min-cadence', + min_cadence, + '--max-cadence', + max_cadence, + ], + capture_output=True, + text=True, + timeout=30, + ) + + assert result.returncode != 0 + assert '--min-cadence must be positive and lower than --max-cadence' in result.stderr + def test_cli_download_models_in_help(self): """Test that --download-models appears in --help output.""" result = subprocess.run( @@ -905,8 +979,9 @@ def predict_from_frame(frame): name="Steps", ) walking = steps >= 3 - step_times = pd.DataFrame( - {"time": frame.index.repeat(steps.to_numpy())}, + step_times = pd.Series( + frame.index.repeat(steps.to_numpy()), + name="time", ) return steps, walking, step_times @@ -922,6 +997,8 @@ def predict_from_frame(frame): ) load_model = MagicMock(return_value=model) monkeypatch.setattr(stepcount, "load_model", load_model) + summarize_cadence = MagicMock(wraps=stepcount.summarize_cadence) + monkeypatch.setattr(stepcount, "summarize_cadence", summarize_cadence) argv = [ "stepcount", @@ -940,6 +1017,10 @@ def predict_from_frame(frame): "0", "--min-walk-per-day", "1", + "--min-cadence", + "45.5", + "--max-cadence", + "175.5", "--quiet", ] if model_type == "ssl": @@ -956,6 +1037,10 @@ def predict_from_frame(frame): assert model.window_len == 60 assert detector.sample_rate == 1 assert detector.verbose is False + assert summarize_cadence.call_count == 2 + for call in summarize_cadence.call_args_list: + assert call.kwargs["min_cadence"] == 45.5 + assert call.kwargs["max_cadence"] == 175.5 if model_type == "ssl": assert detector.device == "cpu" @@ -983,6 +1068,8 @@ def predict_from_frame(frame): for adjusted in ("", "Adjusted"): for cohort in ("", "_Weekend", "_Weekday"): assert f"CadencePeak{peak_minutes}{adjusted}(steps/min){cohort}" in result_info + assert result_info["StepCountArgs"]["min_cadence"] == 45.5 + assert result_info["StepCountArgs"]["max_cadence"] == 175.5 daily = pd.read_csv(result_dir / "synthetic-Daily.csv.gz") assert daily["Date"].tolist() == ["2024-01-16"] assert daily["WearTime(hours)"].tolist() == [1.0]