Coverage for tests/io/test_npz.py: 100%

1"""NPZ decoder/encoder tests: numpy interoperability, compression, geometry 

2and malformed-archive errors. (Bulk/chunked/windowed round-trips are covered 

3by the cross-format matrix in test_roundtrip_matrix.py.)""" 

4import numpy as np 

5import pytest 

6 

7from evutils.io import EventReader, EventWriter 

8from evutils.types import Event_dtype 

9 

10from typing import Any 

11 

12def make_events(n: int=10_000, seed: int=11) -> Any: 

13 rng = np.random.default_rng(seed) 

14 ev = np.zeros(n, dtype=Event_dtype) 

15 ev["t"] = np.sort(rng.integers(0, 1_000_000, n)) 

16 ev["x"] = rng.integers(0, 1280, n) 

17 ev["y"] = rng.integers(0, 720, n) 

18 ev["p"] = rng.integers(0, 2, n) 

19 return ev 

20 

21 

22def test_np_load_reads_our_files(tmp_path: Any) -> None: 

23 """Files we write are plain npz archives np.load understands.""" 

24 ev = make_events() 

25 p = tmp_path / "ours.npz" 

26 with EventWriter(p, width=640, height=480) as w: 

27 w.write(ev) 

28 

29 d = np.load(p) 

30 assert sorted(d.files) == ["height", "p", "t", "width", "x", "y"] 

31 assert np.array_equal(d["t"], ev["t"]) 

32 assert np.array_equal(d["x"], ev["x"]) 

33 assert (int(d["width"]), int(d["height"])) == (640, 480) 

34 

35 

36def test_we_read_np_savez_soa(tmp_path: Any) -> None: 

37 ev = make_events() 

38 p = tmp_path / "soa.npz" 

39 np.savez(p, t=ev["t"], x=ev["x"], y=ev["y"], p=ev["p"]) 

40 with EventReader(p) as r: 

41 assert np.array_equal(np.asarray(r.read_all()), ev) 

42 

43 

44def test_we_read_np_savez_structured(tmp_path: Any) -> None: 

45 """A single structured 'events' member (compressed) is accepted too.""" 

46 ev = make_events() 

47 p = tmp_path / "aos.npz" 

48 np.savez_compressed(p, events=ev) 

49 with EventReader(p) as r: 

50 assert np.array_equal(np.asarray(r.read_all()), ev) 

51 

52 

53def test_compressed_write(tmp_path: Any) -> None: 

54 ev = make_events() 

55 plain = tmp_path / "plain.npz" 

56 packed = tmp_path / "packed.npz" 

57 with EventWriter(plain) as w: 

58 w.write(ev) 

59 with EventWriter(packed, compressed=True) as w: 

60 w.write(ev) 

61 assert packed.stat().st_size < plain.stat().st_size 

62 with EventReader(packed) as r: 

63 assert np.array_equal(np.asarray(r.read_all()), ev) 

64 

65 

66def test_geometry_roundtrip(tmp_path: Any) -> None: 

67 p = tmp_path / "geo.npz" 

68 with EventWriter(p, width=346, height=260) as w: 

69 w.write(make_events(10)) 

70 with EventReader(p) as r: 

71 r.read() 

72 assert r.shape() == (346, 260) 

73 

74 

75def test_missing_event_keys_errors(tmp_path: Any) -> None: 

76 p = tmp_path / "bogus.npz" 

77 np.savez(p, foo=np.arange(5)) 

78 with pytest.raises(ValueError, match="does not contain event data"): 

79 EventReader(p).read_all() 

80 

81 

82def test_mismatched_column_lengths_errors(tmp_path: Any) -> None: 

83 p = tmp_path / "ragged.npz" 

84 np.savez(p, t=np.arange(10), x=np.arange(9, dtype=np.uint16), 

85 y=np.zeros(10, np.uint16), p=np.zeros(10, np.uint8)) 

86 with pytest.raises(ValueError, match="mismatched lengths"): 

87 EventReader(p).read_all() 

88 

89 

90def test_large_timestamps(tmp_path: Any) -> None: 

91 ev = make_events(100) 

92 ev["t"] += 2**40 # far beyond 32 bits 

93 p = tmp_path / "big_t.npz" 

94 with EventWriter(p) as w: 

95 w.write(ev) 

96 with EventReader(p) as r: 

97 out = r.read_all() 

98 assert not isinstance(out, tuple) 

99 assert np.array_equal(out["t"], ev["t"]) 

100 

101def test_NPZ_empty_arrays(tmp_path: Any) -> None: 

102 p = tmp_path / "empty_arrays.npz" 

103 np.savez(p, t=np.zeros(0, dtype=np.int64), x=np.zeros(0, dtype=np.uint16), 

104 y=np.zeros(0, dtype=np.uint16), p=np.zeros(0, dtype=np.uint8)) 

105 with EventReader(p) as r: 

106 out = r.read_all() 

107 assert len(out) == 0 

108 

109def test_NPZ_empty_file(tmp_path: Any) -> None: 

110 import zipfile 

111 p = tmp_path / "empty_file.npz" 

112 p.touch() 

113 with pytest.raises(zipfile.BadZipFile): 

114 with EventReader(p) as r: 

115 r.read_all()