Coverage for tests/io/test_roundtrip_matrix.py: 100%
119 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-15 09:33 +0000
1"""Cross-format round-trip matrix.
3Every read+write format goes through the same battery: bulk round-trip,
4chunked writing, windowed reading (``n_events`` and ``delta_t``), reset,
5iteration, empty files and timestamp normalization. Format-specific edge
6cases (timestamp wraps, container layouts, ...) live in the per-format test
7modules.
8"""
9import numpy as np
10import pytest
12from typing import Any
13from evutils.io import EventReader, EventWriter
14from evutils.types import Event_dtype
16@pytest.fixture(autouse=True)
17def skip_missing_deps(request: Any) -> None:
18 if "fmt" in request.fixturenames:
19 fmt = request.getfixturevalue("fmt")
21 if fmt == "hdf5":
22 pytest.importorskip("h5py")
23 pytest.importorskip("hdf5plugin")
25# name -> (suffix, writer kwargs, capabilities)
26FORMATS = {
27 "evt3": (".raw", {"format": "evt3"}, {}),
28 "evt21": (".raw", {"format": "evt21"}, {}),
29 "evt2": (".raw", {"format": "evt2"}, {}),
30 "dat": (".dat", {}, {}),
31 "aer": (".aer", {}, {"lossy_t": True, "coord_max": 512}),
32 "npz": (".npz", {}, {}),
33 "hdf5": (".h5", {}, {}),
34 "csv": (".csv", {}, {}),
35}
37PARAMS = sorted(FORMATS)
40def make_events(n: int=5000, coord_max: Any=None, seed: int=42, t_max: int=100_000) -> Any:
41 rng = np.random.default_rng(seed)
42 ev = np.zeros(n, dtype=Event_dtype)
43 ev["t"] = np.sort(rng.integers(0, t_max, n))
44 ev["x"] = rng.integers(0, coord_max or 1280, n)
45 ev["y"] = rng.integers(0, coord_max or 720, n)
46 ev["p"] = rng.integers(0, 2, n)
47 return ev
50def expected(ev: Any, caps: Any) -> Any:
51 """The array a lossless read should return (AER drops timestamps)."""
52 if caps.get("lossy_t"):
53 ev = ev.copy()
54 ev["t"] = 0
55 return ev
58def write_file(tmp_path: Any, fmt: str, ev: Any, n_chunks: int=1) -> Any:
59 suffix, kwargs, _ = FORMATS[fmt]
60 p = tmp_path / f"events_{fmt}{suffix}"
61 with EventWriter(p, **kwargs) as w: # type: ignore
62 for part in np.array_split(ev, n_chunks):
63 w.write(part)
64 return p
67def assert_events_equal(out: Any, ref: Any, context: str="") -> None:
68 out = np.asarray(out)
69 assert len(out) == len(ref), f"{context}: length {len(out)} != {len(ref)}"
70 for f in ("t", "x", "y", "p"):
71 assert np.array_equal(out[f], ref[f]), f"{context}: field {f!r} differs"
74@pytest.mark.parametrize("fmt", PARAMS)
75def test_bulk_roundtrip(tmp_path: Any, fmt: str) -> None:
76 caps = FORMATS[fmt][2]
77 ev = make_events(coord_max=caps.get("coord_max"))
78 p = write_file(tmp_path, fmt, ev)
79 with EventReader(p) as r:
80 assert_events_equal(r.read_all(), expected(ev, caps), fmt)
83@pytest.mark.parametrize("fmt", PARAMS)
84def test_chunked_write_roundtrip(tmp_path: Any, fmt: str) -> None:
85 """Writing in many small chunks must be byte-equivalent to one bulk write."""
86 caps = FORMATS[fmt][2]
87 ev = make_events(coord_max=caps.get("coord_max"))
88 p = write_file(tmp_path, fmt, ev, n_chunks=13)
89 with EventReader(p) as r:
90 assert_events_equal(r.read_all(), expected(ev, caps), fmt)
93@pytest.mark.parametrize("fmt", PARAMS)
94def test_windowed_n_events(tmp_path: Any, fmt: str) -> None:
95 caps = FORMATS[fmt][2]
96 ev = make_events(coord_max=caps.get("coord_max"))
97 p = write_file(tmp_path, fmt, ev)
98 with EventReader(p, n_events=137) as r:
99 chunks = [np.asarray(c) for c in r]
100 assert all(len(c) <= 137 for c in chunks)
101 assert_events_equal(np.concatenate(chunks), expected(ev, caps), fmt)
104@pytest.mark.parametrize("fmt", PARAMS)
105def test_windowed_delta_t(tmp_path: Any, fmt: str) -> None:
106 caps = FORMATS[fmt][2]
107 if caps.get("lossy_t"):
108 pytest.skip("format has no timestamps")
109 ev = make_events()
110 p = write_file(tmp_path, fmt, ev)
111 with EventReader(p, delta_t=1000) as r:
112 chunks = [np.asarray(c) for c in r]
113 for c in chunks:
114 if len(c) > 1:
115 assert int(c["t"].max()) - int(c["t"].min()) <= 1000
116 assert_events_equal(np.concatenate(chunks), ev, fmt)
119@pytest.mark.parametrize("fmt", PARAMS)
120def test_reset(tmp_path: Any, fmt: str) -> None:
121 caps = FORMATS[fmt][2]
122 ev = make_events(coord_max=caps.get("coord_max"))
123 p = write_file(tmp_path, fmt, ev)
124 with EventReader(p) as r:
125 first = np.asarray(r.read_all()).copy()
126 r.reset()
127 second = np.asarray(r.read_all())
128 assert_events_equal(second, first, fmt)
131@pytest.mark.parametrize("fmt", PARAMS)
132def test_empty_write_then_read(tmp_path: Any, fmt: str) -> None:
133 """A writer that never receives events must still produce a readable,
134 zero-event file."""
135 suffix, kwargs, _ = FORMATS[fmt]
136 p = tmp_path / f"empty{suffix}"
137 with EventWriter(p, **kwargs) as w: # type: ignore
138 w.write(np.zeros(0, dtype=Event_dtype))
139 with EventReader(p) as r:
140 out = r.read_all()
141 assert not isinstance(out, tuple)
142 assert len(out) == 0
143 # The empty columns keep their canonical dtypes.
144 assert out["t"].dtype == np.int64
145 assert out["x"].dtype == np.uint16
148@pytest.mark.parametrize("fmt", PARAMS)
149def test_read_after_eof_is_empty(tmp_path: Any, fmt: str) -> None:
150 caps = FORMATS[fmt][2]
151 ev = make_events(n=100, coord_max=caps.get("coord_max"))
152 p = write_file(tmp_path, fmt, ev)
153 with EventReader(p) as r:
154 r.read_all()
155 assert r.is_eof()
156 assert len(r.read_all()) == 0
159@pytest.mark.parametrize("fmt", PARAMS)
160def test_normalize_ts(tmp_path: Any, fmt: str) -> None:
161 caps = FORMATS[fmt][2]
162 if caps.get("lossy_t"):
163 pytest.skip("format has no timestamps")
164 ev = make_events()
165 ev["t"] += 5000 # ensure a non-zero first timestamp
166 p = write_file(tmp_path, fmt, ev)
167 with EventReader(p, normalize_ts=True) as r:
168 out = r.read_all()
169 assert not isinstance(out, tuple)
170 assert int(out["t"][0]) == 0
171 assert np.array_equal(np.diff(out["t"]), np.diff(ev["t"].astype(np.int64)))