Coverage for tests / test_activity_enmo.py: 100%

1import os 

2import pytest 

3from datetime import datetime, timezone 

4 

5from numpy import arange 

6from pandas import read_csv 

7from physiodsp.sensors.imu.accelerometer import AccelerometerData 

8from physiodsp.activity.enmo import ENMO 

9 

10 

11test_folder_path = os.path.dirname(os.path.realpath(__file__)) 

12 

13# Fixed, minute-aligned reference timestamp. ENMO.aggregate() bins timestamps 

14# by flooring to the aggregation window (e.g. `// 60 * 60`), so seeding from 

15# datetime.now() made the test flaky: whenever it happened to run within a 

16# few seconds of a real minute boundary, the generated samples spanned two 

17# bins instead of one. 

18FIXED_TIMESTAMP_START = datetime(2024, 1, 1, tzinfo=timezone.utc).timestamp() 

19 

20 

21@pytest.mark.parametrize( 

22 "n_samples,fs", 

23 [ 

24 (128, 32), 

25 (256, 64), 

26 (256, 32) 

27 ] 

28) 

29def test_activity_enmo(n_samples, fs): 

30 

31 df = read_csv(os.path.join(test_folder_path, "accelerometer.csv"), usecols=["x", "y", "z"]) 

32 

33 timestamps = FIXED_TIMESTAMP_START + arange(start=0, step=1/fs, stop=int(n_samples/fs)) 

34 

35 accelerometer = AccelerometerData( 

36 timestamps=timestamps, 

37 x=df.x.values[:n_samples], 

38 y=df.y.values[:n_samples], 

39 z=df.z.values[:n_samples], 

40 fs=fs 

41 ) 

42 

43 enmo_processor = ENMO().run(accelerometer=accelerometer).aggregate() 

44 

45 enmo_1s = enmo_processor.biomarker 

46 enmo_60s = enmo_processor.biomarker_agg 

47 

48 assert len(enmo_1s) == int(n_samples / fs) - 1 

49 assert len(enmo_60s) == 1