Coverage for src/evutils/io/_index.py: 89%
257 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-18 05:24 +0000
1"""Seek index for timestamp / event-index random access.
3A :class:`SeekIndex` is a coarse, monotonic map from *absolute timestamp* and
4*cumulative event count* to a *word offset* into a decoder's payload. It lets a
5seekable decoder jump close to a target (the nearest bookmark at or before it)
6and then decode forward the small remainder to the exact position.
8Two ways to obtain one:
10* **Build it** from a fresh sequential decode pass (:func:`build_seek_index`),
11 recording one bookmark per parse step. The timestamps recorded this way are
12 the decoder's own absolute timestamps, so they already include any TIME_HIGH
13 wrap accumulation -- which is exactly what a mid-file jump + parser reset
14 would otherwise lose.
15* **Read an OpenEB / Metavision** ``<file>.raw.tmp_index`` sidecar
16 (:func:`read_metavision_index`), reusing an index another tool already built.
17 Metavision timestamps are shifted by ``ts_shift_us`` relative to the raw
18 stream; we add it back so bookmarks live in the same (raw-absolute) timeline
19 the evutils EVT decoder produces.
21The wrap accumulator that a fresh ``parser.reset()`` cannot recover is restored
22at seek time from a bookmark's absolute timestamp: ``W = bookmark_ts -
23first_decoded_ts`` (a multiple of the format's wrap period), added to every
24timestamp decoded after the jump.
25"""
26from __future__ import annotations
28import struct
29from dataclasses import dataclass
30from pathlib import Path
32import numpy as np
35from typing import Protocol, TYPE_CHECKING
36if TYPE_CHECKING:
37 import numpy as np
39class SeekIndex(Protocol):
40 """Monotonic (timestamp, cumulative-count) -> word-offset bookmarks."""
42 @property
43 def n_events(self) -> int | None:
44 """Total event count of the indexed stream (None if not fully indexed)."""
45 ...
47 def bookmark_for_time(self, t: int) -> tuple[int, int, int]:
48 """Return (word_offset, cum_count, ts) for the last bookmark whose timestamp is <= t."""
49 ...
51 def bookmark_for_event(self, n: int) -> tuple[int, int, int]:
52 """Return (word_offset, cum_count, ts) for the last bookmark whose cumulative count is <= n."""
53 ...
55@dataclass
56class StaticSeekIndex:
57 """A fully built, read-only seek index."""
58 ts: np.ndarray
59 word_offset: np.ndarray
60 cum_count: np.ndarray
61 n_events: int
63 def bookmark_for_time(self, t: int) -> tuple[int, int, int]:
64 if len(self.ts) == 0:
65 return 0, 0, 0
66 i = int(np.searchsorted(self.ts, t, side="right")) - 1
67 i = max(i, 0)
68 return int(self.word_offset[i]), int(self.cum_count[i]), int(self.ts[i])
70 def bookmark_for_event(self, n: int) -> tuple[int, int, int]:
71 if len(self.cum_count) == 0:
72 return 0, 0, 0
73 i = int(np.searchsorted(self.cum_count, n, side="right")) - 1
74 i = max(i, 0)
75 return int(self.word_offset[i]), int(self.cum_count[i]), int(self.ts[i])
77 def __len__(self) -> int:
78 return len(self.ts)
80class IncrementalSeekIndex:
81 """A seek index that builds itself by decoding words on demand.
83 Bookmarks are aligned to TIME_HIGH words: each segment boundary is the
84 first TIME_HIGH at/after the previous boundary plus ``stride_words``. This
85 is what makes the seek-time wrap correction sound -- a parser reset at a
86 bookmark re-establishes its absolute time base *before* decoding any
87 event, so ``bookmark_ts - first_decoded_ts`` is an exact multiple of the
88 format's wrap period. A bookmark at an arbitrary parse-step boundary does
89 NOT have that property (events decoded before the next TIME_HIGH carry
90 only their low timestamp bits, and the snap lands a whole wrap period
91 off).
93 ``time_high`` is the ``(type-shift, type-code)`` descriptor of the
94 format's TIME_HIGH word. Without it (no descriptor known) the index
95 degrades to a single bookmark at ``start_offset`` -- correct, just no
96 mid-file acceleration.
97 """
98 def __init__(self, words: np.ndarray, start_offset: int, input_cls: type, parser_cls: type,
99 tail_pad: int, word_dtype: np.dtype, chunk_cap: int = 65_536,
100 time_high: "tuple[int, int] | None" = None, stride_words: int = 1 << 16):
101 self._words = words
102 self._start_offset = start_offset
103 self._input_cls = input_cls
104 self._parser = parser_cls()
105 self._tail_pad = tail_pad
106 self._word_dtype = word_dtype
107 self._time_high = time_high
108 self._stride = max(int(stride_words), 64)
110 cap = max(int(chunk_cap), 128)
111 from ._native_core import EventSoABuffers, TriggerSoABuffers
112 self._ev = EventSoABuffers(cap)
113 self._tr = TriggerSoABuffers(max(cap // 16, 1))
115 self._ts_list: list[int] = []
116 self._off_list: list[int] = []
117 self._cum_list: list[int] = []
119 self._cum = 0
120 self._off = int(start_offset)
121 self._is_eof = False
123 self._ts_arr = np.empty(0, dtype=np.int64)
124 self._off_arr = np.empty(0, dtype=np.int64)
125 self._cum_arr = np.empty(0, dtype=np.int64)
126 self._arrs_stale = False
128 @property
129 def n_events(self) -> int | None:
130 return self._cum if self._is_eof else None
132 def _update_arrs(self):
133 if self._arrs_stale:
134 self._ts_arr = np.asarray(self._ts_list, dtype=np.int64)
135 self._off_arr = np.asarray(self._off_list, dtype=np.int64)
136 self._cum_arr = np.asarray(self._cum_list, dtype=np.int64)
137 self._arrs_stale = False
139 def _next_time_high(self, start: int) -> int:
140 """Word offset of the first TIME_HIGH at/after ``start`` (or n_words).
142 Vectorized block scan, same shape as the decoder's
143 ``_find_first_time_high``.
144 """
145 n = len(self._words)
146 if self._time_high is None:
147 return n
148 shift, code = self._time_high
149 start = min(max(start, 0), n)
150 block = 1 << 16
151 while start < n:
152 stop = min(start + block, n)
153 seg = self._words[start:stop]
154 hits = ((seg >> shift) & 0xF) == code
155 i = int(np.argmax(hits))
156 if hits[i]:
157 return start + i
158 start = stop
159 block = min(block * 4, 1 << 24)
160 return n
162 def _build_until(self, target_t: int | None = None, target_n: int | None = None) -> None:
163 if self._is_eof:
164 return
165 if target_t is not None and len(self._ts_list) > 0 and self._ts_list[-1] >= target_t:
166 return
167 if target_n is not None and self._cum >= target_n:
168 return
170 from ._native_core import events_view, parse_step
171 words = self._words
172 n_words = len(words)
173 while self._off < n_words:
174 seg_start = self._off
175 seg_end = self._next_time_high(seg_start + self._stride)
176 first_ts: int | None = None
177 cum0 = self._cum
179 # Decode the whole segment [seg_start, seg_end); parser state is
180 # continuous across segments, only the *bookmark offsets* are
181 # TIME_HIGH-aligned.
182 while self._off < n_words and self._off < seg_end:
183 final = seg_end >= n_words
184 self._ev.reset()
185 self._tr.reset()
186 appended, new_off = parse_step(
187 words if final else words[:seg_end], self._off,
188 self._input_cls, self._parser, self._ev, self._tr,
189 tail_pad=self._tail_pad if final else 0,
190 word_dtype=self._word_dtype,
191 )
192 if appended == 0 and new_off <= self._off:
193 self._is_eof = True # zero progress on full input
194 break
195 if (not final and self._tail_pad and appended == 0
196 and new_off >= seg_end):
197 # Look-ahead formats (EVT3) stall a few words short of the
198 # sliced boundary and parse_step would skip that residue.
199 # Flush it through a zero-padded scratch copy: counts and
200 # timestamps stay exact (only decoder x/y state, which the
201 # index never records, can be perturbed by the pad words).
202 tail = words[self._off:seg_end]
203 if len(tail):
204 scratch = np.zeros(len(tail) + self._tail_pad,
205 dtype=self._word_dtype)
206 scratch[:len(tail)] = tail
207 self._ev.reset()
208 self._tr.reset()
209 parse_step(scratch, 0, self._input_cls, self._parser,
210 self._ev, self._tr, tail_pad=0,
211 word_dtype=self._word_dtype)
212 appended = self._ev.size
213 if appended and first_ts is None:
214 first_ts = int(events_view(self._ev).t[0])
215 self._cum += appended
216 self._off = seg_end
217 break
218 if appended:
219 if first_ts is None:
220 first_ts = int(events_view(self._ev).t[0])
221 self._cum += appended
222 self._off = new_off
224 if first_ts is not None:
225 self._ts_list.append(first_ts)
226 self._off_list.append(seg_start)
227 self._cum_list.append(cum0)
228 self._arrs_stale = True
230 if self._is_eof or self._off >= n_words:
231 self._is_eof = True
232 break
233 if target_t is not None and first_ts is not None and first_ts >= target_t:
234 break
235 if target_n is not None and self._cum >= target_n:
236 break
238 def bookmark_for_time(self, t: int) -> tuple[int, int, int]:
239 self._build_until(target_t=t)
240 self._update_arrs()
241 if len(self._ts_arr) == 0:
242 return self._start_offset, 0, 0
243 i = int(np.searchsorted(self._ts_arr, t, side="right")) - 1
244 i = max(i, 0)
245 return int(self._off_arr[i]), int(self._cum_arr[i]), int(self._ts_arr[i])
247 def bookmark_for_event(self, n: int) -> tuple[int, int, int]:
248 self._build_until(target_n=n)
249 self._update_arrs()
250 if len(self._cum_arr) == 0:
251 return self._start_offset, 0, 0
252 i = int(np.searchsorted(self._cum_arr, n, side="right")) - 1
253 i = max(i, 0)
254 return int(self._off_arr[i]), int(self._cum_arr[i]), int(self._ts_arr[i])
256 def build_all(self) -> None:
257 """Force the index to build every bookmark, all the way to EOF."""
258 # A target no real count can reach drives _build_until to the end of the
259 # stream (it stops on EOF, not on the target).
260 self._build_until(target_n=1 << 62)
261 self._update_arrs()
263 def to_static(self) -> "StaticSeekIndex":
264 """Fully build, then snapshot into an immutable :class:`StaticSeekIndex`
265 (with exact evutils cumulative counts) -- the form persisted to a
266 ``.evidx`` sidecar."""
267 self.build_all()
268 return StaticSeekIndex(
269 ts=self._ts_arr.copy(),
270 word_offset=self._off_arr.copy(),
271 cum_count=self._cum_arr.copy(),
272 n_events=int(self._cum),
273 )
276# --------------------------------------------------------------------------- #
277# Metavision / OpenEB `<file>.raw.tmp_index` sidecar
278# --------------------------------------------------------------------------- #
280#: On-disk bookmark record: (int64 timestamp, uint64 byte_offset, uint32 count).
281#: Serialization order per OpenEB's serialize_bookmark (NOT struct order).
282_MV_RECORD = np.dtype([("ts", "<i8"), ("byte_offset", "<u8"), ("count", "<u4")])
283_MV_RECORD_SIZE = 20 # packed; np.dtype above is already 20 (no padding)
286def metavision_index_path(raw_path: "str | Path") -> Path:
287 """Sidecar path for a raw file: ``<file>.raw.tmp_index``."""
288 p = Path(raw_path)
289 return p.with_name(p.name + ".tmp_index")
292def _parse_mv_header(buf: bytes) -> tuple[dict[str, str], int]:
293 """Parse the ``% key value`` text header; return (fields, payload offset)."""
294 fields: dict[str, str] = {}
295 off = 0
296 n = len(buf)
297 while off < n and buf[off:off + 1] == b"%":
298 nl = buf.find(b"\n", off)
299 if nl < 0:
300 break
301 line = buf[off:nl].decode("ascii", "ignore").strip()
302 off = nl + 1
303 if line == "% end":
304 break
305 parts = line.split(None, 2) # "%", key, value
306 if len(parts) >= 3:
307 fields[parts[1].lower()] = parts[2]
308 return fields, off
311def read_metavision_index(index_path: "str | Path", raw_path: "str | Path",
312 payload_off: int, word_size: int) -> "SeekIndex | None":
313 """Read a Metavision ``.tmp_index`` sidecar into a :class:`SeekIndex`.
315 Returns ``None`` if the sidecar is missing or stale (its stored ``size``
316 does not match the raw file's current byte size, matching Metavision's
317 freshness check). Byte offsets are converted to payload word offsets and
318 timestamps are shifted by ``ts_shift_us`` into the raw-stream timeline.
319 """
320 index_path = Path(index_path)
321 raw_path = Path(raw_path)
322 if not index_path.is_file() or not raw_path.is_file():
323 return None
325 data = index_path.read_bytes()
326 fields, body = _parse_mv_header(data)
328 # Freshness: stored size must equal the current raw byte size.
329 try:
330 if int(fields.get("size", "-1")) != raw_path.stat().st_size:
331 return None
332 except ValueError:
333 return None
335 ts_shift = 0
336 try:
337 ts_shift = int(fields.get("ts_shift_us", "0"))
338 except ValueError:
339 ts_shift = 0
341 payload = data[body:]
342 n_rec = len(payload) // _MV_RECORD_SIZE
343 if n_rec <= 1:
344 return None
345 recs = np.frombuffer(payload, dtype=_MV_RECORD, count=n_rec)
347 # The final record is the magic-number completeness marker (random bytes),
348 # not a bookmark -- drop it. Then drop leading pre-time-base bookmarks
349 # (ts < 0) and shift the rest into the raw-absolute timeline.
350 recs = recs[:-1]
351 recs = recs[recs["ts"] >= 0]
352 if len(recs) == 0:
353 return None
355 ts = recs["ts"].astype(np.int64) + ts_shift
356 byte_off = recs["byte_offset"].astype(np.int64)
357 word_offset = (byte_off - int(payload_off)) // int(word_size)
358 cum_count = np.cumsum(recs["count"].astype(np.int64)) - recs["count"].astype(np.int64)
360 return StaticSeekIndex(
361 ts=ts,
362 word_offset=word_offset,
363 cum_count=cum_count,
364 n_events=int(cum_count[-1] + recs["count"][-1]),
365 )
368# --------------------------------------------------------------------------- #
369# evutils own exact index `<file>.evidx` sidecar (persist across runs)
370# --------------------------------------------------------------------------- #
372#: Magic + version + freshness header for the `.evidx` sidecar. The header
373#: stores the raw file's byte size AND mtime; a mismatch on either means the
374#: sidecar is stale and is ignored (rebuilt), mirroring the Metavision reader's
375#: size check but a touch stricter.
376_EVIDX_MAGIC = b"EVUTLIDX"
377_EVIDX_VERSION = 1
378#: magic(8) + version(u32) + raw_size(i64) + raw_mtime_ns(i64) + n_events(i64)
379#: + n_bookmarks(i64), little-endian.
380_EVIDX_HEADER = struct.Struct("<8sIqqqq")
383def evutils_index_path(raw_path: "str | Path") -> Path:
384 """Sidecar path for evutils' own exact index: ``<file>.evidx``.
386 Distinct from the Metavision ``.tmp_index`` sidecar so the two never
387 collide -- this one holds evutils' TIME_HIGH-aligned bookmarks with exact
388 (evutils-counted) cumulative counts.
389 """
390 p = Path(raw_path)
391 return p.with_name(p.name + ".evidx")
394def save_seek_index(index: "SeekIndex", index_path: "str | Path",
395 raw_path: "str | Path") -> None:
396 """Serialize a fully-built seek index to a ``.evidx`` sidecar.
398 ``index`` may be an :class:`IncrementalSeekIndex` (fully built and snapshot
399 first) or an already-static :class:`StaticSeekIndex`. The raw file's current
400 size and mtime are stored in the header for the load-time freshness check.
401 """
402 if isinstance(index, IncrementalSeekIndex):
403 index = index.to_static()
404 if not isinstance(index, StaticSeekIndex):
405 raise TypeError("save_seek_index requires a Static/IncrementalSeekIndex")
407 raw = Path(raw_path)
408 st = raw.stat()
409 ts = np.ascontiguousarray(index.ts, dtype="<i8")
410 off = np.ascontiguousarray(index.word_offset, dtype="<i8")
411 cum = np.ascontiguousarray(index.cum_count, dtype="<i8")
412 n_bm = len(ts)
414 header = _EVIDX_HEADER.pack(
415 _EVIDX_MAGIC, _EVIDX_VERSION, int(st.st_size), int(st.st_mtime_ns),
416 int(index.n_events), int(n_bm),
417 )
418 # Write to a temp file then atomically replace, so a crash mid-write never
419 # leaves a truncated sidecar that would later read back as corrupt.
420 tmp = Path(index_path).with_name(Path(index_path).name + ".tmp")
421 with open(tmp, "wb") as f:
422 f.write(header)
423 f.write(ts.tobytes())
424 f.write(off.tobytes())
425 f.write(cum.tobytes())
426 tmp.replace(index_path)
429def load_seek_index(index_path: "str | Path",
430 raw_path: "str | Path") -> "StaticSeekIndex | None":
431 """Load a ``.evidx`` sidecar into a :class:`StaticSeekIndex`.
433 Returns ``None`` if the sidecar is missing, unreadable, has the wrong magic
434 / version, is truncated, or is stale (its stored raw-file size or mtime no
435 longer matches the raw file) -- in every case the caller rebuilds.
436 """
437 index_path = Path(index_path)
438 raw_path = Path(raw_path)
439 if not index_path.is_file() or not raw_path.is_file():
440 return None
442 data = index_path.read_bytes()
443 if len(data) < _EVIDX_HEADER.size:
444 return None
445 magic, version, raw_size, raw_mtime, n_events, n_bm = _EVIDX_HEADER.unpack_from(data, 0)
446 if magic != _EVIDX_MAGIC or version != _EVIDX_VERSION:
447 return None
449 # Freshness: the raw file must be byte-for-byte the one this index was built
450 # from (size AND mtime). Any change => stale => rebuild.
451 st = raw_path.stat()
452 if raw_size != st.st_size or raw_mtime != st.st_mtime_ns:
453 return None
455 if n_bm < 0 or n_events < 0:
456 return None
457 body = np.frombuffer(data, dtype="<i8", offset=_EVIDX_HEADER.size)
458 if len(body) != 3 * n_bm:
459 return None
460 ts = np.ascontiguousarray(body[:n_bm], dtype=np.int64)
461 word_offset = np.ascontiguousarray(body[n_bm:2 * n_bm], dtype=np.int64)
462 cum_count = np.ascontiguousarray(body[2 * n_bm:], dtype=np.int64)
463 return StaticSeekIndex(
464 ts=ts, word_offset=word_offset, cum_count=cum_count, n_events=int(n_events),
465 )