Coverage for tests/io/test_csv.py: 100%
232 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-18 05:24 +0000
1import warnings
3import numpy as np
4import pytest
11####################################
12####################################
13# Test CSV module
14####################################
15####################################
21from typing import Any
22def test_CSV_writer(tmp_path: Any, test_events: Any) -> None:
23 from evutils.io import EventWriter
24 from evutils.io import EventReader
27 d = tmp_path / "sub"
28 d.mkdir()
29 p = d / "test.csv"
30 writer = EventWriter(p)
31 writer.write(test_events)
32 writer.close()
34 # Check if the file is created
35 assert p.is_file()
37 # load last line:
38 with open(p, 'r') as f:
39 lines = f.readlines()
40 last_line = lines[-1]
41 assert last_line == f"{test_events[-1]['t']},{test_events[-1]['x']},{test_events[-1]['y']},{test_events[-1]['p']}\n"
43 assert len(lines) == len(test_events) + 1
46 reader = EventReader(p, n_events=len(test_events))
47 events = reader.read()
48 assert np.array_equal(events, test_events)
50def test_CSV_reader_nevents(event_files: Any, test_events: Any) -> None:
51 csv_file_path = event_files['csv']
53 from evutils.io import EventReader
55 STEP=100
57 with EventReader(csv_file_path, n_events=STEP) as reader:
59 for i in range(0, len(test_events), STEP):
60 events = reader.read()
61 assert np.array_equal(events, test_events[i:i+STEP])
64def test_CSV_reader_gen(event_files: Any, test_events: Any) -> None:
65 csv_file_path = event_files['csv']
67 from evutils.io import EventReader
69 STEP=100
70 i = 0
72 for ev in EventReader(csv_file_path, n_events=STEP):
74 assert np.array_equal(ev, test_events[i:i+STEP])
75 i += STEP
78####################################
79# Header / column-order handling
80####################################
82def _small_events(n: int = 50) -> Any:
83 from evutils.types import Event_dtype
84 ev = np.zeros(n, dtype=Event_dtype)
85 ev['t'] = np.arange(n, dtype=np.int64) * 10
86 ev['x'] = np.arange(n) % 1280
87 ev['y'] = np.arange(n) % 720
88 ev['p'] = np.arange(n) % 2
89 return ev
92def test_CSV_headerless_default_order(tmp_path: Any) -> None:
93 """No header, no order parameter: the default [t, x, y, p] is assumed."""
94 from evutils.io import EventReader, EventWriter
95 ev = _small_events()
96 p = tmp_path / "noheader.csv"
97 with EventWriter(p, header=False) as w:
98 w.write(ev)
99 assert not open(p).readline().startswith("t") # really headerless
101 with EventReader(p, mode="all") as r:
102 assert np.array_equal(np.asarray(r.read()), ev)
105def test_CSV_headerless_custom_order(tmp_path: Any) -> None:
106 """Write and read back with a shuffled column order (no header)."""
107 from evutils.io import EventReader, EventWriter
108 ev = _small_events()
109 order = ['x', 'y', 'p', 't']
110 p = tmp_path / "xypt.csv"
111 with EventWriter(p, header=False, order=order) as w:
112 w.write(ev)
114 with EventReader(p, mode="all", order=order) as r:
115 assert np.array_equal(np.asarray(r.read()), ev)
118def test_CSV_header_with_custom_order(tmp_path: Any) -> None:
119 """A shuffled header alone must drive the column mapping on read."""
120 from evutils.io import EventReader, EventWriter
121 ev = _small_events()
122 p = tmp_path / "hdr_xypt.csv"
123 with EventWriter(p, header=True, order=['x', 'y', 'p', 't']) as w:
124 w.write(ev)
125 assert open(p).readline().strip() == "x,y,p,t"
127 with EventReader(p, mode="all") as r: # order inferred from header
128 assert np.array_equal(np.asarray(r.read()), ev)
131def test_CSV_header_overrides_order_param_with_warning(tmp_path: Any) -> None:
132 """When the file header disagrees with the order parameter, the header
133 wins and a warning is emitted."""
134 import warnings
135 from evutils.io import EventReader, EventWriter
136 ev = _small_events()
137 p = tmp_path / "conflict.csv"
138 with EventWriter(p, header=True) as w: # header: t,x,y,p
139 w.write(ev)
141 with warnings.catch_warnings(record=True) as caught:
142 warnings.simplefilter("always")
143 with EventReader(p, mode="all", order=['x', 'y', 'p', 't']) as r:
144 out = r.read()
145 assert any("precedence" in str(w.message) for w in caught)
146 assert np.array_equal(np.asarray(out), ev)
149def test_CSV_invalid_order_rejected(tmp_path: Any) -> None:
150 from evutils.io import EventReader
151 p = tmp_path / "dummy.csv"
152 p.write_text("t,x,y,p\n0,1,2,1\n")
153 with pytest.raises(ValueError):
154 EventReader(p, order=['t', 'x', 'y']) # not 4 entries
155 with pytest.raises(ValueError):
156 EventReader(p, order=['t', 'x', 'y', 'z']) # missing 'p'
159####################################
160# Delimiters
161####################################
163@pytest.mark.parametrize("delim", [";", "\t", " "])
164def test_CSV_delimiter_roundtrip(tmp_path: Any, delim: Any) -> None:
165 from evutils.io import EventReader, EventWriter
166 ev = _small_events()
167 p = tmp_path / "delim.csv"
168 with EventWriter(p, sep=delim) as w:
169 w.write(ev)
170 assert delim in open(p).readline()
172 with EventReader(p, mode="all", delimiter=delim) as r:
173 assert np.array_equal(np.asarray(r.read()), ev)
176####################################
177# Malformed / edge-case inputs
178####################################
180def test_CSV_empty_file(tmp_path: Any) -> None:
181 from evutils.io import EventReader
182 p = tmp_path / "empty.csv"
183 p.touch()
184 with EventReader(p, mode="all") as r:
185 assert len(r.read()) == 0
188def test_CSV_header_only_file(tmp_path: Any) -> None:
189 from evutils.io import EventReader
190 p = tmp_path / "hdr_only.csv"
191 p.write_text("t,x,y,p\n")
192 with EventReader(p, mode="all") as r:
193 assert len(r.read()) == 0
196def test_CSV_missing_trailing_newline(tmp_path: Any) -> None:
197 """The final line must be parsed even without a terminating newline."""
198 from evutils.io import EventReader
199 p = tmp_path / "notrail.csv"
200 p.write_text("t,x,y,p\n10,1,2,1\n20,3,4,0") # no final \n
201 with EventReader(p, mode="all") as r:
202 out = r.read()
203 assert len(out) == 2
204 assert out['t'][-1] == 20 and out['x'][-1] == 3 and out['p'][-1] == 0
207def test_CSV_blank_lines_skipped(tmp_path: Any) -> None:
208 from evutils.io import EventReader
209 p = tmp_path / "blanks.csv"
210 p.write_text("t,x,y,p\n10,1,2,1\n\n\n20,3,4,0\n\n")
211 with EventReader(p, mode="all") as r:
212 out = r.read()
213 assert np.array_equal(out['t'], [10, 20])
216def test_CSV_crlf_line_endings(tmp_path: Any) -> None:
217 from evutils.io import EventReader
218 p = tmp_path / "crlf.csv"
219 p.write_bytes(b"t,x,y,p\r\n10,1,2,1\r\n20,3,4,0\r\n")
220 with EventReader(p, mode="all") as r:
221 out = r.read()
222 assert np.array_equal(out['t'], [10, 20])
223 assert np.array_equal(out['p'], [1, 0])
226def test_CSV_non_numeric_field_parses_as_zero(tmp_path: Any) -> None:
227 """Documented C-parser semantics: garbage fields decode to 0 (no crash,
228 no exception, no line skip)."""
229 from evutils.io import EventReader
230 p = tmp_path / "garbage.csv"
231 p.write_text("t,x,y,p\n10,abc,2,1\n20,3,4,0\n")
232 with EventReader(p, mode="all") as r:
233 out = r.read()
234 assert np.array_equal(out['t'], [10, 20])
235 assert out['x'][0] == 0 # 'abc' -> 0
236 assert out['x'][1] == 3
239def test_CSV_extra_columns_ignored(tmp_path: Any) -> None:
240 from evutils.io import EventReader
241 p = tmp_path / "extra.csv"
242 p.write_text("t,x,y,p\n10,1,2,1,999,777\n20,3,4,0,888\n")
243 with EventReader(p, mode="all") as r:
244 out = r.read()
245 assert np.array_equal(out['t'], [10, 20])
246 assert np.array_equal(out['x'], [1, 3])
249def test_CSV_short_line_does_not_crash(tmp_path: Any) -> None:
250 """A line with missing fields must not crash or derail later lines.
252 (Values for the absent fields are unspecified; only stability and the
253 surrounding rows are asserted.)"""
254 from evutils.io import EventReader
255 p = tmp_path / "short.csv"
256 p.write_text("t,x,y,p\n10,1,2,1\n20,3\n30,5,6,0\n")
257 with EventReader(p, mode="all") as r:
258 out = r.read()
259 assert len(out) == 3
260 assert out['t'][0] == 10 and out['t'][2] == 30
261 assert out['x'][2] == 5 and out['p'][2] == 0
264def test_CSV_short_line_warns(tmp_path: Any) -> None:
265 """A short row (missing a mapped column) surfaces a malformed-row warning,
266 matching the binary parsers' malformed-packet warning."""
267 from evutils.io import EventReader
268 p = tmp_path / "short_warn.csv"
269 p.write_text("t,x,y,p\n10,1,2,1\n20,3\n30,5,6,0\n")
270 with warnings.catch_warnings(record=True) as caught:
271 warnings.simplefilter("always")
272 with EventReader(p, mode="all") as r:
273 out = r.read()
274 assert len(out) == 3
275 msgs = [str(w.message) for w in caught]
276 assert any("malformed" in m.lower() and "CSV row" in m for m in msgs), msgs
279def test_CSV_short_line_strict_raises(tmp_path: Any) -> None:
280 """strict=True turns a malformed CSV row into an error, like the binary
281 formats."""
282 from evutils.io import EventReader
283 p = tmp_path / "short_strict.csv"
284 p.write_text("t,x,y,p\n10,1,2,1\n20,3\n30,5,6,0\n")
285 with pytest.raises(RuntimeError, match="malformed"):
286 with EventReader(p, mode="all", strict=True) as r:
287 r.read()
290def test_CSV_wellformed_no_warning(tmp_path: Any) -> None:
291 """Clean rows (incl. extra trailing columns) emit no malformed warning."""
292 from evutils.io import EventReader
293 p = tmp_path / "clean.csv"
294 p.write_text("t,x,y,p\n10,1,2,1,999\n20,3,4,0\n")
295 with warnings.catch_warnings(record=True) as caught:
296 warnings.simplefilter("always")
297 with EventReader(p, mode="all") as r:
298 r.read()
299 assert not any("malformed" in str(w.message).lower() for w in caught)
302def test_CSV_negative_and_extreme_values(tmp_path: Any) -> None:
303 """Signed 64-bit timestamps and full uint16 coordinates round-trip."""
304 from evutils.io import EventReader, EventWriter
305 from evutils.types import Event_dtype
306 ev = np.zeros(4, dtype=Event_dtype)
307 ev['t'] = [-1_000_000, 0, 2**62, 2**62 + 1]
308 ev['x'] = [0, 65535, 1, 2]
309 ev['y'] = [65535, 0, 3, 4]
310 ev['p'] = [0, 1, 1, 0]
311 p = tmp_path / "extreme.csv"
312 with EventWriter(p) as w:
313 w.write(ev)
314 with EventReader(p, mode="all") as r:
315 assert np.array_equal(np.asarray(r.read()), ev)
318def test_CSV_whitespace_padding(tmp_path: Any) -> None:
319 """Fields padded with spaces/tabs still parse."""
320 from evutils.io import EventReader
321 p = tmp_path / "padded.csv"
322 p.write_text("t,x,y,p\n 10 , 1 ,2, 1\n\t20,3,\t4,0\n")
323 with EventReader(p, mode="all") as r:
324 out = r.read()
325 assert np.array_equal(out['t'], [10, 20])
326 assert np.array_equal(out['x'], [1, 3])
327 assert np.array_equal(out['p'], [1, 0])
330def test_CSV_chunked_reads_across_refills(tmp_path: Any) -> None:
331 """Events spanning multiple internal 4 MB refills stay contiguous and
332 ordered (exercises the buffered consume/refill loop)."""
333 from evutils.io import EventReader, EventWriter
334 from evutils.types import Event_dtype
335 n = 200_000 # ~2.5 MB of text; chunk_size below forces many parse calls
336 ev = np.zeros(n, dtype=Event_dtype)
337 ev['t'] = np.arange(n, dtype=np.int64)
338 ev['x'] = np.arange(n) % 1280
339 ev['y'] = np.arange(n) % 720
340 ev['p'] = np.arange(n) % 2
341 p = tmp_path / "big.csv"
342 with EventWriter(p) as w:
343 w.write(ev)
345 total = 0
346 with EventReader(p, n_events=7_777) as r:
347 for chunk in r:
348 assert np.array_equal(chunk['t'], ev['t'][total:total + len(chunk)])
349 total += len(chunk)
350 assert total == n