Coverage for src/evutils/io/buffer.py: 100%
90 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-15 09:33 +0000
1"""Event buffering module.
3Provides `EventAccumulator` for buffering and rotating event structures
4in memory efficiently.
5"""
7import numpy as np
9from ..types import EventArray, TriggerArray
10from ._native_core import EventSoABuffers, TriggerSoABuffers
13class EventAccumulator():
14 """Reused struct-of-arrays staging buffer that native decoders fill in place.
16 The whole point is to avoid a copy on the read path: instead of a decoder
17 parsing into its own buffer and the reader copying that into a ring, the
18 decoder's parser writes *directly* into this accumulator's storage (via
19 :meth:`prepare` + ``decoder.parse_step``). Consumed events at the front are
20 reclaimed by :meth:`_rotate` (moving only the small unconsumed remainder), so
21 the backing arrays are allocated once and never re-faulted. Only the final
22 window handed to the caller is copied (:meth:`slice_copy`).
24 Timestamps are stored as ``uint64`` (matching the native ``timestamp64_t``)
25 and exposed as ``int64`` via a zero-copy ``.view`` (values are positive and
26 in range).
27 """
29 def __init__(self, capacity: int):
30 """Initialize the event accumulator.
32 Parameters
33 ----------
34 capacity : int
35 The initial capacity for the event buffer.
37 """
38 self._capacity = int(capacity)
39 self._buf = EventSoABuffers(self._capacity)
40 self._tr_capacity = max(self._capacity // 16, 1)
41 self._tr = TriggerSoABuffers(self._tr_capacity)
42 self._start = 0 # events before this index are consumed
43 self._tr_start = 0
45 def __len__(self) -> int:
46 return self._buf.size - self._start
48 def t_window(self) -> np.ndarray:
49 """int64 view of the currently unconsumed timestamps (zero-copy).
51 Returns
52 -------
53 np.ndarray
54 The unconsumed timestamps as an int64 array.
56 """
57 s, e = self._start, self._buf.size
58 return self._buf.t[s:e].view(np.int64)
60 def t_window_tr(self) -> np.ndarray:
61 """int64 view of the currently unconsumed trigger timestamps (zero-copy).
63 Returns
64 -------
65 np.ndarray
66 The unconsumed trigger timestamps as an int64 array.
68 """
69 s, e = self._tr_start, self._tr.size
70 return self._tr.t[s:e].view(np.int64)
72 def _rotate(self) -> None:
73 """Move the unconsumed remainder to the front, freeing consumed space."""
74 s = self._start
75 if s > 0:
76 b = self._buf
77 n = b.size - s
78 if n > 0:
79 b.t[:n] = b.t[s:b.size]
80 b.x[:n] = b.x[s:b.size]
81 b.y[:n] = b.y[s:b.size]
82 b.p[:n] = b.p[s:b.size]
83 b.c.size = n
84 self._start = 0
86 s_tr = self._tr_start
87 if s_tr > 0:
88 t = self._tr
89 n_tr = t.size - s_tr
90 if n_tr > 0:
91 t.t[:n_tr] = t.t[s_tr:t.size]
92 t.id[:n_tr] = t.id[s_tr:t.size]
93 t.p[:n_tr] = t.p[s_tr:t.size]
94 t.c.size = n_tr
95 self._tr_start = 0
97 def prepare(self, step: int) -> tuple[EventSoABuffers, TriggerSoABuffers]:
98 """Ready the buffer for the decoder to append up to ``step`` events, and
99 return ``(events_soa, triggers_soa)`` for ``decoder.parse_step``.
101 Rotates out consumed events if there is not enough tail room, then caps
102 the SoA capacity the parser sees to ``size + step`` so a single step does
103 not overshoot the requested window by more than one step's worth.
105 Parameters
106 ----------
107 step : int
108 The number of events the decoder is expected to append.
110 Returns
111 -------
112 tuple
113 A tuple of (events_soa, triggers_soa) buffers.
115 """
116 b = self._buf
117 t = self._tr
118 if self._capacity - b.size < step or t.capacity - t.size < step // 16:
119 self._rotate()
120 b.c.capacity = min(self._capacity, b.size + step)
121 t.c.capacity = t.capacity
122 return b, t
124 def append(self, data: EventArray, triggers: TriggerArray | None = None) -> None:
125 """Copy an EventArray in (fallback path for non-native decoders).
127 Parameters
128 ----------
129 data : EventArray
130 The events to append.
131 triggers : TriggerArray, optional
132 The triggers to append.
134 """
135 n = len(data)
136 if n > 0:
137 b = self._buf
138 if self._capacity - b.size < n:
139 self._rotate()
140 if self._capacity - b.size < n:
141 raise ValueError(
142 f"EventAccumulator full: can't append {n} ({self._capacity - b.size} free)"
143 )
144 e = b.size
145 b.t[e:e + n] = data.t
146 b.x[e:e + n] = data.x
147 b.y[e:e + n] = data.y
148 b.p[e:e + n] = data.p
149 b.c.size = e + n
151 if triggers is not None and len(triggers) > 0:
152 n_tr = len(triggers)
153 t = self._tr
154 if t.capacity - t.size < n_tr:
155 self._rotate()
156 if t.capacity - t.size < n_tr:
157 t.grow(t.size + n_tr)
158 e_tr = t.size
159 t.t[e_tr:e_tr + n_tr] = triggers.t
160 t.id[e_tr:e_tr + n_tr] = triggers.id
161 t.p[e_tr:e_tr + n_tr] = triggers.p
162 t.c.size = e_tr + n_tr
164 def slice_copy(self, k: int, tr_k: int = 0) -> tuple[EventArray, TriggerArray]:
165 """Return an independent copy of the first ``k`` unconsumed events and
166 advance past them.
168 Parameters
169 ----------
170 k : int
171 The number of unconsumed events to slice and copy.
172 tr_k : int, optional
173 The number of unconsumed triggers to slice and copy.
175 Returns
176 -------
177 tuple
178 A tuple of (EventArray, TriggerArray) containing the copied slices.
180 """
181 s = self._buf
182 i = self._start
183 out = EventArray(
184 s.t[i:i + k].view(np.int64), s.x[i:i + k], s.y[i:i + k], s.p[i:i + k]
185 ).copy()
186 self._start += k
188 t = self._tr
189 j = self._tr_start
190 out_tr = TriggerArray(
191 t.t[j:j + tr_k].view(np.int64), t.p[j:j + tr_k], t.id[j:j + tr_k]
192 ).copy()
193 self._tr_start += tr_k
195 return out, out_tr
197 def reset(self) -> None:
198 """Reset the buffer and trigger size to 0 and clear consumed offsets."""
199 self._buf.c.size = 0
200 self._tr.c.size = 0
201 self._start = 0
202 self._tr_start = 0