Coverage for physiodsp / activity / enmo.py: 100%

1from pandas import Series, DataFrame 

2from pydantic import BaseModel, Field, PositiveInt 

3 

4from physiodsp.base import BaseAlgorithm 

5from physiodsp.sensors.imu.accelerometer import AccelerometerData 

6 

7 

8class ENMOSettings(BaseModel): 

9 

10 window_len: PositiveInt = Field(default=1, description="processing window length in seconds") 

11 

12 aggregation_window: PositiveInt = Field(default=60, description="aggregation window length in seconds") 

13 

14 

15class ENMO(BaseAlgorithm): 

16 """Euclidean Norm Minus One""" 

17 

18 _algorithm_name = "ENMO" 

19 _version = "v0.1.0" 

20 

21 def __init__(self, 

22 settings: ENMOSettings = ENMOSettings() 

23 ) -> None: 

24 self.settings = settings 

25 self._window_len = settings.window_len 

26 self._aggregation_window = settings.aggregation_window 

27 return None 

28 

29 def run(self, accelerometer: AccelerometerData): 

30 """_summary_ 

31 

32 Args: 

33 accelerometer (AccelerometerData): _description_ 

34 

35 Returns: 

36 ENMO: _description_ 

37 """ 

38 

39 enmo = accelerometer.magnitude - 1 

40 enmo[enmo < 0] = 0 

41 

42 self.timestamps = Series(accelerometer.timestamps).rolling( 

43 window=int(self._window_len * accelerometer.fs), 

44 step=int(self._window_len * accelerometer.fs), 

45 min_periods=int(self._window_len * accelerometer.fs), 

46 closed="left" 

47 ).max()[1:] 

48 

49 self.values = Series(enmo).rolling( 

50 window=int(self._window_len * accelerometer.fs), 

51 step=int(self._window_len * accelerometer.fs), 

52 min_periods=int(self._window_len * accelerometer.fs), 

53 closed="left" 

54 ).mean()[1:] 

55 

56 self.biomarker = DataFrame( 

57 list(zip(self.timestamps, self.values)), 

58 columns=['timestamps', 'values'] 

59 ) 

60 

61 return self 

62 

63 def aggregate(self, 

64 method: str = 'mean' 

65 ): 

66 super().aggregate( 

67 self.timestamps, 

68 self.values, 

69 method 

70 ) 

71 return self