Coverage for src/evutils/io/_native_dat.py: 100%

1"""ctypes bindings for the native Prophesee DAT parser. 

2 

3:class:`DatInput` views the 2x-uint32 records zero-copy; :class:`DatParser` 

4owns the opaque C state (tracks the 32-bit timestamp overflow). Shared SoA 

5buffers and parse helpers live in :mod:`evutils.io._native_core`. 

6""" 

7from __future__ import annotations 

8import ctypes 

9from ctypes import POINTER, c_uint32, cast as c_cast, c_char, c_void_p, byref 

10from typing import cast 

11import numpy as np 

12from ._native_core import register_bindings, NativeError, EventSoABuffers, TriggerSoABuffers, ParserResult, lib 

13 

14class DatInputBuffer(ctypes.Structure): 

15 _fields_ = [("begin", POINTER(c_uint32)), ("end", POINTER(c_uint32))] 

16 

17class DatInput: 

18 __slots__ = ("arr", "c") 

19 def __init__(self, words: np.ndarray): 

20 if words.dtype != np.uint32 or not words.flags["C_CONTIGUOUS"]: raise NativeError("DatInput needs a C-contiguous uint32 array") 

21 self.arr = words 

22 base = words.ctypes.data 

23 self.c = DatInputBuffer() 

24 self.c.begin = c_cast(base, POINTER(c_uint32)) 

25 self.c.end = c_cast(base + words.nbytes, POINTER(c_uint32)) 

26 def consumed(self, result: ParserResult) -> int: 

27 cur = c_cast(result.current, c_void_p).value or self.arr.ctypes.data 

28 return (cur - self.arr.ctypes.data) // 4 

29 

30def _bind_dat(handle: ctypes.CDLL) -> None: 

31 from ._native_core import EventBufferSOA, TriggerBufferSOA, ParserResult 

32 if hasattr(handle, "DAT_state_size"): 

33 handle.DAT_state_size.argtypes = [] 

34 handle.DAT_state_size.restype = ctypes.c_size_t 

35 if hasattr(handle, "DAT_parse_chunk_soa"): 

36 handle.DAT_parse_chunk_soa.argtypes = [c_void_p, POINTER(DatInputBuffer), POINTER(EventBufferSOA), POINTER(TriggerBufferSOA)] 

37 handle.DAT_parse_chunk_soa.restype = ParserResult 

38 

39register_bindings(_bind_dat) 

40 

41class DatParser: 

42 __slots__ = ("_state", "_buf") 

43 def __init__(self) -> None: 

44 self._buf = (c_char * int(lib().DAT_state_size()))() 

45 self._state = c_cast(self._buf, c_void_p) 

46 def reset(self) -> None: 

47 ctypes.memset(self._buf, 0, len(self._buf)) 

48 def parse_chunk_soa(self, inp: DatInput, events: EventSoABuffers, triggers: TriggerSoABuffers) -> ParserResult: 

49 return cast(ParserResult, lib().DAT_parse_chunk_soa(self._state, byref(inp.c), byref(events.c), byref(triggers.c))) 

50 def __enter__(self) -> "DatParser": return self