Coverage for tests / test_activity_step_count.py: 100%
167 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-07-14 19:36 +0000
1from __future__ import annotations
3import numpy as np
4import pytest
6from physiodsp.activity.step_count import (
7 StepBout,
8 StepCount,
9 StepCountSettings,
10 autocorrelation_peak,
11 detect_steps_in_window,
12 filter_short_bouts,
13 highpass_filter,
14 interpolate_missed_steps,
15 is_locomotion_window,
16 merge_close_bouts,
17 steps_to_cadence_series,
18)
19from physiodsp.sensors.imu.accelerometer import AccelerometerData
22# ---------------------------------------------------------------------------
23# Helpers
24# ---------------------------------------------------------------------------
26FS = 50.0 # Hz — a common sampling rate for testing
29def make_sine_accel(
30 freq_hz: float,
31 duration_s: float,
32 amplitude: float = 0.5,
33 fs: float = FS,
34 noise_std: float = 0.0,
35) -> np.ndarray:
36 """Periodic sinusoidal signal centred at 0, simulating dynamic acceleration."""
37 t = np.arange(int(duration_s * fs)) / fs
38 sig = amplitude * np.sin(2 * np.pi * freq_hz * t)
39 if noise_std > 0:
40 rng = np.random.default_rng(42)
41 sig += rng.normal(0, noise_std, size=len(sig))
42 return sig
45def make_accel_data(
46 dynamic_signal: np.ndarray,
47 fs: float = FS,
48 gravity: float = 1.0,
49) -> AccelerometerData:
50 """Wrap a 1-D dynamic signal into an AccelerometerData instance.
51 The signal is injected on the Z axis; X and Y carry only gravity / noise.
52 """
53 n = len(dynamic_signal)
54 t = np.arange(n) / fs
55 x = np.zeros(n)
56 y = np.zeros(n)
57 z = gravity + dynamic_signal # gravity on Z; dynamic component added
58 return AccelerometerData(timestamps=t, x=x, y=y, z=z, fs=int(fs))
61def run_step_count_on_sine(
62 freq_hz: float,
63 duration_s: float = 30.0,
64 amplitude: float = 0.5,
65 noise_std: float = 0.0,
66 settings: StepCountSettings | None = None,
67) -> StepCount:
68 """Helper to run StepCount algorithm on a synthetic sine signal."""
69 sig = make_sine_accel(freq_hz, duration_s, amplitude, noise_std=noise_std)
70 accel = make_accel_data(sig)
71 sc = StepCount(settings=settings or StepCountSettings())
72 return sc.run(accel)
75def make_test_bouts() -> list[StepBout]:
76 """Helper to create a set of sample StepBout objects."""
77 return [
78 StepBout(0.0, 5.0, 8, 100.0, "walk", [0.5 * i for i in range(8)]),
79 StepBout(20.0, 25.0, 2, 80.0, "walk", [20.0 + 0.5 * i for i in range(2)]),
80 StepBout(30.0, 40.0, 15, 120.0, "walk", [30.0 + 0.5 * i for i in range(15)]),
81 ]
84# ---------------------------------------------------------------------------
85# Pre-processing Tests
86# ---------------------------------------------------------------------------
88def test_step_count_highpass_removes_dc():
89 """HP filter must remove the 1 g DC gravity component."""
90 n = 1000
91 sig = np.ones(n) * 1.0 + 0.3 * np.sin(2 * np.pi * 1.5 * np.arange(n) / FS)
92 filtered = highpass_filter(sig, fs=FS, cutoff_hz=0.5)
93 # DC component must be removed: mean of filtered signal ≈ 0
94 assert abs(np.mean(filtered)) < 0.05
97def test_step_count_highpass_preserves_gait_frequencies():
98 """HP filter must not attenuate a 1.5 Hz gait signal significantly."""
99 duration_s = 10.0
100 sig = make_sine_accel(freq_hz=1.5, duration_s=duration_s)
101 filtered = highpass_filter(sig, fs=FS, cutoff_hz=0.5)
102 # RMS of filtered signal should be > 80 % of input RMS
103 assert np.std(filtered) > 0.8 * np.std(sig)
106# ---------------------------------------------------------------------------
107# Motion Gate Tests
108# ---------------------------------------------------------------------------
110def test_step_count_sleep_window_rejected():
111 """Below-sleep-threshold SMA -> is_locomotion_window returns False."""
112 sig = np.random.randn(int(2.0 * FS)) * 0.005
113 sma = float(np.mean(np.abs(sig)))
114 is_loco, _, state = is_locomotion_window(
115 sig, fs=FS, sma=sma,
116 sma_sleep_thr=0.02, sma_adl_thr=0.05,
117 periodicity_threshold=0.5, spectral_purity_threshold=8.0
118 )
119 assert not is_loco
120 assert state == "sleep"
123# ---------------------------------------------------------------------------
124# Periodicity Detector Tests
125# ---------------------------------------------------------------------------
127@pytest.mark.parametrize("freq", [1.2, 1.6, 2.0, 2.5, 3.0])
128def test_step_count_pure_sine_is_periodic(freq: float):
129 """A pure sinusoid at any valid gait frequency must be detected as periodic."""
130 sig = make_sine_accel(freq_hz=freq, duration_s=4.0)
131 peak, detected_freq = autocorrelation_peak(sig, fs=FS)
132 assert peak > 0.5, f"AC peak {peak:.3f} too low for {freq} Hz sine"
133 assert abs(detected_freq - freq) < 0.3, (
134 f"Detected freq {detected_freq:.2f} Hz too far from {freq} Hz"
135 )
138def test_step_count_noise_is_not_periodic():
139 """White noise must not exceed the periodicity threshold."""
140 rng = np.random.default_rng(0)
141 sig = rng.normal(0, 0.3, size=int(4.0 * FS))
142 peak, _ = autocorrelation_peak(sig, fs=FS)
143 assert peak < 0.5, f"Noise produced AC peak {peak:.3f} >= 0.5"
146def test_step_count_walk_window_detected():
147 """Walking-frequency signal above SMA threshold -> locomotion detected."""
148 sig = make_sine_accel(freq_hz=1.6, duration_s=4.0, amplitude=0.4)
149 sma = float(np.mean(np.abs(sig)))
150 is_loco, cadence, state = is_locomotion_window(
151 sig, fs=FS, sma=sma,
152 sma_sleep_thr=0.02, sma_adl_thr=0.05,
153 periodicity_threshold=0.5, spectral_purity_threshold=8.0
154 )
155 assert is_loco
156 assert state in ("walk", "run")
157 assert 1.0 < cadence < 2.5
160def test_step_count_run_window_detected():
161 """Running-frequency signal (2.8 Hz) must be classified as run."""
162 sig = make_sine_accel(freq_hz=2.8, duration_s=4.0, amplitude=1.5)
163 sma = float(np.mean(np.abs(sig)))
164 is_loco, cadence, state = is_locomotion_window(
165 sig, fs=FS, sma=sma,
166 sma_sleep_thr=0.02, sma_adl_thr=0.05,
167 periodicity_threshold=0.5, spectral_purity_threshold=8.0
168 )
169 assert is_loco
170 assert state == "run"
173# ---------------------------------------------------------------------------
174# Adaptive Peak Detector Tests
175# ---------------------------------------------------------------------------
177def test_step_count_slow_walk_accuracy():
178 """Slow walking at 1.3 Hz for 10 s -> ~13 steps expected."""
179 freq = 1.3
180 duration = 10.0
181 sig = make_sine_accel(freq_hz=freq, duration_s=duration, amplitude=0.3)
182 peaks = detect_steps_in_window(sig, fs=FS, cadence_hz=freq)
183 expected = int(freq * duration)
184 assert abs(len(peaks) - expected) <= 2, (
185 f"Expected ~{expected} steps, got {len(peaks)}"
186 )
189def test_step_count_running_accuracy():
190 """Running at 2.8 Hz for 10 s -> ~28 steps expected."""
191 freq = 2.8
192 duration = 10.0
193 sig = make_sine_accel(freq_hz=freq, duration_s=duration, amplitude=1.5)
194 peaks = detect_steps_in_window(sig, fs=FS, cadence_hz=freq)
195 expected = int(freq * duration)
196 assert abs(len(peaks) - expected) <= 3, (
197 f"Expected ~{expected} steps, got {len(peaks)}"
198 )
201def test_step_count_amplitude_invariance():
202 """Step count should be the same regardless of signal amplitude (adaptive threshold)."""
203 freq = 1.6
204 duration = 10.0
205 steps_low = len(detect_steps_in_window(
206 make_sine_accel(freq, duration, amplitude=0.3), fs=FS, cadence_hz=freq))
207 steps_high = len(detect_steps_in_window(
208 make_sine_accel(freq, duration, amplitude=2.0), fs=FS, cadence_hz=freq))
209 assert abs(steps_low - steps_high) <= 2, (
210 f"Amplitude should not affect count: low={steps_low}, high={steps_high}"
211 )
214# ---------------------------------------------------------------------------
215# Post-processing Tests
216# ---------------------------------------------------------------------------
218def test_step_count_filter_short_bouts():
219 bouts = make_test_bouts()
220 filtered = filter_short_bouts(bouts, min_steps=3)
221 assert all(b.step_count >= 3 for b in filtered)
222 assert len(filtered) == 2
225def test_step_count_merge_close_bouts():
226 """Two bouts separated by 2 s gap should be merged with merge_gap_s=3."""
227 b1 = StepBout(0.0, 5.0, 8, 100.0, "walk", list(np.linspace(0, 5, 8)))
228 b2 = StepBout(7.0, 12.0, 8, 100.0, "walk", list(np.linspace(7, 12, 8)))
229 merged = merge_close_bouts([b1, b2], merge_gap_s=3.0)
230 assert len(merged) == 1
231 assert merged[0].step_count == 16
234def test_step_count_no_merge_far_bouts():
235 """Bouts separated by > merge_gap_s must NOT be merged."""
236 b1 = StepBout(0.0, 5.0, 8, 100.0, "walk", list(np.linspace(0, 5, 8)))
237 b2 = StepBout(15.0, 20.0, 8, 100.0, "walk", list(np.linspace(15, 20, 8)))
238 merged = merge_close_bouts([b1, b2], merge_gap_s=3.0)
239 assert len(merged) == 2
242def test_step_count_interpolate_missed_steps():
243 """A gap of ~2 periods should produce one interpolated step."""
244 cadence_hz = 2.0 # 0.5 s per step
245 steps = [0.0, 0.5, 1.5, 2.0]
246 filled = interpolate_missed_steps(steps, cadence_hz=cadence_hz)
247 assert len(filled) == 5
250def test_step_count_cadence_series_bin_count():
251 """steps_to_cadence_series must produce exactly ceil(duration/bin_s) bins."""
252 steps = np.array([1.0, 2.0, 3.0, 5.0, 6.0])
253 bin_ts, cad = steps_to_cadence_series(steps, total_duration_s=10.0, bin_s=1.0)
254 assert len(bin_ts) == 10
255 assert len(cad) == 10
258def test_step_count_cadence_series_values():
259 """Bins with exactly 2 steps at bin_s=1 s should give 120 spm."""
260 steps = np.array([0.1, 0.9])
261 _, cad = steps_to_cadence_series(steps, total_duration_s=3.0, bin_s=1.0)
262 assert cad[0] == pytest.approx(120.0)
265# ---------------------------------------------------------------------------
266# End-to-end Integration Tests
267# ---------------------------------------------------------------------------
269@pytest.mark.parametrize("freq,speed_label", [
270 (1.2, "slow_walk"),
271 (1.6, "normal_walk"),
272 (2.0, "brisk_walk"),
273 (2.5, "jog"),
274 (3.0, "run"),
275])
276def test_step_count_accuracy_all_speeds(freq: float, speed_label: str):
277 duration = 30.0
278 result = run_step_count_on_sine(freq, duration_s=duration, amplitude=0.5)
279 expected = int(freq * duration)
280 error_pct = abs(result.total_steps - expected) / expected * 100
281 assert error_pct <= 15.0, (
282 f"[{speed_label}] Expected ~{expected} steps, got {result.total_steps} "
283 f"(error {error_pct:.1f} %)"
284 )
287def test_step_count_zero_steps_sleep():
288 """Near-zero acceleration (sleep) must produce 0 steps."""
289 rng = np.random.default_rng(42)
290 n = int(30.0 * FS)
291 t = np.arange(n) / FS
292 sig = rng.normal(0, 0.005, n)
293 accel = AccelerometerData(
294 timestamps=t, x=sig, y=sig,
295 z=np.ones(n) + sig, fs=int(FS)
296 )
297 result = StepCount().run(accel)
298 assert result.total_steps == 0
301def test_step_count_biomarker_schema():
302 """biomarker DataFrame must have the required columns and correct length."""
303 result = run_step_count_on_sine(1.6, duration_s=20.0)
304 assert "timestamps" in result.biomarker.columns
305 assert "step_count" in result.biomarker.columns
306 assert "cadence_spm" in result.biomarker.columns
307 assert "activity_state" in result.biomarker.columns
308 assert len(result.biomarker) == 20
311def test_step_count_total_steps_consistency():
312 """The per-bin step_count column must sum to total_steps."""
313 result = run_step_count_on_sine(1.6, duration_s=20.0)
314 assert int(result.biomarker["step_count"].sum()) == result.total_steps
317def test_step_count_noisy_free_living_walk():
318 """Walking signal with realistic free-living noise should still be detected."""
319 result = run_step_count_on_sine(
320 freq_hz=1.6, duration_s=30.0, amplitude=0.4, noise_std=0.15
321 )
322 expected = int(1.6 * 30)
323 error_pct = abs(result.total_steps - expected) / expected * 100
324 assert error_pct <= 20.0, (
325 f"Free-living noisy walk: expected ~{expected}, got {result.total_steps} "
326 f"(error {error_pct:.1f} %)"
327 )
330def test_step_count_bout_merge_fragmentation():
331 """With merge_gap_s > 0, a walk-pause-walk sequence should produce one bout."""
332 freq = 1.6
333 fs = FS
334 walk1 = make_sine_accel(freq, 10.0, amplitude=0.5)
335 pause = np.zeros(int(2.0 * fs))
336 walk2 = make_sine_accel(freq, 10.0, amplitude=0.5)
337 sig = np.concatenate([walk1, pause, walk2])
338 t = np.arange(len(sig)) / fs
339 z = 1.0 + sig
340 accel = AccelerometerData(
341 timestamps=t, x=np.zeros_like(sig),
342 y=np.zeros_like(sig), z=z, fs=int(fs)
343 )
345 merged_result = StepCount(settings=StepCountSettings(bout_merge_gap_s=3.0)).run(accel)
346 no_merge_result = StepCount(settings=StepCountSettings(bout_merge_gap_s=0.0)).run(accel)
348 assert len(merged_result.bouts) <= len(no_merge_result.bouts)
351def test_step_count_aggregate_output_shape():
352 """aggregate() must produce a DataFrame with timestamps and values columns."""
353 result = run_step_count_on_sine(1.6, duration_s=120.0)
354 result.aggregate(method="mean")
355 assert hasattr(result, "biomarker_agg")
356 assert "timestamps" in result.biomarker_agg.columns
357 assert "values" in result.biomarker_agg.columns
358 assert len(result.biomarker_agg) == 2