Coverage for physiodsp / base.py: 86%

1from abc import ABC 

2from pandas import DataFrame 

3from numpy import ndarray 

4 

5 

6class BaseAlgorithm(ABC): 

7 

8 # Class Attributes 

9 _algorithm_name = 'BaseAlgorithm' 

10 _version = 'v0.1.0' 

11 _window_len = 1 

12 _aggregation_window = 60 

13 

14 def __init__(self) -> None: 

15 return None 

16 

17 @property 

18 def algorithm_name(self) -> str: 

19 """Algorithm Name""" 

20 return self._algorithm_name 

21 

22 @property 

23 def version(self) -> str: 

24 """Algorithm Version""" 

25 return self._version 

26 

27 @property 

28 def window_len(self) -> int: 

29 """Window lenght in seconds""" 

30 return self._window_len 

31 

32 @property 

33 def aggregation_window(self) -> int: 

34 """Aggregation Window in seconds""" 

35 return self._aggregation_window 

36 

37 @classmethod 

38 def preprocess(self): 

39 raise NotImplementedError 

40 

41 @classmethod 

42 def run(self): 

43 raise NotImplementedError 

44 

45 @classmethod 

46 def aggregate(self, 

47 timestamps: ndarray, 

48 values: ndarray, 

49 method: str = 'mean' 

50 ): 

51 

52 df = DataFrame( 

53 list(zip(timestamps, values)), 

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

55 ) 

56 

57 df['timestamps'] = df[ 

58 'timestamps'].apply(lambda x: (x // self._aggregation_window) * self._aggregation_window) 

59 

60 if method == 'mean': 

61 df_agg = df.groupby('timestamps')[ 

62 'values'].mean().reset_index(drop=False) 

63 

64 self.biomarker_agg = df_agg 

65 return self