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]