Coverage for physiodsp / activity / pim.py: 81%

1from numpy import abs 

2from pandas import DataFrame 

3from pydantic import BaseModel 

4 

5from physiodsp.base import BaseAlgorithm 

6from physiodsp.sensors.imu.base import IMUData 

7 

8 

9class PIMSettings(BaseModel): 

10 """PIM Algorithm Settings""" 

11 aggregation_window: int = 5 

12 

13 

14class PIMAlgorithm(BaseAlgorithm): 

15 """Proportional Integration Mode""" 

16 

17 _algorithm_name = "PIMAlgorithm" 

18 _version = "0.1.0" 

19 

20 def __init__(self, 

21 settings: PIMSettings = PIMSettings() 

22 ) -> None: 

23 self.settings = settings 

24 self._aggregation_window = settings.aggregation_window 

25 return None 

26 

27 def run(self, data: IMUData): 

28 self.data = data 

29 self.values_x = abs(data.x) 

30 self.values_y = abs(data.y) 

31 self.values_z = abs(data.z) 

32 

33 # Raw biomarker: combine absolute values for raw signal representation 

34 self.biomarker = DataFrame({ 

35 'timestamps': data.timestamps, 

36 'x': self.values_x, 

37 'y': self.values_y, 

38 'z': self.values_z 

39 }) 

40 

41 return self 

42 

43 def aggregate(self, 

44 method: str = 'sum' 

45 ): 

46 

47 df = self.biomarker.copy() 

48 

49 df['timestamps'] = (df['timestamps'] // self.aggregation_window) * self.aggregation_window 

50 

51 df_agg = df.groupby('timestamps')[["x", "y", "z"]].agg(method).reset_index(drop=False) 

52 

53 self.biomarker_agg = df_agg 

54 

55 return self