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 

10from ._native_core import register_bindings 

11 

12def _bind_csv(handle: ctypes.CDLL) -> None: 

13 if hasattr(handle, "evutils_read_csv"): 

14 handle.evutils_read_csv.argtypes = [ 

15 c_char_p, # buffer 

16 c_size_t, # buffer_len 

17 c_char, # delimiter 

18 POINTER(c_void_p), # out_arrays 

19 POINTER(c_int), # array_types 

20 POINTER(c_int), # col_mapping 

21 c_int, # max_csv_cols 

22 c_size_t, # max_events 

23 POINTER(c_size_t), # bytes_consumed 

24 POINTER(c_size_t), # events_parsed 

25 ] 

26 handle.evutils_read_csv.restype = c_int 

27 

28 if hasattr(handle, "evutils_write_csv"): 

29 handle.evutils_write_csv.argtypes = [ 

30 POINTER(c_void_p), # in_arrays 

31 POINTER(c_int), # array_types 

32 c_int, # num_columns 

33 c_char, # delimiter 

34 c_size_t, # num_events 

35 c_char_p, # out_buffer 

36 c_size_t, # out_buffer_len 

37 POINTER(c_size_t), # bytes_written 

38 POINTER(c_size_t), # events_written 

39 ] 

40 handle.evutils_write_csv.restype = c_int 

41 

42register_bindings(_bind_csv)