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

1"""ctypes bindings for the native CSV/TXT parser and writer. 

2 

3Registers ``evutils_read_csv`` / ``evutils_write_csv`` (csrc/csv.c) with the 

4shared library handle; the Python-side chunking logic lives in 

5:mod:`evutils.io._csv`. 

6""" 

7from __future__ import annotations 

8import ctypes 

9from ctypes import POINTER, c_void_p, c_int, c_char_p, c_size_t, c_char, Structure, CDLL 

10from ._native_core import register_bindings 

11 

12class ParserResult(Structure): 

13 _fields_ = [ 

14 ("current", c_void_p), 

15 ("status", c_int) 

16 ] 

17 

18def _bind_csv(handle: CDLL) -> None: 

19 if hasattr(handle, "evutils_read_csv"): 

20 handle.evutils_read_csv.argtypes = [ 

21 c_char_p, # const char *buffer 

22 c_size_t, # size_t buffer_len 

23 c_char, # char delimiter 

24 POINTER(c_void_p), # void **out_arrays 

25 POINTER(c_int), # int *array_types 

26 POINTER(c_int), # int *col_mapping 

27 c_int, # int max_csv_cols 

28 c_size_t, # size_t max_events 

29 POINTER(c_size_t), # size_t *events_parsed 

30 POINTER(c_size_t), # size_t *malformed 

31 ] 

32 handle.evutils_read_csv.restype = ParserResult 

33 

34 if hasattr(handle, "evutils_write_csv"): 

35 handle.evutils_write_csv.argtypes = [ 

36 POINTER(c_void_p), # in_arrays 

37 POINTER(c_int), # array_types 

38 c_int, # num_columns 

39 c_char, # delimiter 

40 c_size_t, # num_events 

41 c_char_p, # out_buffer 

42 c_size_t, # out_buffer_len 

43 POINTER(c_size_t), # bytes_written 

44 POINTER(c_size_t), # events_written 

45 ] 

46 handle.evutils_write_csv.restype = c_int 

47 

48register_bindings(_bind_csv)