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

1import numpy as np 

2 

3from evutils.types import Event_dtype 

4 

5 

6from typing import Any 

7def test_AER_writer(tmp_path: Any) -> None: 

8 """AER carries no timestamps and only 9-bit coords, so round-trip preserves 

9 x/y/p (with t == 0) for events that fit in 512x512.""" 

10 from evutils.io import EventReader, EventWriter 

11 

12 np.random.seed(7) 

13 N = 1000 

14 ev = np.zeros(N, dtype=Event_dtype) 

15 ev["x"] = np.random.randint(0, 320, N) 

16 ev["y"] = np.random.randint(0, 320, N) 

17 ev["p"] = np.random.randint(0, 2, N) 

18 

19 p = tmp_path / "events.aer" 

20 with EventWriter(p) as writer: 

21 writer.write(ev) 

22 

23 assert p.is_file() 

24 assert p.stat().st_size == N * 4 # 4 bytes/event, no header 

25 

26 with EventReader(p) as reader: 

27 out = reader.read() 

28 

29 assert not isinstance(out, tuple) 

30 assert len(out) == N 

31 assert np.array_equal(out["x"], ev["x"]) 

32 assert np.array_equal(out["y"], ev["y"]) 

33 assert np.array_equal(out["p"], ev["p"]) 

34 assert bool(np.all(out["t"] == 0)) 

35 

36 

37def _write_aer(tmp_path: Any, n: int=1000, seed: int=7) -> Any: 

38 from evutils.io import EventWriter 

39 

40 np.random.seed(seed) 

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

42 ev["x"] = np.random.randint(0, 512, n) 

43 ev["y"] = np.random.randint(0, 512, n) 

44 ev["p"] = np.random.randint(0, 2, n) 

45 p = tmp_path / "events.aer" 

46 with EventWriter(p) as writer: 

47 writer.write(ev) 

48 return p, ev 

49 

50 

51def test_AER_timestamps_sequential(tmp_path: Any) -> None: 

52 """timestamps='sequential' generates t = t_start + i * t_step, carried 

53 across chunks and restarted by reset().""" 

54 from evutils.io import EventReader 

55 

56 p, _ = _write_aer(tmp_path) 

57 with EventReader(p, timestamps="sequential", t_start=100, t_step=5) as r: 

58 out = r.read_all() 

59 assert not isinstance(out, tuple) 

60 assert np.array_equal(out["t"], 100 + 5 * np.arange(1000)) 

61 

62 with EventReader(p, n_events=300, timestamps="sequential") as r: 

63 ts = np.concatenate([np.asarray(c[0] if isinstance(c, tuple) else c)["t"] for c in r]) 

64 assert np.array_equal(ts, np.arange(1000)) 

65 r.reset() 

66 out = r.read_all() 

67 assert not isinstance(out, tuple) 

68 assert np.array_equal(out["t"], np.arange(1000)) 

69 

70 

71def test_AER_timestamps_custom(tmp_path: Any) -> None: 

72 """A user-provided array is assigned positionally, in bulk and chunked.""" 

73 from evutils.io import EventReader 

74 

75 p, _ = _write_aer(tmp_path) 

76 custom = np.sort(np.random.randint(0, 10**9, 1000)) 

77 

78 with EventReader(p, timestamps=custom) as r: 

79 out = r.read_all() 

80 assert not isinstance(out, tuple) 

81 assert np.array_equal(out["t"], custom) 

82 

83 with EventReader(p, n_events=300, timestamps=custom) as r: 

84 ts = np.concatenate([np.asarray(c[0] if isinstance(c, tuple) else c)["t"] for c in r]) 

85 assert np.array_equal(ts, custom) 

86 

87 

88def test_AER_timestamps_validation(tmp_path: Any) -> None: 

89 """Bad mode strings and too-short custom arrays raise ValueError.""" 

90 import pytest 

91 

92 from evutils.io import EventReader 

93 

94 p, _ = _write_aer(tmp_path) 

95 with pytest.raises(ValueError, match="timestamps must be"): 

96 EventReader(p, timestamps="bogus").read_all() 

97 with pytest.raises(ValueError, match="custom timestamps array has"): 

98 EventReader(p, timestamps=np.arange(10)).read_all() 

99 

100def test_AER_empty_file(tmp_path: Any) -> None: 

101 from evutils.io import EventReader 

102 p = tmp_path / "empty.aer" 

103 p.touch() 

104 with EventReader(p) as r: 

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

106 

107def test_AER_truncated_payload(tmp_path: Any) -> None: 

108 from evutils.io import EventReader 

109 p = tmp_path / "truncated.aer" 

110 p.write_bytes(b"\x01\x02\x03") # 3 bytes, AER is 4 bytes 

111 with EventReader(p) as r: 

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