Coverage for src/evutils/io/decoders.py: 85%
78 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-15 09:33 +0000
1"""Decoders module.
3Provides mapping and resolution of event decoders based on file extensions or
4magic bytes. Backends whose optional dependencies are missing (e.g. pandas for
5CSV, h5py for HDF5) are not registered; asking for them raises an
6``ImportError`` that names the extra to install.
7"""
9from pathlib import Path
10from typing import Type, Any, cast
12from .common import EventDecoder
14#: Extension -> decoder class, for available backends only.
15_READER_MAPPING: dict[str, Type[EventDecoder]] = {}
17#: Extension -> reason it is unavailable (missing optional dependency).
18_UNAVAILABLE: dict[str, str] = {}
21from ._aedat import EventDecoder_Aedat
22_READER_MAPPING[".aedat"] = EventDecoder_Aedat
23_READER_MAPPING[".aedat4"] = EventDecoder_Aedat
25from ._bin import EventDecoder_Bin
26_READER_MAPPING[".bin"] = EventDecoder_Bin
28try:
29 from ._csv import EventDecoder_Csv
30 _READER_MAPPING[".csv"] = EventDecoder_Csv
31 _READER_MAPPING[".txt"] = EventDecoder_Csv
32except ImportError:
33 _UNAVAILABLE[".csv"] = _UNAVAILABLE[".txt"] = (
34 "reading CSV/TXT event files requires the evutils native library "
35 "(build it with `uv pip install -e .`)"
36 )
38from ._dat import EventDecoder_Dat
39_READER_MAPPING[".dat"] = EventDecoder_Dat
41try:
42 from ._hdf5 import EventDecoder_HDF5
43 _READER_MAPPING[".hdf5"] = EventDecoder_HDF5
44 _READER_MAPPING[".h5"] = EventDecoder_HDF5
45except ImportError:
46 _UNAVAILABLE[".hdf5"] = _UNAVAILABLE[".h5"] = (
47 "reading HDF5 event files requires h5py/hdf5plugin: install `evutils[hdf5]`"
48 )
50from ._npz import EventDecoder_Npz
51_READER_MAPPING[".npz"] = EventDecoder_Npz
53from ._evt import EventDecoder_EVT
54_READER_MAPPING[".raw"] = EventDecoder_EVT
55_READER_MAPPING[".evt"] = EventDecoder_EVT
56_READER_MAPPING[".evt3"] = EventDecoder_EVT
57_READER_MAPPING[".evt2"] = EventDecoder_EVT
58_READER_MAPPING[".evt21"] = EventDecoder_EVT
59_READER_MAPPING[".evt4"] = EventDecoder_EVT
61from ._aer import EventDecoder_AER
62_READER_MAPPING[".aer"] = EventDecoder_AER
65def _lookup(ext: str) -> Type[EventDecoder]:
66 """Resolve an extension to a decoder class, or raise a helpful error."""
67 if ext in _READER_MAPPING:
68 return _READER_MAPPING[ext]
69 if ext in _UNAVAILABLE:
70 raise ImportError(f"File extension {ext} is supported, but {_UNAVAILABLE[ext]}")
71 raise ValueError(
72 f"File extension {ext} not supported, available extensions: "
73 f"{sorted(_READER_MAPPING.keys() | _UNAVAILABLE.keys())}"
74 )
77def get_reader_from_filename(file: Path) -> Type[EventDecoder]:
78 """Get the appropriate reader for the given file.
80 Parameters
81 ----------
82 file
83 File to read
85 Returns
86 -------
87 EventDecoder
88 Reader object for the file
90 """
91 return _lookup(file.suffix.lower())
94# Content sniffers: (predicate over the first bytes -> decoder class). Tried in
95# order when the filename extension is unknown or absent (streams, USB).
96def _header_lines(head: bytes) -> list[str]:
97 """The ``"% ..."`` ASCII header lines at the start of ``head``."""
98 text = head.decode("ascii", "ignore")
99 return [ln for ln in text.split("\n") if ln.startswith("% ")]
102def _sniff_dat(head: bytes) -> bool:
103 """Prophesee DAT: header carries ``% Version`` / ``% Data file containing``.
105 DAT and RAW/EVT both open with a ``%`` header, so they can only be told
106 apart by the keywords inside it.
107 """
108 for ln in _header_lines(head):
109 low = ln.lower()
110 if "data file containing" in low or low.startswith("% version"):
111 return True
112 return False
115def _sniff_evt(head: bytes) -> bool:
116 """Prophesee RAW/EVT: header carries ``% evt`` / ``% format EVT`` / ``% geometry``."""
117 for ln in _header_lines(head):
118 low = ln.lower()
119 if (low.startswith("% evt")
120 or low.startswith("% format evt")
121 or low.startswith("% geometry")):
122 return True
123 return False
126def _sniff_prophesee(head: bytes) -> bool:
127 """Fallback: any other ``"% "``-headed stream is treated as RAW/EVT."""
128 return head[:2] == b"% "
131def _sniff_aedat(head: bytes) -> bool:
132 """Check if the first bytes match an AEDAT version line.
134 Parameters
135 ----------
136 head : bytes
137 The first bytes of the file/stream.
139 Returns
140 -------
141 bool
142 True if it matches the AEDAT format, False otherwise.
144 """
145 return head.startswith(b"#!AER-DAT")
148_SNIFFERS = [
149 (_sniff_dat, "EventDecoder_Dat"),
150 (_sniff_evt, "EventDecoder_EVT"),
151 (_sniff_aedat, "EventDecoder_Aedat"),
152 (_sniff_prophesee, "EventDecoder_EVT"), # generic "% "-headed fallback
153]
156def resolve_decoder_cls(source: Any) -> Type[EventDecoder]:
157 """Determine the decoder class for a :class:`ByteSource`.
159 Tries the filename extension first (cheap, usually right), then falls back
160 to sniffing the leading bytes -- which works for streams and USB devices
161 that have no filename.
163 Parameters
164 ----------
165 source
166 A ByteSource (see :mod:`evutils.io._source`).
168 Returns
169 -------
170 Type[EventDecoder]
171 The decoder class to instantiate with the source.
173 """
174 name = getattr(source, "name", None)
175 if name:
176 ext = Path(name).suffix.lower()
177 if ext in _READER_MAPPING or ext in _UNAVAILABLE:
178 return _lookup(ext)
180 try:
181 head = source.peek(512)
182 except Exception:
183 head = b""
185 for matches, cls_name in _SNIFFERS:
186 if matches(head):
187 return cast(Type[EventDecoder], globals()[cls_name])
189 raise ValueError(
190 "Could not determine the event format: unknown/absent extension "
191 f"({name!r}) and no known magic bytes. Pass an explicit decoder."
192 )
195__all__ = ["EventDecoder", 'EventDecoder_Aedat', 'EventDecoder_Bin', 'EventDecoder_Csv', 'EventDecoder_Dat', 'EventDecoder_HDF5', 'EventDecoder_Npz', 'EventDecoder_EVT', 'EventDecoder_AER', 'get_reader_from_filename', 'resolve_decoder_cls']