Coverage for tests/test_transforms.py: 100%
149 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-18 05:24 +0000
1import numpy as np
2import pytest
3from evutils.transforms import (
4 drop_random_events,
5 Compose,
6 DropEvent,
7 DropRandomEvents,
8 DropEventByTime,
9 RandomFlipLR,
10 SpatialJitter,
11 TimeSkew,
12 TimeNormalize,
13 TimeJitter,
14 RefractoryPeriod,
15)
16from evutils.transforms.functional import normalize_ts
17from evutils.types import Event_dtype, EventArray
20def test_normalize_ts_functional():
21 events = np.array(
22 [(100, 0, 0, 1), (200, 1, 1, 1), (300, 2, 2, 0)],
23 dtype=Event_dtype
24 )
26 # Normal case -- earliest event shifts to 0; input untouched (non-mutating).
27 norm = normalize_ts(events)
28 assert norm['t'].tolist() == [0, 100, 200]
29 assert events['t'].tolist() == [100, 200, 300]
31 # start_ts != 0
32 assert normalize_ts(events, start_ts=50)['t'].tolist() == [50, 150, 250]
34 # Empty array
35 assert len(normalize_ts(np.array([], dtype=Event_dtype))) == 0
37 # EventArray in -> EventArray out
38 ea = EventArray(t=[100, 200], x=[1, 2], y=[1, 2], p=[0, 1])
39 out = normalize_ts(ea)
40 assert isinstance(out, EventArray)
41 assert out.t.tolist() == [0, 100]
44def test_time_normalize_transform():
45 ea = EventArray(t=[100, 200, 300], x=[0, 1, 2], y=[0, 1, 2], p=[1, 1, 0])
46 assert TimeNormalize()(ea).t.tolist() == [0, 100, 200]
47 assert TimeNormalize(start_ts=10)(ea).t.tolist() == [10, 110, 210]
48 # Composable alongside other time transforms.
49 out = Compose([TimeNormalize(), TimeSkew(coefficient=1.0, offset=0)])(ea)
50 assert out.t.tolist() == [0, 100, 200]
52def test_drop_random_events():
53 events = np.array(
54 [(i, i, i*100, 1) for i in range(100)],
55 dtype=Event_dtype
56 )
58 # Normal case
59 dropped = drop_random_events(events, drop_rate=0.1)
60 assert 70 <= len(dropped) <= 100
61 # Survivors must stay in temporal order
62 assert np.all(np.diff(dropped['t']) >= 0)
64 # Value errors
65 with pytest.raises(ValueError, match="drop_rate must be between 0 and 1"):
66 drop_random_events(events, drop_rate=0.0)
68 with pytest.raises(ValueError, match="drop_rate must be between 0 and 1"):
69 drop_random_events(events, drop_rate=1.0)
71 with pytest.raises(ValueError, match="drop_rate must be between 0 and 1"):
72 drop_random_events(events, drop_rate=-0.1)
74 with pytest.raises(ValueError, match="drop_rate must be between 0 and 1"):
75 drop_random_events(events, drop_rate=1.5)
77 # Empty array
78 empty_events = np.array([], dtype=Event_dtype)
79 assert len(drop_random_events(empty_events, drop_rate=0.5)) == 0
82def test_compose_and_transforms():
83 events = EventArray(
84 t=np.arange(100),
85 x=np.arange(100),
86 y=np.arange(100),
87 p=np.zeros(100, dtype=np.uint8)
88 )
90 # 1. Test single transform standalone
91 transform = DropEvent(p=0.1)
92 dropped = transform(events)
93 # The JIT drop rate is binomial, so it won't be exactly 90
94 assert 70 <= len(dropped) <= 100
95 assert isinstance(dropped, EventArray)
97 # 2. Test compose with multiple transforms
98 pipeline = Compose([
99 DropEvent(p=0.1),
100 DropEvent(p=0.1)
101 ])
103 # Check execution plan
104 assert len(pipeline._execution_plan) == 1
105 assert pipeline._execution_plan[0][0] == "jit"
106 assert len(pipeline._execution_plan[0][1]) == 2
108 dropped_twice = pipeline(events)
109 assert 60 <= len(dropped_twice) <= 100
110 assert isinstance(dropped_twice, EventArray)
112 # 3. Test interop with standard callables
113 def dummy_transform(evs):
114 # A non-JIT transform
115 return evs
117 pipeline_mixed = Compose([
118 DropEvent(p=0.1),
119 dummy_transform,
120 DropEvent(p=0.1)
121 ])
123 # Execution plan should be jit -> standard -> jit
124 assert len(pipeline_mixed._execution_plan) == 3
125 assert pipeline_mixed._execution_plan[0][0] == "jit"
126 assert pipeline_mixed._execution_plan[1][0] == "standard"
127 assert pipeline_mixed._execution_plan[2][0] == "jit"
129 dropped_mixed = pipeline_mixed(events)
130 assert 60 <= len(dropped_mixed) <= 100
131 assert isinstance(dropped_mixed, EventArray)
133def test_target_transformation():
134 from evutils.transforms import Transform
136 # Custom transform that modifies target
137 class DummyTargetCrop(Transform):
138 def _forward_jit(self, t, x, y, p):
139 return t, x, y, p
141 def _transform_target(self, target):
142 if isinstance(target, dict) and "bbox" in target:
143 target["bbox"] = [v - 10 for v in target["bbox"]]
144 return target
146 events = EventArray(t=[1], x=[1], y=[1], p=[1])
147 target = {"class": "car", "bbox": [50, 50, 100, 100]}
149 # 1. Standalone
150 transform = DummyTargetCrop()
151 out_events, out_target = transform(events, target=target.copy())
152 assert out_target["bbox"] == [40, 40, 90, 90]
154 # 2. Compose
155 pipeline = Compose([
156 DropEvent(p=0.1),
157 DummyTargetCrop()
158 ])
160 out_events, out_target = pipeline(events, target=target.copy())
161 assert out_target["bbox"] == [40, 40, 90, 90]
164# --------------------------------------------------------------------------- #
165# New tonic-style transforms
166# --------------------------------------------------------------------------- #
168def _make_events(n=200, w=64, h=48):
169 """Deterministic-ish structured event array spread across a small sensor."""
170 return np.array(
171 [(i * 10, i % w, i % h, i % 2) for i in range(n)],
172 dtype=Event_dtype,
173 )
176def test_drop_event_rename_and_validation():
177 # Alias points at the same class.
178 assert DropRandomEvents is DropEvent
180 events = _make_events()
181 dropped = DropEvent(p=0.2)(events)
182 assert 130 <= len(dropped) <= 200
183 # p=0 is a no-op (tonic-compatible), not an error.
184 assert len(DropEvent(p=0.0)(events)) == len(events)
186 for bad in (1.0, 1.5, -0.1, np.nan):
187 with pytest.raises(ValueError, match=r"p must be in"):
188 DropEvent(p=bad)
190 # Tuple range samples a valid probability.
191 ranged = DropEvent(p=(0.1, 0.3))(events)
192 assert 100 <= len(ranged) <= 200
195def test_random_flip_lr():
196 events = EventArray(t=[0, 1, 2], x=[0, 10, 63], y=[1, 2, 3], p=[0, 1, 0])
197 flipped = RandomFlipLR(sensor_size=(64, 48, 2), p=1.0)(events)
198 assert list(flipped.x) == [63, 53, 0] # width - 1 - x
199 assert list(flipped.y) == [1, 2, 3] # y untouched
200 assert flipped.x.dtype == np.uint16
201 # p=0 never flips.
202 assert list(RandomFlipLR(sensor_size=(64, 48, 2), p=0.0)(events).x) == [0, 10, 63]
205def test_time_skew():
206 events = EventArray(t=[0, 100, 200], x=[1, 2, 3], y=[1, 2, 3], p=[1, 0, 1])
207 out = TimeSkew(coefficient=2.0, offset=10)(events)
208 assert list(out.t) == [10, 210, 410]
209 assert out.t.dtype == np.int64
212def test_time_jitter_clip_and_sort():
213 events = EventArray(
214 t=np.arange(500) * 100,
215 x=np.zeros(500), y=np.zeros(500), p=np.zeros(500),
216 )
217 out = TimeJitter(std=50.0, clip_negative=True, sort_timestamps=True)(events)
218 assert np.all(out.t >= 0)
219 assert np.all(np.diff(out.t) >= 0) # sorted
222def test_spatial_jitter_clip_keeps_in_bounds():
223 events = EventArray(
224 t=np.arange(1000),
225 x=np.full(1000, 32), y=np.full(1000, 24), p=np.zeros(1000),
226 )
227 out = SpatialJitter(sensor_size=(64, 48, 2), var_x=25.0, var_y=25.0,
228 clip_outliers=True)(events)
229 assert np.all(out.x < 64) and np.all(out.y < 48)
230 assert len(out) <= 1000
233def test_refractory_period():
234 # Two pixels. Pixel (0,0) fires at 20,25,200; pixel (1,1) at 21,22.
235 events = EventArray(
236 t=[20, 21, 22, 25, 200],
237 x=[0, 1, 1, 0, 0],
238 y=[0, 1, 1, 0, 0],
239 p=[1, 1, 1, 1, 1],
240 )
241 out = RefractoryPeriod(delta=10)(events)
242 # (0,0): keep t=20 (first), drop t=25 (gap 5<=10), keep t=200 (gap 175>10).
243 # (1,1): keep t=21 (first), drop t=22 (gap 1<=10).
244 assert sorted(out.t.tolist()) == [20, 21, 200]
247def test_drop_by_time_removes_a_window():
248 events = _make_events(n=300)
249 out = DropEventByTime(duration_ratio=0.3)(events)
250 assert len(out) < len(events)
251 # duration_ratio=0 is a no-op.
252 assert len(DropEventByTime(duration_ratio=0.0)(events)) == len(events)
255def test_ndarray_and_eventarray_dispatch():
256 """Same transform works on structured ndarray and EventArray alike."""
257 aos = _make_events(n=50)
258 soa = EventArray.from_aos(aos)
259 skew = TimeSkew(coefficient=3.0)
260 assert isinstance(skew(aos), np.ndarray)
261 assert isinstance(skew(soa), EventArray)
262 np.testing.assert_array_equal(skew(aos)["t"], skew(soa).t)
265def test_metadata_propagates_through_transforms():
266 events = EventArray(t=[0, 1, 2], x=[0, 10, 63], y=[1, 2, 3], p=[0, 1, 0],
267 metadata={"sensor_size": (64, 48)})
268 # Single transform keeps metadata.
269 out = TimeSkew(coefficient=2.0)(events)
270 assert out.sensor_size == (64, 48)
271 # Through a Compose block too.
272 out2 = Compose([TimeSkew(coefficient=2.0), DropEvent(p=0.1)])(events)
273 assert out2.sensor_size == (64, 48)
276def test_spatial_transform_uses_events_sensor_size():
277 events = EventArray(t=[0, 1, 2], x=[0, 10, 63], y=[1, 2, 3], p=[0, 1, 0],
278 metadata={"sensor_size": (64, 48)})
279 # No explicit sensor_size: falls back to events metadata (standalone).
280 assert list(RandomFlipLR(p=1.0)(events).x) == [63, 53, 0]
281 # And inside Compose.
282 out = Compose([RandomFlipLR(p=1.0)])(events)
283 assert list(out.x) == [63, 53, 0]
284 # Explicit sensor_size wins over metadata.
285 assert list(RandomFlipLR(sensor_size=(100, 50), p=1.0)(events).x) == [99, 89, 36]
288def test_spatial_transform_without_sensor_size_raises():
289 events = EventArray(t=[0], x=[1], y=[1], p=[0]) # no metadata
290 with pytest.raises(ValueError, match="sensor_size"):
291 RandomFlipLR(p=1.0)(events)
294def test_compose_survives_emptying_midblock():
295 """A drop that empties the stream must not crash a later kernel."""
296 events = _make_events(n=100)
297 pipeline = Compose([
298 DropEvent(p=0.999999), # almost certainly empties the stream
299 RefractoryPeriod(delta=10), # would call x.max() on empty input
300 ])
301 out = pipeline(events) # must not raise
302 assert len(out) <= len(events)