Coverage for src/evutils/io/_native_core.py: 84%

1"""Low-level ctypes binding and common structs for the native library. 

2""" 

3from __future__ import annotations 

4 

5import ctypes 

6import os 

7import sys 

8from ctypes import ( 

9 POINTER, Structure, byref, cast as c_cast, c_char_p, c_char, 

10 c_int, c_size_t, c_uint8, c_uint16, c_uint64, c_void_p, 

11) 

12from pathlib import Path 

13from typing import cast, Callable 

14 

15import numpy as np 

16from ..types import EventArray, TriggerArray 

17 

18__all__ = [ 

19 "NativeError", "lib", "register_bindings", 

20 "EventBufferSOA", "TriggerBufferSOA", 

21 "ParserResult", "EventSoABuffers", "TriggerSoABuffers", 

22 "EVENT_DTYPE", "TRIGGER_DTYPE", 

23 "events_view", "triggers_view", 

24 "parse_step", "decode_all_soa", "_handle_parse_warning", 

25 "EVUTILS_PARSE_OK", "EVUTILS_PARSE_INPUT_EMPTY", "EVUTILS_PARSE_OUTPUT_FULL", 

26 "EVUTILS_PARSE_ERROR", "EVUTILS_PARSE_INCOMPLETE", "EVUTILS_PARSE_WINDOW_DONE", "EVUTILS_PARSE_WARNING", 

27 "_T_DTYPE", "_X_DTYPE", "_Y_DTYPE", "_P_DTYPE", "_ID_DTYPE" 

28] 

29 

30class NativeError(RuntimeError): 

31 pass 

32 

33EVENT_DTYPE = np.dtype({"names": ["t", "x", "y", "p"], "formats": ["<u4", "<u2", "<u2", "u1"], "offsets": [0, 4, 6, 8], "itemsize": 12}) 

34TRIGGER_DTYPE = np.dtype({"names": ["t", "id", "p"], "formats": ["<u4", "u1", "u1"], "offsets": [0, 4, 5], "itemsize": 8}) 

35 

36_T_DTYPE = np.uint64 

37_X_DTYPE = np.uint16 

38_Y_DTYPE = np.uint16 

39_P_DTYPE = np.uint8 

40_ID_DTYPE = np.uint8 

41 

42class EventBufferSOA(Structure): 

43 _fields_ = [ 

44 ("t", POINTER(c_uint64)), ("x", POINTER(c_uint16)), 

45 ("y", POINTER(c_uint16)), ("p", POINTER(c_uint8)), 

46 ("capacity", c_size_t), ("size", c_size_t), 

47 ] 

48 

49class TriggerBufferSOA(Structure): 

50 _fields_ = [ 

51 ("t", POINTER(c_uint64)), ("id", POINTER(c_uint8)), ("p", POINTER(c_uint8)), 

52 ("capacity", c_size_t), ("size", c_size_t), 

53 ] 

54 

55EVUTILS_PARSE_OK = 0 

56EVUTILS_PARSE_INPUT_EMPTY = 1 

57EVUTILS_PARSE_OUTPUT_FULL = 2 

58EVUTILS_PARSE_ERROR = 3 

59EVUTILS_PARSE_INCOMPLETE = 4 

60EVUTILS_PARSE_WINDOW_DONE = 5 

61EVUTILS_PARSE_WARNING = 6 

62 

63class ParserResult(Structure): 

64 _fields_ = [("current", POINTER(c_uint16)), ("status", c_int)] 

65 

66def _candidate_filenames() -> list[str]: 

67 base = "evutils_native" 

68 if sys.platform.startswith("win"): return [f"{base}.dll", f"lib{base}.dll"] 

69 if sys.platform == "darwin": return [f"lib{base}.dylib", f"{base}.dylib"] 

70 return [f"lib{base}.so", f"{base}.so"] 

71 

72def _search_roots() -> list[Path]: 

73 here = Path(__file__).resolve().parent 

74 roots = [here] 

75 for parent in (here, *here.parents[:5]): 

76 if (parent / "pyproject.toml").exists(): 

77 roots.append(parent / "build") 

78 break 

79 roots.append(Path.cwd() / "build") 

80 return roots 

81 

82def _find_library() -> str: 

83 override = os.environ.get("EVUTILS_NATIVE_LIB") 

84 if override: return override 

85 names = set(_candidate_filenames()) 

86 for root in _search_roots(): 

87 if not root.exists(): continue 

88 for name in names: 

89 p = root / name 

90 if p.is_file(): return str(p) 

91 for p in root.glob("**/*evutils_native*"): 

92 if p.is_file() and p.name in names: return str(p) 

93 raise NativeError("Could not find the evutils native library.") 

94 

95_BINDINGS: list[Callable[[ctypes.CDLL], None]] = [] 

96 

97def register_bindings(binder: Callable[[ctypes.CDLL], None]) -> None: 

98 _BINDINGS.append(binder) 

99 # If already loaded, bind immediately 

100 if _LIB is not None: 

101 binder(_LIB) 

102 

103def _bind(handle: ctypes.CDLL) -> ctypes.CDLL: 

104 handle.evutils_version.argtypes = [] 

105 handle.evutils_version.restype = c_char_p 

106 for binder in _BINDINGS: 

107 binder(handle) 

108 return handle 

109 

110_LIB: ctypes.CDLL | None = None 

111 

112def lib() -> ctypes.CDLL: 

113 global _LIB 

114 if _LIB is None: 

115 try: 

116 handle = ctypes.CDLL(_find_library()) 

117 except OSError as exc: 

118 raise NativeError(f"Failed to load evutils native library: {exc}") from exc 

119 _LIB = _bind(handle) 

120 return _LIB 

121 

122class EventSoABuffers: 

123 __slots__ = ("capacity", "t", "x", "y", "p", "c") 

124 def __init__(self, capacity: int): 

125 self.capacity = int(capacity) 

126 self.t = np.empty(self.capacity, dtype=_T_DTYPE) 

127 self.x = np.empty(self.capacity, dtype=_X_DTYPE) 

128 self.y = np.empty(self.capacity, dtype=_Y_DTYPE) 

129 self.p = np.empty(self.capacity, dtype=_P_DTYPE) 

130 self.c = EventBufferSOA() 

131 self.c.t = self.t.ctypes.data_as(POINTER(c_uint64)) 

132 self.c.x = self.x.ctypes.data_as(POINTER(c_uint16)) 

133 self.c.y = self.y.ctypes.data_as(POINTER(c_uint16)) 

134 self.c.p = self.p.ctypes.data_as(POINTER(c_uint8)) 

135 self.c.capacity = self.capacity 

136 self.c.size = 0 

137 @property 

138 def size(self) -> int: return int(self.c.size) 

139 def reset(self) -> None: self.c.size = 0 

140 def grow(self, new_capacity: int) -> None: 

141 new_capacity = int(new_capacity) 

142 if new_capacity <= self.capacity: return 

143 n = self.size 

144 for name, dtype, field, ptr in ( 

145 ("t", _T_DTYPE, "t", c_uint64), ("x", _X_DTYPE, "x", c_uint16), 

146 ("y", _Y_DTYPE, "y", c_uint16), ("p", _P_DTYPE, "p", c_uint8), 

147 ): 

148 grown = np.empty(new_capacity, dtype=dtype) 

149 grown[:n] = getattr(self, name)[:n] 

150 setattr(self, name, grown) 

151 setattr(self.c, field, grown.ctypes.data_as(POINTER(ptr))) 

152 self.capacity = new_capacity 

153 self.c.capacity = new_capacity 

154 def view(self) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]: 

155 n = self.size 

156 return self.t[:n], self.x[:n], self.y[:n], self.p[:n] 

157 

158class TriggerSoABuffers: 

159 __slots__ = ("capacity", "t", "id", "p", "c") 

160 def __init__(self, capacity: int): 

161 self.capacity = int(capacity) 

162 self.t = np.empty(self.capacity, dtype=_T_DTYPE) 

163 self.id = np.empty(self.capacity, dtype=_ID_DTYPE) 

164 self.p = np.empty(self.capacity, dtype=_P_DTYPE) 

165 self.c = TriggerBufferSOA() 

166 self.c.t = self.t.ctypes.data_as(POINTER(c_uint64)) 

167 self.c.id = self.id.ctypes.data_as(POINTER(c_uint8)) 

168 self.c.p = self.p.ctypes.data_as(POINTER(c_uint8)) 

169 self.c.capacity = self.capacity 

170 self.c.size = 0 

171 @property 

172 def size(self) -> int: return int(self.c.size) 

173 def reset(self) -> None: self.c.size = 0 

174 def grow(self, new_capacity: int) -> None: 

175 new_capacity = int(new_capacity) 

176 if new_capacity <= self.capacity: return 

177 n = self.size 

178 for name, dtype, ptr in ( 

179 ("t", _T_DTYPE, c_uint64), ("id", _ID_DTYPE, c_uint8), ("p", _P_DTYPE, c_uint8), 

180 ): 

181 grown = np.empty(new_capacity, dtype=dtype) 

182 grown[:n] = getattr(self, name)[:n] 

183 setattr(self, name, grown) 

184 setattr(self.c, name, grown.ctypes.data_as(POINTER(ptr))) 

185 self.capacity = new_capacity 

186 self.c.capacity = new_capacity 

187 def view(self) -> tuple[np.ndarray, np.ndarray, np.ndarray]: 

188 n = self.size 

189 return self.t[:n], self.id[:n], self.p[:n] 

190 

191def events_view(ev: EventSoABuffers) -> EventArray: 

192 n = ev.size 

193 return EventArray(ev.t[:n].view(np.int64), ev.x[:n], ev.y[:n], ev.p[:n]) 

194 

195def triggers_view(tr: TriggerSoABuffers) -> TriggerArray: 

196 n = tr.size 

197 return TriggerArray(tr.t[:n].view(np.int64), tr.p[:n], tr.id[:n]) 

198 

199def _handle_parse_warning(loc: "int | str", strict: bool, fmt: "str | None" = None) -> None: 

200 """Raise (strict) or warn (default) on an ``EVUTILS_PARSE_WARNING`` -- a 

201 malformed packet the parser skipped over. 

202 

203 Single source of truth for the strict-vs-warn decision, shared by every 

204 parser binding (``parse_step``, ``decode_all_soa``, the EVT delta_t parser 

205 and the CSV reader). 

206 

207 ``loc`` is the location descriptor appended to the message: an int word 

208 offset (rendered as ``"near word N"``) for the binary parsers, or a ready 

209 string (e.g. CSV's malformed-row count) for text formats. ``fmt`` prefixes 

210 the format name and switches the strict exception to :class:`NativeError` 

211 (the EVT decoder path); without it a bare ``RuntimeError`` is raised. 

212 """ 

213 where = f"near word {loc}" if isinstance(loc, int) else str(loc) 

214 if strict: 

215 if fmt: 

216 raise NativeError(f"{fmt} malformed packet {where} (strict mode)") 

217 raise RuntimeError(f"malformed packet {where} (strict mode)") 

218 import warnings 

219 if fmt: 

220 warnings.warn(f"{fmt} malformed packets ignored {where}") 

221 else: 

222 warnings.warn(f"Malformed packets ignored {where}") 

223 

224 

225def parse_step(words: "np.ndarray", offset: int, make_input: "Callable", parser: "Callable", events: "EventArray", triggers: "TriggerArray", *, tail_pad: int = 0, word_dtype: "np.dtype | None" = None, strict: bool = False) -> tuple[int, int]: 

226 n_words = len(words) 

227 if offset >= n_words: return 0, n_words 

228 before = events.size 

229 inp = make_input(words[offset:]) 

230 while True: 

231 res = parser.parse_chunk_soa(inp, events, triggers) 

232 consumed = inp.consumed(res) 

233 if res.status == EVUTILS_PARSE_WARNING: 

234 _handle_parse_warning(offset + consumed, strict) 

235 if consumed > 0: 

236 offset += consumed 

237 inp = make_input(words[offset:]) 

238 continue 

239 elif res.status == EVUTILS_PARSE_ERROR: 

240 raise RuntimeError(f"parse error near word {offset + consumed}") 

241 break 

242 if consumed == 0: 

243 # Distinguish WHY nothing was consumed. A full output buffer means the 

244 # input is intact and the caller must drain/grow and retry -- flushing 

245 # the tail here would silently skip unread input. 

246 if res.status == EVUTILS_PARSE_OUTPUT_FULL: 

247 return events.size - before, offset 

248 # Input truly drained mid-group: only the sub-padding tail remains. 

249 if tail_pad and word_dtype is not None: 

250 tail = words[offset:] 

251 if len(tail): 

252 scratch = np.zeros(len(tail) + tail_pad, dtype=word_dtype) 

253 scratch[: len(tail)] = tail 

254 parser.parse_chunk_soa(make_input(scratch), events, triggers) 

255 return events.size - before, n_words 

256 return events.size - before, offset + consumed 

257 

258def decode_all_soa(words: "np.ndarray", start_offset: int, make_input: "Callable", parser: "Callable", *, est_events_per_word: float = 1.0, tail_pad: int = 0, word_dtype: "np.dtype | None" = None, trigger_cap: int = 1 << 12, strict: bool = False) -> tuple[EventArray, int]: 

259 n_words = len(words) if words is not None else 0 

260 if n_words == 0 or start_offset >= n_words: return EventArray.empty(), n_words 

261 remaining = n_words - start_offset 

262 cap = int(remaining * est_events_per_word) + 1024 

263 ev = EventSoABuffers(cap) 

264 tr = TriggerSoABuffers(max(trigger_cap, 1)) 

265 offset = start_offset 

266 while offset < n_words: 

267 if ev.capacity - ev.size < 128: 

268 # Extrapolate the true event count from the fraction of input 

269 # consumed so far and grow straight to it (+15% slack). A too-low 

270 # est_events_per_word therefore self-corrects in a *single* realloc 

271 # instead of repeatedly growing by a constant factor -- dense EVT2.1 

272 # vector streams emit up to 32 events per 64-bit word, which the old 

273 # 1.5x-per-grow schedule reached only after ~7 full-buffer copies. 

274 consumed_frac = (offset - start_offset) / remaining 

275 if consumed_frac > 0.02: 

276 projected = int(ev.size / consumed_frac * 1.15) + (1 << 16) 

277 else: 

278 projected = ev.capacity * 2 + (1 << 16) 

279 ev.grow(max(projected, ev.capacity + (1 << 16))) 

280 if tr.capacity - tr.size < 64: tr.grow(tr.capacity * 2 + 64) 

281 inp = make_input(words[offset:]) 

282 while True: 

283 res = parser.parse_chunk_soa(inp, ev, tr) 

284 consumed = inp.consumed(res) 

285 if res.status == EVUTILS_PARSE_WARNING: 

286 _handle_parse_warning(offset + consumed, strict) 

287 if consumed > 0: 

288 offset += consumed 

289 inp = make_input(words[offset:]) 

290 continue 

291 elif res.status == EVUTILS_PARSE_ERROR: 

292 raise RuntimeError(f"parse error near word {offset + consumed}") 

293 break 

294 if consumed == 0: 

295 if tail_pad and word_dtype is not None: 

296 tail = words[offset:] 

297 if len(tail): 

298 if ev.capacity - ev.size < len(tail) + 128: ev.grow(ev.size + len(tail) + (1 << 16)) 

299 scratch = np.zeros(len(tail) + tail_pad, dtype=word_dtype) 

300 scratch[: len(tail)] = tail 

301 parser.parse_chunk_soa(make_input(scratch), ev, tr) 

302 offset = n_words 

303 break 

304 offset += consumed 

305 n = ev.size 

306 if n == 0: return EventArray.empty(), offset 

307 out = EventArray(ev.t[:n].view(np.int64), ev.x[:n], ev.y[:n], ev.p[:n]) 

308 return out, offset