Coverage for tests/test_chunking.py: 100%

1import numpy as np 

2import pytest 

3from evutils.chunking import window_delta_t, sliding_window, sort_events, get_dt_events 

4from evutils.random import random_events 

5from evutils.types import Event_dtype 

6 

7def test_window_delta_t(): 

8 # Empty array 

9 events = np.array([], dtype=Event_dtype) 

10 chunks = list(window_delta_t(events, delta_t=1000)) 

11 assert len(chunks) == 0 

12 

13 # Normal case 

14 events = random_events(100, start_ts=0, end_ts=30_000) 

15 chunks = list(window_delta_t(events, delta_t=10_000)) 

16 for chunk in chunks: 

17 if len(chunk) > 0: 

18 assert chunk['t'].max() - chunk['t'].min() <= 10_000 

19 

20 # Events exactly on boundary 

21 events = np.array([(0,0,0,0), (10_000,0,0,0), (20_000,0,0,0)], dtype=Event_dtype) 

22 chunks = list(window_delta_t(events, delta_t=10_000)) 

23 assert len(chunks) == 3 

24 assert len(chunks[0]) == 1 

25 assert len(chunks[1]) == 1 

26 assert len(chunks[2]) == 1 

27 

28def test_sliding_window(): 

29 events = np.array([], dtype=Event_dtype) 

30 chunks = list(sliding_window(events)) 

31 assert len(chunks) == 0 

32 

33 events = random_events(100, start_ts=0, end_ts=50_000) 

34 

35 # full_window = False 

36 chunks = list(sliding_window(events, delta_t=10_000, window_size=20_000, full_window=False)) 

37 assert len(chunks) > 0 

38 

39 # full_window = True 

40 chunks_full = list(sliding_window(events, delta_t=10_000, window_size=20_000, full_window=True)) 

41 # Number of full chunks should be less than or equal to total chunks 

42 assert len(chunks_full) <= len(chunks) 

43 

44def test_sort_events(): 

45 events = np.array([], dtype=Event_dtype) 

46 sorted_events = sort_events(events) 

47 assert len(sorted_events) == 0 

48 

49 events = random_events(10) 

50 events['t'] = np.arange(10, 0, -1) 

51 sorted_events = sort_events(events) 

52 assert sorted_events['t'][0] == 1 

53 assert sorted_events['t'][-1] == 10 

54 

55def test_get_dt_events(): 

56 events = np.array([], dtype=Event_dtype) 

57 assert len(get_dt_events(events)) == 0 

58 

59 events = random_events(100, start_ts=0, end_ts=50_000) 

60 sub = get_dt_events(events, dt=10_000) 

61 if len(sub) > 0: 

62 assert (sub['t'] <= events['t'][0] + 10_000).all() 

63 

64 # dt larger than stream 

65 sub2 = get_dt_events(events, dt=100_000) 

66 assert len(sub2) == len(events) 

67 

68def test_stream_delta_t_shape_mirrors_input(): 

69 """A trigger-less stream must yield bare EventArrays (not tuples), and the 

70 windows must reassemble to the input stream. Regression: the generators 

71 used to test the accumulator's always-present trigger buffer and yielded 

72 (events, triggers) tuples unconditionally.""" 

73 from evutils.chunking import stream_delta_t, stream_n_events 

74 from evutils.types import EventArray 

75 

76 events = random_events(5000, start_ts=0, end_ts=100_000) 

77 events = sort_events(events) 

78 src = EventArray.from_aos(events) 

79 

80 for gen in (stream_delta_t(iter([src]), 10_000), 

81 stream_n_events(iter([src]), 700)): 

82 chunks = list(gen) 

83 assert all(not isinstance(c, tuple) for c in chunks) 

84 total = np.concatenate([np.asarray(c) for c in chunks]) 

85 assert len(total) == len(events) 

86 assert np.array_equal(total['t'], events['t'])