Coverage for tests/io/test_reader_modes.py: 100%

1"""Windowing-mode tests for EventReader: ``mixed`` and ``all``. 

2 

3``delta_t`` / ``n_events`` modes are exercised throughout the per-format 

4tests; the combined (``mixed``) and whole-file (``all``) modes had no 

5dedicated coverage. Window semantics under test: 

6 

7* n_events cutoff: exactly ``n_events`` events per chunk while available. 

8* delta_t cutoff: half-open window ``[start, start + delta_t)``. 

9* mixed: whichever cutoff is hit first wins, per chunk. 

10* all: one read returns everything; the next read is empty. 

11""" 

12import numpy as np 

13import pytest 

14 

15from typing import Any 

16 

17 

18def _uniform_events(n: int = 1000, dt: int = 1000) -> Any: 

19 """n events at exactly one event per ``dt`` microseconds.""" 

20 ev = np.zeros(n, dtype=np.dtype([('t', np.int64), ('x', np.uint16), ('y', np.uint16), ('p', np.uint8)])) 

21 ev['t'] = np.arange(n, dtype=np.int64) * dt 

22 ev['x'] = np.arange(n) % 1280 

23 ev['y'] = np.arange(n) % 720 

24 ev['p'] = np.arange(n) % 2 

25 return ev 

26 

27 

28@pytest.fixture 

29def uniform_file(tmp_path: Any) -> Any: 

30 from evutils.io import EventWriter 

31 ev = _uniform_events() 

32 p = tmp_path / "uniform.raw" 

33 with EventWriter(p, format="evt2") as w: 

34 w.write(ev) 

35 return p, ev 

36 

37 

38#################################### 

39# mixed mode 

40#################################### 

41 

42def test_mixed_mode_n_events_cuts_first(uniform_file: Any) -> None: 

43 """delta_t generous (100 events worth), n_events tight (20): every chunk 

44 must be cut by the event count.""" 

45 from evutils.io import EventReader 

46 p, ev = uniform_file 

47 with EventReader(p, mode="mixed", delta_t=100_000, n_events=20) as r: 

48 chunks = list(r) 

49 assert all(len(c) == 20 for c in chunks) 

50 assert sum(len(c) for c in chunks) == len(ev) 

51 out = np.concatenate([np.asarray(c) for c in chunks]) 

52 assert np.array_equal(out['t'], ev['t']) 

53 

54 

55def test_mixed_mode_delta_t_cuts_first(uniform_file: Any) -> None: 

56 """n_events generous (500), delta_t tight (5 events worth): every chunk 

57 must span < delta_t and hold 5 events.""" 

58 from evutils.io import EventReader 

59 p, ev = uniform_file 

60 with EventReader(p, mode="mixed", delta_t=5_000, n_events=500) as r: 

61 chunks = [c for c in r if len(c) > 0] 

62 assert all(len(c) == 5 for c in chunks) 

63 for c in chunks: 

64 assert c['t'][-1] - c['t'][0] < 5_000 # half-open window 

65 out = np.concatenate([np.asarray(c) for c in chunks]) 

66 assert np.array_equal(out['t'], ev['t']) 

67 

68 

69def test_mixed_is_default_mode(uniform_file: Any) -> None: 

70 """auto mode with both parameters set resolves to mixed.""" 

71 from evutils.io import EventReader 

72 p, _ = uniform_file 

73 with EventReader(p, delta_t=100_000, n_events=20) as r: 

74 first = r.read() 

75 assert len(first) == 20 

76 

77 

78def test_mixed_mode_alternating_cutoffs(tmp_path: Any) -> None: 

79 """A burst then a gap: the burst chunk is cut by n_events, the sparse 

80 region by delta_t.""" 

81 from evutils.io import EventReader, EventWriter 

82 # 50 events in the first 50 us (burst), then 50 events at 1 ms apart. 

83 t = np.concatenate([np.arange(50), 1_000_000 + np.arange(50) * 1000]) 

84 ev = _uniform_events(100) 

85 ev['t'] = t 

86 p = tmp_path / "burst.raw" 

87 with EventWriter(p, format="evt2") as w: 

88 w.write(ev) 

89 

90 with EventReader(p, mode="mixed", delta_t=10_000, n_events=30) as r: 

91 first = r.read() 

92 # burst: 30 events arrive well inside 10 ms -> n_events cut 

93 assert len(first) == 30 

94 total = len(first) 

95 for c in r: 

96 # sparse tail: 10 ms holds at most 10 events -> delta_t cut 

97 assert len(c) <= 30 

98 total += len(c) 

99 assert total == 100 

100 

101 

102#################################### 

103# all mode 

104#################################### 

105 

106def test_all_mode_single_read(uniform_file: Any) -> None: 

107 from evutils.io import EventReader 

108 p, ev = uniform_file 

109 with EventReader(p, mode="all") as r: 

110 out = r.read() 

111 assert np.array_equal(np.asarray(out), ev) 

112 # stream exhausted: subsequent reads are empty 

113 assert len(r.read()) == 0 

114 

115 

116def test_all_mode_iteration_yields_once(uniform_file: Any) -> None: 

117 from evutils.io import EventReader 

118 p, ev = uniform_file 

119 with EventReader(p, mode="all") as r: 

120 chunks = [c for c in r if len(c) > 0] 

121 assert len(chunks) == 1 

122 assert np.array_equal(np.asarray(chunks[0]), ev) 

123 

124 

125def test_all_mode_matches_read_all(uniform_file: Any) -> None: 

126 from evutils.io import EventReader 

127 p, _ = uniform_file 

128 with EventReader(p, mode="all") as r1, EventReader(p) as r2: 

129 assert np.array_equal(np.asarray(r1.read()), np.asarray(r2.read_all())) 

130 

131 

132#################################### 

133# parameter validation 

134#################################### 

135 

136def test_explicit_mode_requires_parameter(uniform_file: Any) -> None: 

137 from evutils.io import EventReader 

138 p, _ = uniform_file 

139 with pytest.raises(ValueError): 

140 EventReader(p, mode="delta_t") # delta_t missing 

141 with pytest.raises(ValueError): 

142 EventReader(p, mode="n_events") # n_events missing 

143 with pytest.raises(ValueError): 

144 EventReader(p, mode="bogus") 

145 

146 

147def test_nonpositive_windows_rejected(uniform_file: Any) -> None: 

148 from evutils.io import EventReader 

149 p, _ = uniform_file 

150 with pytest.raises(ValueError): 

151 EventReader(p, n_events=0) 

152 with pytest.raises(ValueError): 

153 EventReader(p, delta_t=-1) 

154 with pytest.raises(TypeError): 

155 EventReader(p, delta_t=1.5) # type: ignore[arg-type] 

156 

157 

158def test_delta_t_mode_respects_n_events_param(uniform_file: Any) -> None: 

159 from evutils.io import EventReader 

160 p, ev = uniform_file 

161 with EventReader(p, mode="delta_t", delta_t=100, n_events=50) as r: 

162 assert r._n_events == 50 

163 

164 

165def test_n_events_mode_respects_delta_t_param(uniform_file: Any) -> None: 

166 from evutils.io import EventReader 

167 p, ev = uniform_file 

168 with EventReader(p, mode="n_events", n_events=500, delta_t=5_000) as r: 

169 assert r._delta_t == 5_000 

170 

171 

172#################################### 

173# reuse_buffers (recycled delta_t window buffers) 

174#################################### 

175 

176def _windows(path: Any, fmt_kwargs: dict, **reader_kwargs: Any) -> list: 

177 from evutils.io import EventReader 

178 out = [] 

179 with EventReader(path, delta_t=10_000, **reader_kwargs) as r: 

180 for c in r: 

181 a = np.asarray(c) 

182 out.append(a.copy()) 

183 return out 

184 

185 

186@pytest.mark.parametrize("fmt", ["evt2", "evt3"]) # Python fast path / C parser path 

187def test_reuse_buffers_matches_default(tmp_path: Any, fmt: str) -> None: 

188 """reuse_buffers=True must yield the identical window sequence as the 

189 default (independent buffers), for both delta_t fast paths.""" 

190 from evutils.io import EventWriter 

191 ev = _uniform_events(5000, dt=13) 

192 p = tmp_path / f"reuse.{'raw' if fmt.startswith('evt') else fmt}" 

193 with EventWriter(p, format=fmt) as w: 

194 w.write(ev) 

195 ref = _windows(p, {}) 

196 got = _windows(p, {}, reuse_buffers=True) 

197 assert len(ref) == len(got) 

198 for a, b in zip(ref, got): 

199 assert np.array_equal(a, b) 

200 assert sum(len(a) for a in got) == len(ev) 

201 

202 

203@pytest.mark.parametrize("fmt", ["evt2", "evt3"]) 

204def test_reuse_buffers_async_matches_default(tmp_path: Any, fmt: str) -> None: 

205 """Recycled buffers must stay valid across the async prefetch queue: the 

206 ring is sized to prefetch_depth, so queued windows are never overwritten 

207 before the consumer copies them.""" 

208 from evutils.io import EventWriter 

209 ev = _uniform_events(5000, dt=13) 

210 p = tmp_path / "reuse_async.raw" 

211 with EventWriter(p, format=fmt) as w: 

212 w.write(ev) 

213 ref = _windows(p, {}) 

214 got = _windows(p, {}, reuse_buffers=True, async_read=True, prefetch_depth=3) 

215 assert len(ref) == len(got) 

216 for a, b in zip(ref, got): 

217 assert np.array_equal(a, b) 

218 

219def test_batch_mode(tmp_path: Any) -> None: 

220 from evutils.io import EventWriter, EventReader 

221 from evutils.types import DataBatch 

222 ev = _uniform_events(50, dt=13) 

223 p = tmp_path / "batch_mode.aedat4" 

224 with EventWriter(p, format="aedat") as w: 

225 w.write(ev) 

226 

227 # test iter 

228 with EventReader(p, batch_mode=True, n_events=10) as r: 

229 batches = list(r) 

230 assert len(batches) == 5 

231 assert all(isinstance(b, DataBatch) for b in batches) 

232 assert sum(len(b.events) for b in batches) == 50 

233 

234 # test read 

235 with EventReader(p, batch_mode=True) as r: 

236 b = r.read() 

237 assert isinstance(b, DataBatch) 

238 assert len(b.events) == 50 

239 

240 # test read_all 

241 with EventReader(p, batch_mode=True) as r: 

242 b = r.read_all() 

243 assert isinstance(b, DataBatch) 

244 assert len(b.events) == 50 

245