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

1"""ctypes bindings for the native EVT2 / EVT2.1 / EVT3 / EVT4 parsers. 

2 

3Input wrappers (:class:`Evt2Input`, :class:`Evt21Input`, :class:`Evt3Input`, 

4:class:`Evt4Input`) view the payload words zero-copy; the ``Evt*Parser`` classes 

5own the opaque C parser state. Shared SoA output buffers and the parse loop 

6helpers live in :mod:`evutils.io._native_core`. 

7""" 

8from __future__ import annotations 

9import ctypes 

10from ctypes import POINTER, c_uint16, c_uint32, c_uint64, cast as c_cast, c_char, c_void_p, byref 

11from typing import cast 

12import numpy as np 

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

14 

15class Evt3InputBuffer(ctypes.Structure): 

16 _fields_ = [("begin", POINTER(c_uint16)), ("end", POINTER(c_uint16))] 

17class Evt2InputBuffer(ctypes.Structure): 

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

19class Evt21InputBuffer(ctypes.Structure): 

20 _fields_ = [("begin", POINTER(c_uint64)), ("end", POINTER(c_uint64))] 

21class Evt4InputBuffer(ctypes.Structure): 

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

23 

24class Evt3Input: 

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

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

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

28 self.arr = words 

29 base = words.ctypes.data 

30 self.c = Evt3InputBuffer() 

31 self.c.begin = c_cast(base, POINTER(c_uint16)) 

32 self.c.end = c_cast(base + words.nbytes, POINTER(c_uint16)) 

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

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

35 return (cur - self.arr.ctypes.data) // 2 

36 

37class Evt2Input: 

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

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

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

41 self.arr = words 

42 base = words.ctypes.data 

43 self.c = Evt2InputBuffer() 

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

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

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

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

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

49 

50class Evt21Input: 

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

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

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

54 self.arr = words 

55 base = words.ctypes.data 

56 self.c = Evt21InputBuffer() 

57 self.c.begin = c_cast(base, POINTER(c_uint64)) 

58 self.c.end = c_cast(base + words.nbytes, POINTER(c_uint64)) 

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

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

61 return (cur - self.arr.ctypes.data) // 8 

62 

63class Evt4Input: 

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

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

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

67 self.arr = words 

68 base = words.ctypes.data 

69 self.c = Evt4InputBuffer() 

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

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

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

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

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

75 

76def _bind_evt(handle: ctypes.CDLL) -> None: 

77 from ._native_core import EventBufferSOA, TriggerBufferSOA, ParserResult 

78 if hasattr(handle, "EVT3_state_size"): 

79 handle.EVT3_state_size.argtypes = [] 

80 handle.EVT3_state_size.restype = ctypes.c_size_t 

81 if hasattr(handle, "EVT3_parse_chunk_soa"): 

82 handle.EVT3_parse_chunk_soa.argtypes = [c_void_p, POINTER(Evt3InputBuffer), POINTER(EventBufferSOA), POINTER(TriggerBufferSOA)] 

83 handle.EVT3_parse_chunk_soa.restype = ParserResult 

84 if hasattr(handle, "EVT2_state_size"): 

85 handle.EVT2_state_size.argtypes = [] 

86 handle.EVT2_state_size.restype = ctypes.c_size_t 

87 if hasattr(handle, "EVT2_parse_chunk_soa"): 

88 handle.EVT2_parse_chunk_soa.argtypes = [c_void_p, POINTER(Evt2InputBuffer), POINTER(EventBufferSOA), POINTER(TriggerBufferSOA)] 

89 handle.EVT2_parse_chunk_soa.restype = ParserResult 

90 if hasattr(handle, "EVT21_state_size"): 

91 handle.EVT21_state_size.argtypes = [] 

92 handle.EVT21_state_size.restype = ctypes.c_size_t 

93 if hasattr(handle, "EVT21_parse_chunk_soa"): 

94 handle.EVT21_parse_chunk_soa.argtypes = [c_void_p, POINTER(Evt21InputBuffer), POINTER(EventBufferSOA), POINTER(TriggerBufferSOA)] 

95 handle.EVT21_parse_chunk_soa.restype = ParserResult 

96 if hasattr(handle, "EVT4_state_size"): 

97 handle.EVT4_state_size.argtypes = [] 

98 handle.EVT4_state_size.restype = ctypes.c_size_t 

99 if hasattr(handle, "EVT4_parse_chunk_soa"): 

100 handle.EVT4_parse_chunk_soa.argtypes = [c_void_p, POINTER(Evt4InputBuffer), POINTER(EventBufferSOA), POINTER(TriggerBufferSOA)] 

101 handle.EVT4_parse_chunk_soa.restype = ParserResult 

102 

103register_bindings(_bind_evt) 

104 

105class Evt3Parser: 

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

107 def __init__(self) -> None: 

108 self._buf = (c_char * int(lib().EVT3_state_size()))() 

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

110 def reset(self) -> None: ctypes.memset(self._buf, 0, len(self._buf)) 

111 def parse_chunk_soa(self, inp: Evt3Input, events: EventSoABuffers, triggers: TriggerSoABuffers) -> ParserResult: 

112 return cast(ParserResult, lib().EVT3_parse_chunk_soa(self._state, byref(inp.c), byref(events.c), byref(triggers.c))) 

113 def __enter__(self) -> "Evt3Parser": return self 

114 

115class Evt2Parser: 

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

117 def __init__(self) -> None: 

118 self._buf = (c_char * int(lib().EVT2_state_size()))() 

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

120 def reset(self) -> None: ctypes.memset(self._buf, 0, len(self._buf)) 

121 def parse_chunk_soa(self, inp: Evt2Input, events: EventSoABuffers, triggers: TriggerSoABuffers) -> ParserResult: 

122 return cast(ParserResult, lib().EVT2_parse_chunk_soa(self._state, byref(inp.c), byref(events.c), byref(triggers.c))) 

123 def __enter__(self) -> "Evt2Parser": return self 

124 

125class Evt21Parser: 

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

127 def __init__(self) -> None: 

128 self._buf = (c_char * int(lib().EVT21_state_size()))() 

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

130 def reset(self) -> None: ctypes.memset(self._buf, 0, len(self._buf)) 

131 def parse_chunk_soa(self, inp: Evt21Input, events: EventSoABuffers, triggers: TriggerSoABuffers) -> ParserResult: 

132 return cast(ParserResult, lib().EVT21_parse_chunk_soa(self._state, byref(inp.c), byref(events.c), byref(triggers.c))) 

133 def __enter__(self) -> "Evt21Parser": return self 

134 

135class Evt4Parser: 

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

137 def __init__(self) -> None: 

138 self._buf = (c_char * int(lib().EVT4_state_size()))() 

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

140 def reset(self) -> None: ctypes.memset(self._buf, 0, len(self._buf)) 

141 def parse_chunk_soa(self, inp: Evt4Input, events: EventSoABuffers, triggers: TriggerSoABuffers) -> ParserResult: 

142 return cast(ParserResult, lib().EVT4_parse_chunk_soa(self._state, byref(inp.c), byref(events.c), byref(triggers.c))) 

143 def __enter__(self) -> "Evt4Parser": return self