Coverage for physiodsp / base.py: 88%

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 length 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 def preprocess(self): 

38 raise NotImplementedError 

39 

40 def run(self): 

41 raise NotImplementedError 

42 

43 def aggregate(self, 

44 timestamps: ndarray, 

45 values: ndarray, 

46 method: str = 'mean' 

47 ): 

48 

49 df = DataFrame({ 

50 'timestamps': timestamps, 

51 'values': values 

52 }) 

53 

54 df['timestamps'] = df[ 

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

56 

57 if method == 'mean': 

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

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

60 

61 self.biomarker_agg = df_agg 

62 return self