Coverage for tests / test_activity_time_above_thr.py: 100%

1import os 

2import pytest 

3from datetime import datetime 

4 

5from numpy import arange, ones, zeros 

6from pandas import read_csv 

7from physiodsp.sensors.imu.base import IMUData 

8from physiodsp.activity.time_above_thr import TimeAboveThr, TimeAboveThrSettings 

9 

10 

11test_folder_path = os.path.dirname(os.path.realpath(__file__)) 

12 

13 

14@pytest.mark.parametrize( 

15 "n_samples,fs,threshold", 

16 [ 

17 (128, 32, 0.1), 

18 (256, 64, 0.2), 

19 (256, 32, 0.05), 

20 ] 

21) 

22def test_activity_time_above_thr(n_samples, fs, threshold): 

23 """Test TimeAboveThr algorithm with various parameters""" 

24 

25 df = read_csv(os.path.join(test_folder_path, "accelerometer.csv"), usecols=["x", "y", "z"]) 

26 

27 timestamp_start = datetime.now().timestamp() 

28 timestamps = timestamp_start + arange(start=0, step=1/fs, stop=int(n_samples/fs)) 

29 

30 imu_data = IMUData( 

31 timestamps=timestamps, 

32 x=df.x.values[:n_samples], 

33 y=df.y.values[:n_samples], 

34 z=df.z.values[:n_samples], 

35 fs=fs 

36 ) 

37 

38 tat_processor = TimeAboveThr(settings=TimeAboveThrSettings(threshold=threshold)).run(imu_data) 

39 tat_1s = tat_processor.values 

40 

41 # Check output structure 

42 assert "timestamps" in tat_1s.columns 

43 assert "x" in tat_1s.columns 

44 assert "y" in tat_1s.columns 

45 assert "z" in tat_1s.columns 

46 

47 # Check output length (rolling window with 1s step) 

48 assert len(tat_1s) > 0 

49 assert len(tat_1s) == int(n_samples / fs) - 1 

50 

51 # Values should be counts (non-negative integers) 

52 assert (tat_1s["x"] >= 0).all() 

53 assert (tat_1s["y"] >= 0).all() 

54 assert (tat_1s["z"] >= 0).all() 

55 

56 

57def test_time_above_thr_all_above_threshold(): 

58 """Test when all samples are above threshold""" 

59 

60 n_samples = 128 

61 fs = 32 

62 threshold = 0.01 # Very low threshold 

63 

64 timestamp_start = datetime.now().timestamp() 

65 timestamps = timestamp_start + arange(start=0, step=1/fs, stop=int(n_samples/fs)) 

66 

67 # Create data with values well above threshold 

68 x_data = ones(n_samples) * 1.0 

69 y_data = ones(n_samples) * 1.0 

70 z_data = ones(n_samples) * 1.0 

71 

72 imu_data = IMUData( 

73 timestamps=timestamps, 

74 x=x_data, 

75 y=y_data, 

76 z=z_data, 

77 fs=fs 

78 ) 

79 

80 tat_processor = TimeAboveThr(settings=TimeAboveThrSettings(threshold=threshold)).run(imu_data) 

81 tat_1s = tat_processor.values 

82 

83 # All samples should be above threshold 

84 # With 1s windows at 32 Hz, each window should have 32 samples above threshold 

85 assert (tat_1s["x"] == fs).all() 

86 assert (tat_1s["y"] == fs).all() 

87 assert (tat_1s["z"] == fs).all() 

88 

89 

90def test_time_above_thr_all_below_threshold(): 

91 """Test when all samples are below threshold""" 

92 

93 n_samples = 128 

94 fs = 32 

95 threshold = 10.0 # Very high threshold 

96 

97 timestamp_start = datetime.now().timestamp() 

98 timestamps = timestamp_start + arange(start=0, step=1/fs, stop=int(n_samples/fs)) 

99 

100 # Create data with values well below threshold 

101 x_data = zeros(n_samples) 

102 y_data = zeros(n_samples) 

103 z_data = zeros(n_samples) 

104 

105 imu_data = IMUData( 

106 timestamps=timestamps, 

107 x=x_data, 

108 y=y_data, 

109 z=z_data, 

110 fs=fs 

111 ) 

112 

113 tat_processor = TimeAboveThr(settings=TimeAboveThrSettings(threshold=threshold)).run(imu_data) 

114 tat_1s = tat_processor.values 

115 

116 # No samples should be above threshold 

117 assert (tat_1s["x"] == 0).all() 

118 assert (tat_1s["y"] == 0).all() 

119 assert (tat_1s["z"] == 0).all() 

120 

121 

122def test_time_above_thr_custom_settings(): 

123 """Test TimeAboveThr with custom settings""" 

124 

125 n_samples = 256 

126 fs = 64 

127 window_len = 2 

128 aggregation_window = 120 

129 threshold = 0.15 

130 

131 timestamp_start = datetime.now().timestamp() 

132 timestamps = timestamp_start + arange(start=0, step=1/fs, stop=int(n_samples/fs)) 

133 

134 df = read_csv(os.path.join(test_folder_path, "accelerometer.csv"), usecols=["x", "y", "z"]) 

135 

136 imu_data = IMUData( 

137 timestamps=timestamps, 

138 x=df.x.values[:n_samples], 

139 y=df.y.values[:n_samples], 

140 z=df.z.values[:n_samples], 

141 fs=fs 

142 ) 

143 

144 settings = TimeAboveThrSettings( 

145 window_len=window_len, 

146 aggregation_window=aggregation_window, 

147 threshold=threshold 

148 ) 

149 

150 tat_processor = TimeAboveThr(settings=settings).run(imu_data) 

151 tat_results = tat_processor.values 

152 

153 # Check that output is generated 

154 assert len(tat_results) > 0 

155 

156 # With 2s window at 64 Hz, each window should have 128 samples 

157 # Number of windows should be approximately n_samples / (window_len * fs) 

158 expected_windows = int(n_samples / (window_len * fs)) - 1 

159 assert len(tat_results) == expected_windows 

160 

161 # All columns should be present 

162 assert "timestamps" in tat_results.columns 

163 assert "x" in tat_results.columns 

164 assert "y" in tat_results.columns 

165 assert "z" in tat_results.columns 

166 

167 

168def test_time_above_thr_algorithm_properties(): 

169 """Test basic properties of TimeAboveThr algorithm""" 

170 

171 tat = TimeAboveThr() 

172 

173 # Check algorithm metadata 

174 assert tat.algorithm_name == "TimeAboveThrAlgorithm" 

175 assert tat.version == "v0.1.0" 

176 

177 # Check default settings 

178 default_settings = TimeAboveThrSettings() 

179 assert default_settings.window_len == 1 

180 assert default_settings.aggregation_window == 60 

181 assert default_settings.threshold == 0.1 

182 

183 

184def test_time_above_thr_output_structure(): 

185 """Test output DataFrame structure""" 

186 

187 n_samples = 128 

188 fs = 32 

189 

190 timestamp_start = datetime.now().timestamp() 

191 timestamps = timestamp_start + arange(start=0, step=1/fs, stop=int(n_samples/fs)) 

192 

193 df = read_csv(os.path.join(test_folder_path, "accelerometer.csv"), usecols=["x", "y", "z"]) 

194 

195 imu_data = IMUData( 

196 timestamps=timestamps, 

197 x=df.x.values[:n_samples], 

198 y=df.y.values[:n_samples], 

199 z=df.z.values[:n_samples], 

200 fs=fs 

201 ) 

202 

203 tat_processor = TimeAboveThr().run(imu_data) 

204 

205 # Check that values attribute exists 

206 assert hasattr(tat_processor, 'values') 

207 assert tat_processor.values is not None 

208 

209 # Check column data types 

210 assert tat_processor.values["timestamps"].dtype in ['float64', 'float32'] 

211 assert tat_processor.values["x"].dtype in ['int64', 'int32', 'float64', 'float32'] 

212 assert tat_processor.values["y"].dtype in ['int64', 'int32', 'float64', 'float32'] 

213 assert tat_processor.values["z"].dtype in ['int64', 'int32', 'float64', 'float32']