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

1"""Format-agnostic checks against the real downloaded recordings. 

2 

3Every file in the reference tarball ships a JSON sidecar (produced by 

4``scripts/generate_metadata.py``) carrying the ground-truth ``format``, 

5``resolution``, event/polarity/trigger counts, etc. This module discovers 

6*all* of them -- whatever the format -- and verifies the decoder output 

7against that ground truth. 

8 

9Adding a new format is zero-code: drop the recording plus its JSON sidecar 

10into the tarball and it is exercised here automatically. The per-file 

11parametrization (fixture ``real_event_file``) is generated in 

12``tests/conftest.py::pytest_generate_tests``. 

13""" 

14import numpy as np 

15import pytest 

16 

17from typing import Any 

18 

19# Recordings whose Metavision reference counts differ slightly from a 

20# byte-exact decode: Metavision filters a handful of events, while our decoder 

21# matches expelliarmus exactly. Allow a tiny relative tolerance on *event* 

22# counts for these (triggers are always exact). See tests/io/test_evt.py for 

23# the investigation behind each number. 

24_COUNT_TOLERANT_FILES = { 

25 'evt2_195_falling_particles.raw', # +14 

26 'evt3_flash.raw', # +3345 

27 'evt3_laser.raw', # +132 

28} 

29 

30 

31def test_real_file_matches_metadata(real_event_file: Any) -> None: 

32 """Decode a real recording end-to-end and check it against its JSON sidecar. 

33 

34 Verifies (only what the metadata provides, so it stays format-agnostic): 

35 sensor resolution, coordinate bounds, total/positive/negative event counts, 

36 external-trigger counts, and an overall upward timestamp trend. 

37 """ 

38 from evutils.io import EventReader 

39 from evutils.io.decoders import get_reader_from_filename 

40 

41 ef = real_event_file 

42 meta = ef.metadata 

43 

44 # Only ask for triggers from formats whose decoder can actually read them; 

45 # for the rest we instead assert the metadata isn't hiding any (below). 

46 supports_triggers = getattr( 

47 get_reader_from_filename(ef.path), "SUPPORTS_EXT_TRIGGERS", False 

48 ) 

49 

50 n = n_pos = n_neg = 0 

51 n_tr_pos = n_tr_neg = 0 

52 x_max = y_max = -1 

53 first_t = last_t = None 

54 

55 with EventReader(ef.path, ext_trigger=supports_triggers, chunk_size=10_000_000) as reader: 

56 shape = list(reader.shape()) 

57 for chunk in reader: 

58 ev, tr = chunk if isinstance(chunk, tuple) else (chunk, None) 

59 if len(ev): 

60 n += len(ev) 

61 n_pos += int(np.count_nonzero(ev["p"] == 1)) 

62 n_neg += int(np.count_nonzero(ev["p"] == 0)) 

63 x_max = max(x_max, int(ev["x"].max())) 

64 y_max = max(y_max, int(ev["y"].max())) 

65 if first_t is None: 

66 first_t = int(ev["t"][0]) 

67 last_t = int(ev["t"][-1]) 

68 if tr is not None and len(tr): 

69 n_tr_pos += int(np.count_nonzero(tr["p"] == 1)) 

70 n_tr_neg += int(np.count_nonzero(tr["p"] == 0)) 

71 

72 assert n > 0, f"{ef.path.name}: decoded no events" 

73 

74 def check(name: str, actual: int, expected: int, tol: int = 0) -> None: 

75 assert abs(actual - expected) <= tol, ( 

76 f"{ef.path.name}: {name} actual {actual} != expected {expected}" 

77 + (f" (tolerance {tol})" if tol else "") 

78 ) 

79 

80 # --- resolution --- 

81 width, height = meta["resolution"] 

82 if shape != [None, None]: 

83 assert shape == meta["resolution"], ( 

84 f"{ef.path.name}: shape {shape} != metadata {meta['resolution']}" 

85 ) 

86 assert x_max < width, f"{ef.path.name}: x max {x_max} >= width {width}" 

87 assert y_max < height, f"{ef.path.name}: y max {y_max} >= height {height}" 

88 

89 # --- event counts (exact, save for the known Metavision-filtered files) --- 

90 rtol = 1e-3 if ef.path.name in _COUNT_TOLERANT_FILES else 0.0 

91 check("total events", n, meta["count"], int(meta["count"] * rtol)) 

92 check("positive events", n_pos, meta["pos_count"], int(meta["pos_count"] * rtol)) 

93 check("negative events", n_neg, meta["neg_count"], int(meta["neg_count"] * rtol)) 

94 

95 # --- external triggers --- 

96 tr_meta = meta.get("external_triggers", {"total": 0, "positive": 0, "negative": 0}) 

97 if supports_triggers: 

98 check("total triggers", n_tr_pos + n_tr_neg, tr_meta["total"]) 

99 check("positive triggers", n_tr_pos, tr_meta["positive"]) 

100 check("negative triggers", n_tr_neg, tr_meta["negative"]) 

101 else: 

102 # A format we cannot read triggers from must not be hiding any. 

103 assert tr_meta["total"] == 0, ( 

104 f"{ef.path.name}: metadata declares {tr_meta['total']} triggers but " 

105 f"'{ef.path.suffix}' decoding cannot read them" 

106 ) 

107 

108 # --- timestamps trend upward overall (streams are not strictly monotonic) --- 

109 assert first_t is not None and first_t <= last_t, ( 

110 f"{ef.path.name}: timestamps not increasing overall ({first_t} -> {last_t})" 

111 )