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

1"""HDF5 decoder/encoder tests: DSEC/RVT layout, Prophesee layout, the 

2``ms_to_idx`` millisecond index and DSEC ``t_offset`` handling.""" 

3import numpy as np 

4import pytest 

5 

6 

7from evutils.io import EventReader, EventWriter 

8from evutils.types import Event_dtype 

9 

10from typing import Any 

11 

12h5py = pytest.importorskip("h5py") 

13 

14def make_events(n: int=20_000, t_max: int=200_000, seed: int=3) -> Any: 

15 rng = np.random.default_rng(seed) 

16 ev = np.zeros(n, dtype=Event_dtype) 

17 ev["t"] = np.sort(rng.integers(0, t_max, n)) 

18 ev["x"] = rng.integers(0, 1280, n) 

19 ev["y"] = rng.integers(0, 720, n) 

20 ev["p"] = rng.integers(0, 2, n) 

21 return ev 

22 

23 

24def test_geometry_attrs_roundtrip(tmp_path: Any) -> None: 

25 p = tmp_path / "geo.h5" 

26 with EventWriter(p, width=640, height=480) as w: 

27 w.write(make_events(100)) 

28 with EventReader(p) as r: 

29 r.read() 

30 assert r.shape() == (640, 480) 

31 

32 

33def test_ms_to_idx_random_access(tmp_path: Any) -> None: 

34 """decoder.read(start_ms, end_ms) must return exactly the events with 

35 start_ms*1000 <= t < end_ms*1000.""" 

36 ev = make_events() 

37 p = tmp_path / "idx.h5" 

38 with EventWriter(p) as w: 

39 for part in np.array_split(ev, 7): # index must survive chunked writes 

40 w.write(part) 

41 

42 with EventReader(p) as r: 

43 r.init() 

44 dec: Any = r._file_decoder 

45 for start_ms, end_ms in [(0, 50), (50, 120), (13, 14), (0, -1), (150, 10_000)]: 

46 got = dec.read(start_ms, end_ms) 

47 lo = start_ms * 1000 

48 hi = ev["t"].max() + 1 if end_ms == -1 or end_ms * 1000 > ev["t"].max() \ 

49 else end_ms * 1000 

50 ref = ev[(ev["t"] >= lo) & (ev["t"] < hi)] 

51 assert len(got) == len(ref), (start_ms, end_ms) 

52 assert np.array_equal(got["t"], ref["t"]) 

53 

54 # Past-the-end start is empty; invalid ranges raise. 

55 assert len(dec.read(10**6, -1)) == 0 

56 with pytest.raises(ValueError): 

57 dec.read(-1, 10) 

58 with pytest.raises(ValueError): 

59 dec.read(100, 50) 

60 

61 

62def test_dsec_t_offset(tmp_path: Any) -> None: 

63 """A DSEC-style ``t_offset`` dataset shifts the decoded timestamps.""" 

64 ev = make_events(1000) 

65 p = tmp_path / "dsec.h5" 

66 with EventWriter(p) as w: 

67 w.write(ev) 

68 with h5py.File(p, "r+") as f: 

69 f.create_dataset("t_offset", data=np.int64(1_000_000)) 

70 

71 with EventReader(p) as r: 

72 out = r.read_all() 

73 assert not isinstance(out, tuple) 

74 assert np.array_equal(out["t"], ev["t"].astype(np.int64) + 1_000_000) 

75 

76 

77def _write_prophesee_layout(path: Any, ev: Any, root: Any="CD") -> None: 

78 """Synthesize the Metavision HDF5 layout: compound CD/events dataset.""" 

79 prophesee_dtype = np.dtype([("x", "<u2"), ("y", "<u2"), ("p", "<i2"), ("t", "<i8")]) 

80 rec = np.zeros(len(ev), dtype=prophesee_dtype) 

81 for f in ("x", "y", "p", "t"): 

82 rec[f] = ev[f] 

83 chunks = (min(16384, len(rec)),) 

84 with h5py.File(path, "w") as f: 

85 if root: 

86 f.create_group(root).create_dataset("events", data=rec, chunks=chunks) 

87 else: 

88 f.create_dataset("events", data=rec, chunks=chunks) 

89 

90 

91def test_prophesee_layout(tmp_path: Any) -> None: 

92 """Uncompressed Metavision-layout files (CD/events compound dataset) read 

93 transparently through the same EventReader interface.""" 

94 ev = make_events() 

95 p = tmp_path / "metavision.hdf5" 

96 _write_prophesee_layout(p, ev) 

97 

98 with EventReader(p) as r: 

99 out = np.asarray(r.read_all()) 

100 for f in ("t", "x", "y", "p"): 

101 assert np.array_equal(out[f], ev[f]), f 

102 

103 # Chunked iteration streams the compound dataset too. 

104 with EventReader(p, n_events=1024) as r: 

105 total = sum(len(c) for c in r) 

106 assert total == len(ev) 

107 

108 

109def test_root_level_structured_dataset(tmp_path: Any) -> None: 

110 """A compound 'events' dataset at the root (no CD group) also reads.""" 

111 ev = make_events(500) 

112 p = tmp_path / "flat.h5" 

113 _write_prophesee_layout(p, ev, root=None) 

114 with EventReader(p) as r: 

115 out = r.read_all() 

116 assert not isinstance(out, tuple) 

117 assert np.array_equal(out["t"], ev["t"]) 

118 

119 

120def test_missing_events_errors(tmp_path: Any) -> None: 

121 p = tmp_path / "bogus.h5" 

122 with h5py.File(p, "w") as f: 

123 f.create_dataset("unrelated", data=np.arange(3)) 

124 with pytest.raises(ValueError, match="neither an 'events'"): 

125 EventReader(p).read_all() 

126 

127 

128def test_no_index_read_raises(tmp_path: Any) -> None: 

129 """Millisecond random access needs ms_to_idx; a clear error otherwise.""" 

130 ev = make_events(100) 

131 p = tmp_path / "noidx.hdf5" 

132 _write_prophesee_layout(p, ev) 

133 with EventReader(p) as r: 

134 r.init() 

135 dec: Any = r._file_decoder 

136 with pytest.raises(ValueError, match="ms_to_idx"): 

137 dec.read(0, 10) 

138 

139 

140def test_large_timestamps(tmp_path: Any) -> None: 

141 """int64 timestamps (e.g. absolute epoch microseconds) survive, and the 

142 millisecond index stays small by anchoring at the first event 

143 (ms_to_idx_offset).""" 

144 epoch = 1_663_249_605_734_020 

145 ev = make_events() 

146 ev["t"] += epoch 

147 p = tmp_path / "epoch.h5" 

148 with EventWriter(p) as w: 

149 w.write(ev) 

150 with EventReader(p) as r: 

151 out = r.read_all() 

152 assert not isinstance(out, tuple) 

153 assert np.array_equal(out["t"], ev["t"]) 

154 # Random access uses absolute milliseconds; the offset is transparent. 

155 dec: Any = r._file_decoder 

156 start_ms = (epoch // 1000) + 50 

157 got = dec.read(start_ms, start_ms + 20) 

158 ref = ev[(ev["t"] >= start_ms * 1000) & (ev["t"] < (start_ms + 20) * 1000)] 

159 assert len(got) == len(ref) 

160 assert np.array_equal(got["t"], ref["t"]) 

161 

162def test_HDF5_empty_dataset(tmp_path: Any) -> None: 

163 p = tmp_path / "empty.h5" 

164 with EventWriter(p) as w: 

165 pass # write nothing 

166 with EventReader(p) as r: 

167 out = r.read_all() 

168 assert len(out) == 0