Coverage for tests/test_robustness_edge_cases.py: 99%

1import pytest 

2import numpy as np 

3from evutils.types import EventArray, TriggerArray 

4from evutils.types import EventsChecker 

5from evutils.chunking import window_delta_t, sliding_window, stream_n_events, stream_delta_t 

6from evutils.filtering import mask_events 

7from evutils.transforms.functional import normalize_ts 

8from evutils.transforms import drop_random_events 

9from evutils.vis.plot3d import plot_3d, plot_3d_timesurface 

10 

11# --- 1. Chunking Edge Cases --- 

12 

13def test_chunking_zero_delta_t_raises(): 

14 events = EventArray(t=[1, 2, 3], x=[0, 0, 0], y=[0, 0, 0], p=[0, 0, 0]) 

15 with pytest.raises(ValueError, match="delta_t must be positive"): 

16 next(window_delta_t(events, delta_t=0)) 

17 

18 with pytest.raises(ValueError, match="delta_t and window_size must be positive"): 

19 next(sliding_window(events, delta_t=0, window_size=10)) 

20 

21def test_stream_n_events_zero_raises(): 

22 def dummy_stream(): 

23 yield EventArray(t=[1], x=[0], y=[0], p=[0]) 

24 

25 with pytest.raises(ValueError, match="n_events must be positive"): 

26 next(stream_n_events(dummy_stream(), n_events=0)) 

27 

28def test_stream_delta_t_trigger_only_stream(): 

29 # Stream with only triggers, no events 

30 def trigger_stream(): 

31 for i in range(1, 4): 

32 ev = EventArray(t=[], x=[], y=[], p=[]) 

33 tr = TriggerArray(t=[i * 1000], p=[1], id=[1]) 

34 yield ev, tr 

35 

36 chunks = list(stream_delta_t(trigger_stream(), delta_t=1000)) 

37 # It should yield chunks based on trigger timestamps 

38 assert len(chunks) > 0 

39 # Last chunk should have triggers 

40 assert len(chunks[-1][1]) > 0 

41 

42 

43# --- 2. Vis Empty Array Edge Cases --- 

44 

45def test_vis_empty_arrays(): 

46 import matplotlib.pyplot as plt 

47 empty_events = EventArray.empty() 

48 

49 # These should not crash 

50 fig, ax = plot_3d(empty_events) 

51 assert ax is not None 

52 plt.close(fig) 

53 

54 fig, ax = plot_3d_timesurface(empty_events) 

55 assert ax is not None 

56 plt.close(fig) 

57 

58 # Open3D 

59 pytest.importorskip("open3d") 

60 from evutils.vis.open3d import o3d_draw_events 

61 o3d_draw_events(empty_events) 

62 

63 

64# --- 3. Types and Utils Edge Cases --- 

65 

66def test_event_array_length_mismatch(): 

67 with pytest.raises(ValueError, match="Length mismatch"): 

68 EventArray(t=[1, 2], x=[1], y=[1, 2], p=[1, 2]) 

69 

70def test_trigger_array_length_mismatch(): 

71 with pytest.raises(ValueError, match="Length mismatch"): 

72 TriggerArray(t=[1, 2], p=[1], id=[1, 2]) 

73 

74def test_event_array_flattening(): 

75 # Multidimensional inputs should be flattened gracefully 

76 events = EventArray(t=[[1, 2]], x=[[1, 2]], y=[[1, 2]], p=[[0, 1]]) 

77 assert len(events) == 2 

78 assert events.t.ndim == 1 

79 

80def test_events_checker_invalid_type(): 

81 with pytest.raises(TypeError, match="events must be a NumPy array or SoaArray"): 

82 EventsChecker([1, 2, 3]) 

83 

84def test_empty_list_indexing(): 

85 events = EventArray(t=[1], x=[1], y=[1], p=[1]) 

86 with pytest.raises(ValueError, match="empty list of fields"): 

87 events[[]] 

88 

89 

90# --- 4. Processing and Augment Edge Cases --- 

91 

92def test_drop_random_events_nan(): 

93 events = EventArray(t=[1, 2], x=[1, 2], y=[1, 2], p=[0, 1]).to_numpy() 

94 with pytest.raises(ValueError, match="drop_rate must be between 0 and 1"): 

95 drop_random_events(events, drop_rate=np.nan) 

96 

97def test_mask_events_negative_coordinates(): 

98 # If using a custom signed dtype 

99 custom_dtype = np.dtype([('t', np.int64), ('x', np.int16), ('y', np.int16), ('p', np.uint8)]) 

100 events = np.array([(100, -5, 10, 1), (200, 10, 10, 1)], dtype=custom_dtype) 

101 mask = np.ones((20, 20), dtype=np.uint8) 

102 

103 with pytest.raises(ValueError, match="non-negative"): 

104 mask_events(events, mask) 

105 

106def test_normalize_ts_readonly(): 

107 events = EventArray(t=[100, 200], x=[1, 2], y=[1, 2], p=[0, 1]).to_numpy() 

108 events.flags.writeable = False # Simulate read-only mmap view 

109 

110 # Should not crash, but return a normalized copy 

111 norm_events = normalize_ts(events) 

112 assert norm_events['t'][0] == 0 

113 assert norm_events['t'][1] == 100 

114 # Original is untouched 

115 assert events['t'][0] == 100 

116 

117def test_normalize_ts_unstructured(): 

118 events = np.array([[100, 1, 1, 1]]) 

119 with pytest.raises(TypeError, match="Unsupported event format"): 

120 normalize_ts(events)