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

1"""EVT4 decoder tests. 

2 

3EVT4 is a 32-bit-word format modelled on OpenEB's ``evt4`` HAL decoder. Its CD, 

4TIME_HIGH and EXT_TRIGGER words reuse EVT2's bit layout with distinct type codes, 

5so the scalar path is validated against the proven EVT2 decoder. The format's 

6novel mechanic is vectorised CD (CD_VEC_OFF=0xC / CD_VEC_ON=0xD): a base word 

7carrying x/y/ts/polarity followed by a 32-bit validity mask, emitting one event 

8per set bit at ``x = base_x + bit_index`` (same y/ts/polarity). No encoder emits 

9CD_VEC, so it is exercised here with a hand-crafted word stream. 

10""" 

11import numpy as np 

12 

13from typing import Any 

14 

15 

16# ---- EVT4 word crafting (see csrc/evt4.c for the bit layout) -------------- 

17 

18def _time_high(t: int) -> int: 

19 return (0xE << 28) | ((t >> 6) & 0x0FFFFFFF) 

20 

21 

22def _cd(t: int, x: int, y: int, p: int) -> int: 

23 # CD_OFF=0xA / CD_ON=0xB 

24 return ((0xA | (p & 1)) << 28) | ((t & 0x3F) << 22) | ((x & 0x7FF) << 11) | (y & 0x7FF) 

25 

26 

27def _cd_vec_base(t: int, base_x: int, y: int, p: int) -> int: 

28 # CD_VEC_OFF=0xC / CD_VEC_ON=0xD; followed by a 32-bit mask word. 

29 return ((0xC | (p & 1)) << 28) | ((t & 0x3F) << 22) | ((base_x & 0x7FF) << 11) | (y & 0x7FF) 

30 

31 

32def _header_only(tmp_path: Any) -> Any: 

33 """An EVT4 file with a valid header and no payload, ready to append raw words.""" 

34 from evutils.io import EventWriter 

35 p = tmp_path / "evt4_raw.raw" 

36 w = EventWriter(p, format="evt4") 

37 w.init() 

38 w.close() 

39 return p 

40 

41 

42def _append(p: Any, words: Any) -> None: 

43 with open(p, "ab") as f: 

44 f.write(np.asarray(words, dtype=np.uint32).tobytes()) 

45 

46 

47# ---- scalar path: EVT4 must decode identically to EVT2 -------------------- 

48 

49def test_evt4_matches_evt2(tmp_path: Any, test_events: Any) -> None: 

50 """Same events encoded as EVT2 and EVT4 must decode to identical arrays 

51 (both formats share the CD / TIME_HIGH layout).""" 

52 from evutils.io import EventWriter, EventReader 

53 out = {} 

54 for fmt in ("evt2", "evt4"): 

55 p = tmp_path / f"{fmt}.raw" 

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

57 w.write(test_events) 

58 with EventReader(p) as r: 

59 out[fmt] = np.asarray(r.read_all()) 

60 assert np.array_equal(out["evt4"], test_events), "evt4 scalar round-trip differs from input" 

61 assert np.array_equal(out["evt4"], out["evt2"]), "evt4 decode differs from evt2" 

62 

63 

64# ---- vectorised CD: the novel mechanic, hand-crafted ---------------------- 

65 

66def test_evt4_vector_cd(tmp_path: Any) -> None: 

67 """A CD_VEC base + 32-bit mask emits one event per set bit at base_x+bit.""" 

68 from evutils.io import EventReader 

69 p = _header_only(tmp_path) 

70 

71 t, base_x, y, pol = 5, 100, 50, 1 

72 mask = (1 << 0) | (1 << 3) | (1 << 31) # bits 0, 3, 31 set 

73 _append(p, [_time_high(t), _cd_vec_base(t, base_x, y, pol), mask]) 

74 

75 with EventReader(p) as r: 

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

77 

78 assert len(out) == 3, f"expected 3 events from the 3-bit mask, got {len(out)}" 

79 assert np.array_equal(out["x"], [100, 103, 131]), "vector x offsets wrong" 

80 assert np.array_equal(out["y"], [50, 50, 50]), "vector y not held constant" 

81 assert np.array_equal(out["p"], [1, 1, 1]), "vector polarity wrong" 

82 assert np.array_equal(out["t"], [5, 5, 5]), "vector timestamp wrong" 

83 

84 

85def test_evt4_vector_empty_mask(tmp_path: Any) -> None: 

86 """A CD_VEC with an all-zero mask emits nothing but still consumes the mask 

87 word, so a following scalar CD decodes correctly.""" 

88 from evutils.io import EventReader 

89 p = _header_only(tmp_path) 

90 _append(p, [ 

91 _time_high(7), 

92 _cd_vec_base(7, 200, 60, 0), 0x00000000, # empty mask -> no events 

93 _cd(7, 300, 70, 1), # must still decode 

94 ]) 

95 with EventReader(p) as r: 

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

97 assert len(out) == 1, f"empty mask should yield only the trailing CD, got {len(out)}" 

98 assert (int(out["x"][0]), int(out["y"][0]), int(out["p"][0])) == (300, 70, 1) 

99 

100 

101def test_evt4_vector_and_scalar_mixed(tmp_path: Any) -> None: 

102 """Interleaved scalar CD, vector CD (both polarities) and TIME_HIGH decode 

103 in order with correct timestamps.""" 

104 from evutils.io import EventReader 

105 p = _header_only(tmp_path) 

106 _append(p, [ 

107 _time_high(10), 

108 _cd(10, 1, 2, 0), # scalar, p=0 

109 _cd_vec_base(10, 500, 300, 1), (1 << 1) | (1 << 2), # x=501,502 p=1 

110 _cd_vec_base(10, 8, 9, 0), (1 << 0), # x=8 p=0 

111 _cd(10, 3, 4, 1), # scalar, p=1 

112 ]) 

113 with EventReader(p) as r: 

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

115 assert np.array_equal(out["x"], [1, 501, 502, 8, 3]) 

116 assert np.array_equal(out["y"], [2, 300, 300, 9, 4]) 

117 assert np.array_equal(out["p"], [0, 1, 1, 0, 1]) 

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

119 

120 

121def test_evt4_vector_full_mask(tmp_path: Any) -> None: 

122 """A fully-set 32-bit mask emits 32 contiguous events base_x .. base_x+31.""" 

123 from evutils.io import EventReader 

124 p = _header_only(tmp_path) 

125 base_x = 64 

126 _append(p, [_time_high(3), _cd_vec_base(3, base_x, 11, 1), 0xFFFFFFFF]) 

127 with EventReader(p) as r: 

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

129 assert len(out) == 32 

130 assert np.array_equal(out["x"], np.arange(base_x, base_x + 32)) 

131 assert np.all(out["y"] == 11) and np.all(out["p"] == 1) and np.all(out["t"] == 3)