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

1import numpy as np 

2import pytest 

3from evutils.io import EventReader, EventWriter 

4from evutils.io._source import StreamSource 

5import io 

6 

7def test_trigger_boundary_sync(tmp_path): 

8 # Triggers and events at exactly the window boundary should be in the same chunk 

9 p = tmp_path / "boundary.raw" 

10 

11 # 2 events: t=0 and t=1000 

12 ev = np.zeros(2, dtype=np.dtype([('t', np.int64), ('x', np.uint16), ('y', np.uint16), ('p', np.uint8)])) 

13 ev['t'] = [0, 1000] 

14 

15 with EventWriter(p, format="evt3") as w: 

16 w.write(ev) 

17 

18 # Append a trigger at t=1000 

19 # EVT3 trigger: TIME_HIGH, TIME_LOW, TRIGGER 

20 t = 1000 

21 words = np.array([ 

22 0x8000 | ((t >> 12) & 0xFFF), 

23 0x6000 | (t & 0xFFF), 

24 0xA000 | ((1 & 0xF) << 8) | 1, 

25 ], dtype=np.uint16) 

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

27 f.write(words.tobytes()) 

28 

29 with EventReader(p, ext_trigger=True, mode="delta_t", delta_t=1000) as r: 

30 chunks = list(r) 

31 

32 # Chunk 1: [0, 1000) 

33 ev1, tr1 = chunks[0] 

34 assert len(ev1) == 1 and ev1.t[0] == 0 

35 assert len(tr1) == 0 # Trigger at 1000 should NOT be here 

36 

37 # Chunk 2: [1000, 2000) 

38 ev2, tr2 = chunks[1] 

39 assert len(ev2) == 1 and ev2.t[0] == 1000 

40 assert len(tr2) == 1 and tr2.t[0] == 1000 # Trigger at 1000 should be here 

41 

42 

43 

44 

45def test_extreme_playback_speed(tmp_path): 

46 p = tmp_path / "extreme.raw" 

47 ev = np.zeros(2, dtype=np.dtype([('t', np.int64), ('x', np.uint16), ('y', np.uint16), ('p', np.uint8)])) 

48 ev['t'] = [0, 1_000_000] # 1 second apart 

49 with EventWriter(p, format="evt3") as w: 

50 w.write(ev) 

51 

52 # extremely fast playback speed should not sleep 

53 import time 

54 start = time.perf_counter() 

55 with EventReader(p, mode="all", real_time=True, playback_speed=1e6) as r: 

56 out = r.read() 

57 elapsed = time.perf_counter() - start 

58 assert elapsed < 0.5 # Should take well under 1 second