Coverage for tests/io/test_aedat.py: 100%
116 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-15 09:33 +0000
1"""AEDAT 1.0 / 2.0 / 3.1 / 4.0 decoder tests against synthesized files.
3The writers below build byte-exact files following the iniVation file-format
4documentation (v1-v3, big-endian jAER layouts) and the DV framework layout
5(v4, FlatBuffers), so the decoder is checked against the spec rather than
6against itself.
7"""
8import numpy as np
9import pytest
11from aedat_synth import make_aedat1, make_aedat2, make_aedat3, make_aedat4 # type: ignore
12from evutils.io import EventReader
15from typing import Any
16def random_events(n: int, width: int, height: int, seed: int=0) -> tuple[Any, Any, Any, Any]:
17 rng = np.random.default_rng(seed)
18 t = np.sort(rng.integers(0, 1_000_000, n))
19 x = rng.integers(0, width, n)
20 y = rng.integers(0, height, n)
21 p = rng.integers(0, 2, n)
22 return t, x, y, p
25def check(out: Any, t: Any, x: Any, y: Any, p: Any) -> None:
26 assert not isinstance(out, tuple)
27 assert len(out) == len(t)
28 assert np.array_equal(out["t"], t)
29 assert np.array_equal(out["x"], x)
30 assert np.array_equal(out["y"], y)
31 assert np.array_equal(out["p"], p)
34# --------------------------------------------------------------------------- #
35# Tests
36# --------------------------------------------------------------------------- #
37def test_aedat1(tmp_path: Any) -> None:
38 t, x, y, p = random_events(500, 128, 128)
39 f = tmp_path / "v1.aedat"
40 f.write_bytes(make_aedat1(t, x, y, p))
41 with EventReader(f) as r:
42 check(r.read_all(), t, x, y, p)
43 assert r.shape() == (128, 128)
46def test_aedat1_headerless(tmp_path: Any) -> None:
47 """A bare-'#' or headerless file defaults to AEDAT 1.0 (jAER convention)."""
48 t, x, y, p = random_events(100, 128, 128, seed=1)
49 f = tmp_path / "v1_bare.aedat"
50 f.write_bytes(make_aedat1(t, x, y, p, header=False))
51 with EventReader(f) as r:
52 check(r.read_all(), t, x, y, p)
55def test_aedat2_davis_skips_aps(tmp_path: Any) -> None:
56 t, x, y, p = random_events(500, 240, 180, seed=2)
57 f = tmp_path / "v2.aedat"
58 f.write_bytes(make_aedat2(t, x, y, p, aps_every=10))
59 with EventReader(f) as r:
60 check(r.read_all(), t, x, y, p)
61 assert r.shape() == (240, 180)
64def test_aedat2_timestamp_wrap(tmp_path: Any) -> None:
65 """32-bit µs timestamps wrap
66 the decoder must extend them to 64 bits."""
67 t32 = np.array([2**32 - 20, 2**32 - 10, 5, 15], dtype=np.uint64)
68 x = np.array([1, 2, 3, 4])
69 y = np.array([5, 6, 7, 8])
70 p = np.array([0, 1, 0, 1])
71 f = tmp_path / "wrap.aedat"
72 f.write_bytes(make_aedat2(t32 & 0xFFFFFFFF, x, y, p))
73 with EventReader(f) as r:
74 out = r.read_all()
75 assert not isinstance(out, tuple)
76 expected = np.array([2**32 - 20, 2**32 - 10, 2**32 + 5, 2**32 + 15], dtype=np.int64)
77 assert np.array_equal(out["t"], expected)
80def test_aedat3(tmp_path: Any) -> None:
81 t, x, y, p = random_events(500, 346, 260, seed=3)
82 f = tmp_path / "v3.aedat"
83 f.write_bytes(make_aedat3(t, x, y, p))
84 with EventReader(f) as r:
85 check(r.read_all(), t, x, y, p)
88def test_aedat3_ts_overflow(tmp_path: Any) -> None:
89 """The packet header's TS-overflow counter extends the 31-bit timestamps."""
90 t, x, y, p = random_events(8, 346, 260, seed=4)
91 f = tmp_path / "v3_ovf.aedat"
92 f.write_bytes(make_aedat3(t, x, y, p, ts_overflow=2))
93 with EventReader(f) as r:
94 out = r.read_all()
95 assert not isinstance(out, tuple)
96 assert np.array_equal(out["t"], t.astype(np.int64) + (2 << 31))
99def test_aedat4_uncompressed(tmp_path: Any) -> None:
100 t, x, y, p = random_events(500, 640, 480, seed=5)
101 t = t + 1_663_249_605_734_020 # DV timestamps are absolute epoch µs
102 f = tmp_path / "v4.aedat4"
103 f.write_bytes(make_aedat4(t, x, y, p))
104 with EventReader(f) as r:
105 check(r.read_all(), t, x, y, p)
106 assert r.shape() == (640, 480)
109def test_aedat4_no_stream_info(tmp_path: Any) -> None:
110 """Without a parseable infoNode the decoder falls back to identifying
111 event packets by their EVTS FlatBuffer identifier."""
112 t, x, y, p = random_events(100, 640, 480, seed=6)
113 f = tmp_path / "v4_noinfo.aedat4"
114 f.write_bytes(make_aedat4(t, x, y, p, info=b"not xml at all"))
115 with EventReader(f) as r:
116 check(r.read_all(), t, x, y, p)
119def test_aedat4_lz4(tmp_path: Any) -> None:
120 pytest.importorskip("lz4")
121 t, x, y, p = random_events(500, 640, 480, seed=7)
122 f = tmp_path / "v4_lz4.aedat4"
123 f.write_bytes(make_aedat4(t, x, y, p, compression=1))
124 with EventReader(f) as r:
125 check(r.read_all(), t, x, y, p)
128def test_aedat_chunked_iteration_and_reset(tmp_path: Any) -> None:
129 t, x, y, p = random_events(1000, 640, 480, seed=8)
130 f = tmp_path / "v4_chunks.aedat4"
131 f.write_bytes(make_aedat4(t, x, y, p, events_per_packet=64))
132 with EventReader(f, n_events=100) as r:
133 chunks = [np.asarray(c) for c in r]
134 assert sum(len(c) for c in chunks) == 1000
135 assert all(len(c) <= 100 for c in chunks)
136 got = np.concatenate(chunks)
137 assert np.array_equal(got["t"], t)
138 r.reset()
139 check(r.read_all(), t, x, y, p)
142def test_aedat_magic_sniffing(tmp_path: Any) -> None:
143 """Version line is recognised even without a known file extension."""
144 t, x, y, p = random_events(50, 128, 128, seed=9)
145 f = tmp_path / "recording.bin_dump"
146 f.write_bytes(make_aedat1(t, x, y, p))
147 with EventReader(f) as r:
148 check(r.read_all(), t, x, y, p)
150def test_AEDAT_empty_file(tmp_path: Any) -> None:
151 f = tmp_path / "empty.aedat"
152 f.touch()
153 with EventReader(f) as r:
154 out = r.read_all()
155 assert len(out) == 0
157def test_AEDAT_truncated_payload(tmp_path: Any) -> None:
158 f = tmp_path / "truncated.aedat"
159 f.write_bytes(b"\x01\x02\x03\x04") # 4 bytes
160 with EventReader(f) as r:
161 out = r.read_all()
162 assert len(out) == 0