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

1from numpy import abs 

2from pandas import DataFrame 

3 

4from physiodsp.base import BaseAlgorithm 

5from physiodsp.sensors.imu.base import IMUData 

6 

7 

8class PIMAlgorithm(BaseAlgorithm): 

9 """Proportional Integration Mode""" 

10 

11 _algorithm_name = "PIMAlgorithm" 

12 _version = 'v0.1.0' 

13 _aggregation_window = 5 

14 

15 def __init__(self) -> None: 

16 

17 return None 

18 

19 def estimate(self, data: IMUData): 

20 

21 self.values_x = abs(data.x) 

22 self.values_y = abs(data.y) 

23 self.values_z = abs(data.z) 

24 

25 return self 

26 

27 def aggregate(self, 

28 method: str = 'sum' 

29 ): 

30 

31 df = DataFrame( 

32 list(zip(self.data.timestamps, self.values_x, self.values_y, self.values_z)), 

33 columns=['timestamps', 'x', 'y', 'z'] 

34 ) 

35 

36 df['timestamps'] = df[ 

37 'timestamps'].apply(lambda x: x // self.aggregation_window) 

38 

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

40 

41 self.biomarker_agg = df_agg 

42 

43 return self