Coverage for tests/io/test_dat.py: 85%

1from typing import Any 

2import numpy as np 

3import pytest 

4 

5def test_DAT_writer(tmp_path: Any, test_events: Any) -> None: 

6 from evutils.io import EventReader, EventWriter 

7 

8 p = tmp_path / "events.dat" 

9 with EventWriter(p) as writer: 

10 writer.write(test_events) 

11 

12 assert p.is_file() 

13 assert p.stat().st_size > 0 

14 

15 with EventReader(p) as reader: 

16 events = reader.read() 

17 assert np.array_equal(events, test_events) 

18 

19 

20def test_DAT_matches_expelliarmus(tmp_path: Any, test_events: Any) -> None: 

21 """Our DAT output must be readable byte-for-byte by the reference reader, 

22 and our EventReader must decode it exactly the same as the reference reader.""" 

23 pytest.importorskip("expelliarmus") 

24 from expelliarmus import Wizard # type: ignore 

25 

26 from evutils.io import EventWriter, EventReader 

27 

28 p = tmp_path / "events.dat" 

29 with EventWriter(p) as writer: 

30 writer.write(test_events) 

31 

32 # 1. Read with expelliarmus 

33 exp_arr = Wizard(encoding="dat").read(str(p)) 

34 

35 # 2. Read with evutils 

36 with EventReader(p) as reader: 

37 evutils_arr = reader.read() 

38 assert not isinstance(evutils_arr, tuple) 

39 

40 # Compare them directly 

41 assert len(evutils_arr) == len(exp_arr) 

42 assert np.array_equal(evutils_arr["t"], exp_arr["t"]) 

43 assert np.array_equal(evutils_arr["x"], exp_arr["x"]) 

44 assert np.array_equal(evutils_arr["y"], exp_arr["y"]) 

45 assert np.array_equal(evutils_arr["p"], exp_arr["p"]) 

46 

47 

48def test_DAT_timestamp_overflow(tmp_path: Any) -> None: 

49 """DAT timestamps are 32-bit us on disk (~71 min) 

50 the decoder must extend 

51 them past the wrap.""" 

52 from evutils.io import EventReader, EventWriter 

53 from evutils.types import Event_dtype 

54 

55 t = np.array([2**32 - 30, 2**32 - 10, 2**32 + 10, 2**32 + 30, 2**33 + 5], dtype=np.int64) 

56 ev = np.zeros(len(t), dtype=Event_dtype) 

57 ev['t'] = t 

58 ev['x'] = np.arange(len(t)) 

59 ev['y'] = np.arange(len(t)) 

60 ev['p'] = [0, 1, 0, 1, 1] 

61 

62 p = tmp_path / "overflow.dat" 

63 with EventWriter(p) as w: 

64 w.write(ev) 

65 with EventReader(p) as r: 

66 out = r.read_all() 

67 assert not isinstance(out, tuple) 

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

69 

70 

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

72 """DAT coordinates are 14-bit 

73 extremes must survive.""" 

74 from evutils.io import EventReader, EventWriter 

75 from evutils.types import Event_dtype 

76 

77 ev = np.zeros(4, dtype=Event_dtype) 

78 ev['t'] = [0, 1, 2, 3] 

79 ev['x'] = [0, 2**14 - 1, 5, 2**14 - 2] 

80 ev['y'] = [2**14 - 1, 0, 6, 2**14 - 3] 

81 ev['p'] = [1, 0, 1, 0] 

82 

83 p = tmp_path / "coords.dat" 

84 with EventWriter(p) as w: 

85 w.write(ev) 

86 with EventReader(p) as r: 

87 out = r.read_all() 

88 assert not isinstance(out, tuple) 

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

90 

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

92 from evutils.io import EventReader 

93 p = tmp_path / "empty.dat" 

94 p.touch() 

95 with EventReader(p) as r: 

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

97 

98def test_DAT_header_only(tmp_path: Any) -> None: 

99 from evutils.io import EventReader 

100 p = tmp_path / "header_only.dat" 

101 p.write_bytes(b"% Header\n") 

102 with EventReader(p) as r: 

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

104 

105def test_DAT_unsupported_event_size(tmp_path: Any) -> None: 

106 from evutils.io import EventReader 

107 p = tmp_path / "unsupported_size.dat" 

108 # 0x0C = CD, size 0x10 (unsupported) 

109 p.write_bytes(b"% Header\n\x0c\x10") 

110 with pytest.raises(NotImplementedError, match="DAT event size 16 not supported"): 

111 with EventReader(p) as r: 

112 r.read() 

113 

114def test_DAT_truncated_payload(tmp_path: Any) -> None: 

115 # 8 bytes per event. Give 4 bytes. 

116 from evutils.io import EventReader 

117 p = tmp_path / "truncated.dat" 

118 p.write_bytes(b"% Header\n\x0c\x08\x01\x02\x03\x04") 

119 with EventReader(p) as r: 

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

121 

122def test_DAT_truncated_header_missing_bytes(tmp_path: Any) -> None: 

123 from evutils.io import EventReader 

124 p = tmp_path / "truncated_header.dat" 

125 p.write_bytes(b"% Header\n\x0c") # 1 byte left after header 

126 with EventReader(p) as r: 

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