Coverage for tests/io/test_csv.py: 100%
204 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-15 09:33 +0000
1import numpy as np
2import pytest
9####################################
10####################################
11# Test CSV module
12####################################
13####################################
19from typing import Any
20def test_CSV_writer(tmp_path: Any, test_events: Any) -> None:
21 from evutils.io import EventWriter
22 from evutils.io import EventReader
25 d = tmp_path / "sub"
26 d.mkdir()
27 p = d / "test.csv"
28 writer = EventWriter(p)
29 writer.write(test_events)
30 writer.close()
32 # Check if the file is created
33 assert p.is_file()
35 # load last line:
36 with open(p, 'r') as f:
37 lines = f.readlines()
38 last_line = lines[-1]
39 assert last_line == f"{test_events[-1]['t']},{test_events[-1]['x']},{test_events[-1]['y']},{test_events[-1]['p']}\n"
41 assert len(lines) == len(test_events) + 1
44 reader = EventReader(p, n_events=len(test_events))
45 events = reader.read()
46 assert np.array_equal(events, test_events)
48def test_CSV_reader_nevents(event_files: Any, test_events: Any) -> None:
49 csv_file_path = event_files['csv']
51 from evutils.io import EventReader
53 STEP=100
55 with EventReader(csv_file_path, n_events=STEP) as reader:
57 for i in range(0, len(test_events), STEP):
58 events = reader.read()
59 assert np.array_equal(events, test_events[i:i+STEP])
62def test_CSV_reader_gen(event_files: Any, test_events: Any) -> None:
63 csv_file_path = event_files['csv']
65 from evutils.io import EventReader
67 STEP=100
68 i = 0
70 for ev in EventReader(csv_file_path, n_events=STEP):
72 assert np.array_equal(ev, test_events[i:i+STEP])
73 i += STEP
76####################################
77# Header / column-order handling
78####################################
80def _small_events(n: int = 50) -> Any:
81 from evutils.types import Event_dtype
82 ev = np.zeros(n, dtype=Event_dtype)
83 ev['t'] = np.arange(n, dtype=np.int64) * 10
84 ev['x'] = np.arange(n) % 1280
85 ev['y'] = np.arange(n) % 720
86 ev['p'] = np.arange(n) % 2
87 return ev
90def test_CSV_headerless_default_order(tmp_path: Any) -> None:
91 """No header, no order parameter: the default [t, x, y, p] is assumed."""
92 from evutils.io import EventReader, EventWriter
93 ev = _small_events()
94 p = tmp_path / "noheader.csv"
95 with EventWriter(p, header=False) as w:
96 w.write(ev)
97 assert not open(p).readline().startswith("t") # really headerless
99 with EventReader(p, mode="all") as r:
100 assert np.array_equal(np.asarray(r.read()), ev)
103def test_CSV_headerless_custom_order(tmp_path: Any) -> None:
104 """Write and read back with a shuffled column order (no header)."""
105 from evutils.io import EventReader, EventWriter
106 ev = _small_events()
107 order = ['x', 'y', 'p', 't']
108 p = tmp_path / "xypt.csv"
109 with EventWriter(p, header=False, order=order) as w:
110 w.write(ev)
112 with EventReader(p, mode="all", order=order) as r:
113 assert np.array_equal(np.asarray(r.read()), ev)
116def test_CSV_header_with_custom_order(tmp_path: Any) -> None:
117 """A shuffled header alone must drive the column mapping on read."""
118 from evutils.io import EventReader, EventWriter
119 ev = _small_events()
120 p = tmp_path / "hdr_xypt.csv"
121 with EventWriter(p, header=True, order=['x', 'y', 'p', 't']) as w:
122 w.write(ev)
123 assert open(p).readline().strip() == "x,y,p,t"
125 with EventReader(p, mode="all") as r: # order inferred from header
126 assert np.array_equal(np.asarray(r.read()), ev)
129def test_CSV_header_overrides_order_param_with_warning(tmp_path: Any) -> None:
130 """When the file header disagrees with the order parameter, the header
131 wins and a warning is emitted."""
132 import warnings
133 from evutils.io import EventReader, EventWriter
134 ev = _small_events()
135 p = tmp_path / "conflict.csv"
136 with EventWriter(p, header=True) as w: # header: t,x,y,p
137 w.write(ev)
139 with warnings.catch_warnings(record=True) as caught:
140 warnings.simplefilter("always")
141 with EventReader(p, mode="all", order=['x', 'y', 'p', 't']) as r:
142 out = r.read()
143 assert any("precedence" in str(w.message) for w in caught)
144 assert np.array_equal(np.asarray(out), ev)
147def test_CSV_invalid_order_rejected(tmp_path: Any) -> None:
148 from evutils.io import EventReader
149 p = tmp_path / "dummy.csv"
150 p.write_text("t,x,y,p\n0,1,2,1\n")
151 with pytest.raises(ValueError):
152 EventReader(p, order=['t', 'x', 'y']) # not 4 entries
153 with pytest.raises(ValueError):
154 EventReader(p, order=['t', 'x', 'y', 'z']) # missing 'p'
157####################################
158# Delimiters
159####################################
161@pytest.mark.parametrize("delim", [";", "\t", " "])
162def test_CSV_delimiter_roundtrip(tmp_path: Any, delim: Any) -> None:
163 from evutils.io import EventReader, EventWriter
164 ev = _small_events()
165 p = tmp_path / "delim.csv"
166 with EventWriter(p, sep=delim) as w:
167 w.write(ev)
168 assert delim in open(p).readline()
170 with EventReader(p, mode="all", delimiter=delim) as r:
171 assert np.array_equal(np.asarray(r.read()), ev)
174####################################
175# Malformed / edge-case inputs
176####################################
178def test_CSV_empty_file(tmp_path: Any) -> None:
179 from evutils.io import EventReader
180 p = tmp_path / "empty.csv"
181 p.touch()
182 with EventReader(p, mode="all") as r:
183 assert len(r.read()) == 0
186def test_CSV_header_only_file(tmp_path: Any) -> None:
187 from evutils.io import EventReader
188 p = tmp_path / "hdr_only.csv"
189 p.write_text("t,x,y,p\n")
190 with EventReader(p, mode="all") as r:
191 assert len(r.read()) == 0
194def test_CSV_missing_trailing_newline(tmp_path: Any) -> None:
195 """The final line must be parsed even without a terminating newline."""
196 from evutils.io import EventReader
197 p = tmp_path / "notrail.csv"
198 p.write_text("t,x,y,p\n10,1,2,1\n20,3,4,0") # no final \n
199 with EventReader(p, mode="all") as r:
200 out = r.read()
201 assert len(out) == 2
202 assert out['t'][-1] == 20 and out['x'][-1] == 3 and out['p'][-1] == 0
205def test_CSV_blank_lines_skipped(tmp_path: Any) -> None:
206 from evutils.io import EventReader
207 p = tmp_path / "blanks.csv"
208 p.write_text("t,x,y,p\n10,1,2,1\n\n\n20,3,4,0\n\n")
209 with EventReader(p, mode="all") as r:
210 out = r.read()
211 assert np.array_equal(out['t'], [10, 20])
214def test_CSV_crlf_line_endings(tmp_path: Any) -> None:
215 from evutils.io import EventReader
216 p = tmp_path / "crlf.csv"
217 p.write_bytes(b"t,x,y,p\r\n10,1,2,1\r\n20,3,4,0\r\n")
218 with EventReader(p, mode="all") as r:
219 out = r.read()
220 assert np.array_equal(out['t'], [10, 20])
221 assert np.array_equal(out['p'], [1, 0])
224def test_CSV_non_numeric_field_parses_as_zero(tmp_path: Any) -> None:
225 """Documented C-parser semantics: garbage fields decode to 0 (no crash,
226 no exception, no line skip)."""
227 from evutils.io import EventReader
228 p = tmp_path / "garbage.csv"
229 p.write_text("t,x,y,p\n10,abc,2,1\n20,3,4,0\n")
230 with EventReader(p, mode="all") as r:
231 out = r.read()
232 assert np.array_equal(out['t'], [10, 20])
233 assert out['x'][0] == 0 # 'abc' -> 0
234 assert out['x'][1] == 3
237def test_CSV_extra_columns_ignored(tmp_path: Any) -> None:
238 from evutils.io import EventReader
239 p = tmp_path / "extra.csv"
240 p.write_text("t,x,y,p\n10,1,2,1,999,777\n20,3,4,0,888\n")
241 with EventReader(p, mode="all") as r:
242 out = r.read()
243 assert np.array_equal(out['t'], [10, 20])
244 assert np.array_equal(out['x'], [1, 3])
247def test_CSV_short_line_does_not_crash(tmp_path: Any) -> None:
248 """A line with missing fields must not crash or derail later lines.
250 (Values for the absent fields are unspecified; only stability and the
251 surrounding rows are asserted.)"""
252 from evutils.io import EventReader
253 p = tmp_path / "short.csv"
254 p.write_text("t,x,y,p\n10,1,2,1\n20,3\n30,5,6,0\n")
255 with EventReader(p, mode="all") as r:
256 out = r.read()
257 assert len(out) == 3
258 assert out['t'][0] == 10 and out['t'][2] == 30
259 assert out['x'][2] == 5 and out['p'][2] == 0
262def test_CSV_negative_and_extreme_values(tmp_path: Any) -> None:
263 """Signed 64-bit timestamps and full uint16 coordinates round-trip."""
264 from evutils.io import EventReader, EventWriter
265 from evutils.types import Event_dtype
266 ev = np.zeros(4, dtype=Event_dtype)
267 ev['t'] = [-1_000_000, 0, 2**62, 2**62 + 1]
268 ev['x'] = [0, 65535, 1, 2]
269 ev['y'] = [65535, 0, 3, 4]
270 ev['p'] = [0, 1, 1, 0]
271 p = tmp_path / "extreme.csv"
272 with EventWriter(p) as w:
273 w.write(ev)
274 with EventReader(p, mode="all") as r:
275 assert np.array_equal(np.asarray(r.read()), ev)
278def test_CSV_whitespace_padding(tmp_path: Any) -> None:
279 """Fields padded with spaces/tabs still parse."""
280 from evutils.io import EventReader
281 p = tmp_path / "padded.csv"
282 p.write_text("t,x,y,p\n 10 , 1 ,2, 1\n\t20,3,\t4,0\n")
283 with EventReader(p, mode="all") as r:
284 out = r.read()
285 assert np.array_equal(out['t'], [10, 20])
286 assert np.array_equal(out['x'], [1, 3])
287 assert np.array_equal(out['p'], [1, 0])
290def test_CSV_chunked_reads_across_refills(tmp_path: Any) -> None:
291 """Events spanning multiple internal 4 MB refills stay contiguous and
292 ordered (exercises the buffered consume/refill loop)."""
293 from evutils.io import EventReader, EventWriter
294 from evutils.types import Event_dtype
295 n = 200_000 # ~2.5 MB of text; chunk_size below forces many parse calls
296 ev = np.zeros(n, dtype=Event_dtype)
297 ev['t'] = np.arange(n, dtype=np.int64)
298 ev['x'] = np.arange(n) % 1280
299 ev['y'] = np.arange(n) % 720
300 ev['p'] = np.arange(n) % 2
301 p = tmp_path / "big.csv"
302 with EventWriter(p) as w:
303 w.write(ev)
305 total = 0
306 with EventReader(p, n_events=7_777) as r:
307 for chunk in r:
308 assert np.array_equal(chunk['t'], ev['t'][total:total + len(chunk)])
309 total += len(chunk)
310 assert total == n