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

1import pytest 

2import numpy as np 

3from evutils.io import EventReader, EventWriter 

4from evutils.types import Event_dtype 

5 

6def test_evt3_71_min_overflow(): 

7 import numpy as np 

8 from evutils.io._native_evt import Evt3Parser, Evt3Input 

9 from evutils.io._native_core import EventSoABuffers, TriggerSoABuffers, parse_step 

10 

11 # Create a sequence of wraps. 

12 # Each wrap is triggered when new_ts_high < local_state.ts_high. 

13 # So we write TIME_HIGH=0xFFF, then TIME_HIGH=0. 

14 # To reach 72 minutes (which is 257 wraps), we need 257 pairs of TIME_HIGH. 

15 

16 words = [] 

17 

18 # 260 wraps = 260 * 16.7 seconds = 4342 seconds = 72.3 minutes 

19 for _ in range(260): 

20 words.append((0x8 << 12) | 0xFFF) # TIME_HIGH = 0xFFF 

21 words.append((0x8 << 12) | 0x000) # TIME_HIGH = 0x000 

22 

23 # Now we are at wrap 260. ts_high_high = 260 * 0x1000000 = 0x104000000 

24 # Let's add a TIME_LOW = 0x123 

25 words.append((0x6 << 12) | 0x123) 

26 # Add an event at x=10, y=20, p=1 

27 words.append((0x0 << 12) | 20) 

28 words.append((0x2 << 12) | 10 | (1 << 11)) 

29 

30 # Pad with 4 words because EVT3 parser needs padding 

31 words.extend([0, 0, 0, 0]) 

32 

33 words = np.array(words, dtype=np.uint16) 

34 

35 parser = Evt3Parser() 

36 ev = EventSoABuffers(100) 

37 ev.c.capacity = 100 

38 tr = TriggerSoABuffers(100) 

39 

40 parse_step(words, 0, Evt3Input, parser, ev, tr, tail_pad=4, word_dtype=np.uint16) 

41 

42 assert ev.size == 1 

43 # Expected timestamp: 0x104000000 (from 260 wraps) + 0 (from ts_high=0) + 0x123 (from ts_low) 

44 expected_ts = (260 * 0x1000000) + 0x123 

45 assert ev.t[0] == expected_ts 

46 assert ev.x[0] == 10 

47 assert ev.y[0] == 20 

48 assert ev.p[0] == 1