#cython: language_level=3
|
|
"""
|
Functions in this module give python-space wrappers for cython functions
|
exposed in numpy/__init__.pxd, so they can be tested in test_cython.py
|
"""
|
cimport numpy as cnp
|
cnp.import_array()
|
|
|
def is_td64(obj):
|
return cnp.is_timedelta64_object(obj)
|
|
|
def is_dt64(obj):
|
return cnp.is_datetime64_object(obj)
|
|
|
def get_dt64_value(obj):
|
return cnp.get_datetime64_value(obj)
|
|
|
def get_td64_value(obj):
|
return cnp.get_timedelta64_value(obj)
|
|
|
def get_dt64_unit(obj):
|
return cnp.get_datetime64_unit(obj)
|
|
|
def is_integer(obj):
|
return isinstance(obj, (cnp.integer, int))
|
|
|
def get_datetime_iso_8601_strlen():
|
return cnp.get_datetime_iso_8601_strlen(0, cnp.NPY_FR_ns)
|
|
|
def convert_datetime64_to_datetimestruct():
|
cdef:
|
cnp.npy_datetimestruct dts
|
cnp.PyArray_DatetimeMetaData meta
|
cnp.int64_t value = 1647374515260292
|
# i.e. (time.time() * 10**6) at 2022-03-15 20:01:55.260292 UTC
|
|
meta.base = cnp.NPY_FR_us
|
meta.num = 1
|
cnp.convert_datetime64_to_datetimestruct(&meta, value, &dts)
|
return dts
|
|
|
def make_iso_8601_datetime(dt: "datetime"):
|
cdef:
|
cnp.npy_datetimestruct dts
|
char result[36] # 36 corresponds to NPY_FR_s passed below
|
int local = 0
|
int utc = 0
|
int tzoffset = 0
|
|
dts.year = dt.year
|
dts.month = dt.month
|
dts.day = dt.day
|
dts.hour = dt.hour
|
dts.min = dt.minute
|
dts.sec = dt.second
|
dts.us = dt.microsecond
|
dts.ps = dts.as = 0
|
|
cnp.make_iso_8601_datetime(
|
&dts,
|
result,
|
sizeof(result),
|
local,
|
utc,
|
cnp.NPY_FR_s,
|
tzoffset,
|
cnp.NPY_NO_CASTING,
|
)
|
return result
|
|
|
cdef cnp.broadcast multiiter_from_broadcast_obj(object bcast):
|
cdef dict iter_map = {
|
1: cnp.PyArray_MultiIterNew1,
|
2: cnp.PyArray_MultiIterNew2,
|
3: cnp.PyArray_MultiIterNew3,
|
4: cnp.PyArray_MultiIterNew4,
|
5: cnp.PyArray_MultiIterNew5,
|
}
|
arrays = [x.base for x in bcast.iters]
|
cdef cnp.broadcast result = iter_map[len(arrays)](*arrays)
|
return result
|
|
|
def get_multiiter_size(bcast: "broadcast"):
|
cdef cnp.broadcast multi = multiiter_from_broadcast_obj(bcast)
|
return multi.size
|
|
|
def get_multiiter_number_of_dims(bcast: "broadcast"):
|
cdef cnp.broadcast multi = multiiter_from_broadcast_obj(bcast)
|
return multi.nd
|
|
|
def get_multiiter_current_index(bcast: "broadcast"):
|
cdef cnp.broadcast multi = multiiter_from_broadcast_obj(bcast)
|
return multi.index
|
|
|
def get_multiiter_num_of_iterators(bcast: "broadcast"):
|
cdef cnp.broadcast multi = multiiter_from_broadcast_obj(bcast)
|
return multi.numiter
|
|
|
def get_multiiter_shape(bcast: "broadcast"):
|
cdef cnp.broadcast multi = multiiter_from_broadcast_obj(bcast)
|
return tuple([multi.dimensions[i] for i in range(bcast.nd)])
|
|
|
def get_multiiter_iters(bcast: "broadcast"):
|
cdef cnp.broadcast multi = multiiter_from_broadcast_obj(bcast)
|
return tuple([<cnp.flatiter>multi.iters[i] for i in range(bcast.numiter)])
|
|
|
def get_default_integer():
|
if cnp.NPY_DEFAULT_INT == cnp.NPY_LONG:
|
return cnp.dtype("long")
|
if cnp.NPY_DEFAULT_INT == cnp.NPY_INTP:
|
return cnp.dtype("intp")
|
return None
|
|
def get_ravel_axis():
|
return cnp.NPY_RAVEL_AXIS
|
|
|
def conv_intp(cnp.intp_t val):
|
return val
|
|
|
def get_dtype_flags(cnp.dtype dtype):
|
return dtype.flags
|
|
|
cdef cnp.NpyIter* npyiter_from_nditer_obj(object it):
|
"""A function to create a NpyIter struct from a nditer object.
|
|
This function is only meant for testing purposes and only extracts the
|
necessary info from nditer to test the functionality of NpyIter methods
|
"""
|
cdef:
|
cnp.NpyIter* cit
|
cnp.PyArray_Descr* op_dtypes[3]
|
cnp.npy_uint32 op_flags[3]
|
cnp.PyArrayObject* ops[3]
|
cnp.npy_uint32 flags = 0
|
|
if it.has_index:
|
flags |= cnp.NPY_ITER_C_INDEX
|
if it.has_delayed_bufalloc:
|
flags |= cnp.NPY_ITER_BUFFERED | cnp.NPY_ITER_DELAY_BUFALLOC
|
if it.has_multi_index:
|
flags |= cnp.NPY_ITER_MULTI_INDEX
|
|
# one of READWRITE, READONLY and WRTIEONLY at the minimum must be specified for op_flags
|
for i in range(it.nop):
|
op_flags[i] = cnp.NPY_ITER_READONLY
|
|
for i in range(it.nop):
|
op_dtypes[i] = cnp.PyArray_DESCR(it.operands[i])
|
ops[i] = <cnp.PyArrayObject*>it.operands[i]
|
|
cit = cnp.NpyIter_MultiNew(it.nop, &ops[0], flags, cnp.NPY_KEEPORDER,
|
cnp.NPY_NO_CASTING, &op_flags[0],
|
<cnp.PyArray_Descr**>NULL)
|
return cit
|
|
|
def get_npyiter_size(it: "nditer"):
|
cdef cnp.NpyIter* cit = npyiter_from_nditer_obj(it)
|
result = cnp.NpyIter_GetIterSize(cit)
|
cnp.NpyIter_Deallocate(cit)
|
return result
|
|
|
def get_npyiter_ndim(it: "nditer"):
|
cdef cnp.NpyIter* cit = npyiter_from_nditer_obj(it)
|
result = cnp.NpyIter_GetNDim(cit)
|
cnp.NpyIter_Deallocate(cit)
|
return result
|
|
|
def get_npyiter_nop(it: "nditer"):
|
cdef cnp.NpyIter* cit = npyiter_from_nditer_obj(it)
|
result = cnp.NpyIter_GetNOp(cit)
|
cnp.NpyIter_Deallocate(cit)
|
return result
|
|
|
def get_npyiter_operands(it: "nditer"):
|
cdef cnp.NpyIter* cit = npyiter_from_nditer_obj(it)
|
try:
|
arr = cnp.NpyIter_GetOperandArray(cit)
|
return tuple([<cnp.ndarray>arr[i] for i in range(it.nop)])
|
finally:
|
cnp.NpyIter_Deallocate(cit)
|
|
|
def get_npyiter_itviews(it: "nditer"):
|
cdef cnp.NpyIter* cit = npyiter_from_nditer_obj(it)
|
result = tuple([cnp.NpyIter_GetIterView(cit, i) for i in range(it.nop)])
|
cnp.NpyIter_Deallocate(cit)
|
return result
|
|
|
def get_npyiter_dtypes(it: "nditer"):
|
cdef cnp.NpyIter* cit = npyiter_from_nditer_obj(it)
|
try:
|
arr = cnp.NpyIter_GetDescrArray(cit)
|
return tuple([<cnp.dtype>arr[i] for i in range(it.nop)])
|
finally:
|
cnp.NpyIter_Deallocate(cit)
|
|
|
def npyiter_has_delayed_bufalloc(it: "nditer"):
|
cdef cnp.NpyIter* cit = npyiter_from_nditer_obj(it)
|
result = cnp.NpyIter_HasDelayedBufAlloc(cit)
|
cnp.NpyIter_Deallocate(cit)
|
return result
|
|
|
def npyiter_has_index(it: "nditer"):
|
cdef cnp.NpyIter* cit = npyiter_from_nditer_obj(it)
|
result = cnp.NpyIter_HasIndex(cit)
|
cnp.NpyIter_Deallocate(cit)
|
return result
|
|
|
def npyiter_has_multi_index(it: "nditer"):
|
cdef cnp.NpyIter* cit = npyiter_from_nditer_obj(it)
|
result = cnp.NpyIter_HasMultiIndex(cit)
|
cnp.NpyIter_Deallocate(cit)
|
return result
|
|
|
def test_get_multi_index_iter_next(it: "nditer", cnp.ndarray[cnp.float64_t, ndim=2] arr):
|
cdef cnp.NpyIter* cit = npyiter_from_nditer_obj(it)
|
cdef cnp.NpyIter_GetMultiIndexFunc get_multi_index = \
|
cnp.NpyIter_GetGetMultiIndex(cit, NULL)
|
cdef cnp.NpyIter_IterNextFunc iternext = \
|
cnp.NpyIter_GetIterNext(cit, NULL)
|
return 1
|
|
|
def npyiter_has_finished(it: "nditer"):
|
cdef cnp.NpyIter* cit
|
try:
|
cit = npyiter_from_nditer_obj(it)
|
cnp.NpyIter_GotoIterIndex(cit, it.index)
|
return not (cnp.NpyIter_GetIterIndex(cit) < cnp.NpyIter_GetIterSize(cit))
|
finally:
|
cnp.NpyIter_Deallocate(cit)
|
|
def compile_fillwithbyte():
|
# Regression test for gh-25878, mostly checks it compiles.
|
cdef cnp.npy_intp dims[2]
|
dims = (1, 2)
|
pos = cnp.PyArray_ZEROS(2, dims, cnp.NPY_UINT8, 0)
|
cnp.PyArray_FILLWBYTE(pos, 1)
|
return pos
|
|
def inc2_cfloat_struct(cnp.ndarray[cnp.cfloat_t] arr):
|
# This works since we compile in C mode, it will fail in cpp mode
|
arr[1].real += 1
|
arr[1].imag += 1
|
# This works in both modes
|
arr[1].real = arr[1].real + 1
|
arr[1].imag = arr[1].imag + 1
|
|
|
def npystring_pack(arr):
|
cdef char *string = "Hello world"
|
cdef size_t size = 11
|
|
allocator = cnp.NpyString_acquire_allocator(
|
<cnp.PyArray_StringDTypeObject *>cnp.PyArray_DESCR(arr)
|
)
|
|
# copy string->packed_string, the pointer to the underlying array buffer
|
ret = cnp.NpyString_pack(
|
allocator, <cnp.npy_packed_static_string *>cnp.PyArray_DATA(arr), string, size,
|
)
|
|
cnp.NpyString_release_allocator(allocator)
|
return ret
|
|
|
def npystring_load(arr):
|
allocator = cnp.NpyString_acquire_allocator(
|
<cnp.PyArray_StringDTypeObject *>cnp.PyArray_DESCR(arr)
|
)
|
|
cdef cnp.npy_static_string sdata
|
sdata.size = 0
|
sdata.buf = NULL
|
|
cdef cnp.npy_packed_static_string *packed_string = <cnp.npy_packed_static_string *>cnp.PyArray_DATA(arr)
|
cdef int is_null = cnp.NpyString_load(allocator, packed_string, &sdata)
|
cnp.NpyString_release_allocator(allocator)
|
if is_null == -1:
|
raise ValueError("String unpacking failed.")
|
elif is_null == 1:
|
# String in the array buffer is the null string
|
return ""
|
else:
|
# Cython syntax for copying a c string to python bytestring:
|
# slice the char * by the length of the string
|
return sdata.buf[:sdata.size].decode('utf-8')
|
|
|
def npystring_pack_multiple(arr1, arr2):
|
cdef cnp.npy_string_allocator *allocators[2]
|
cdef cnp.PyArray_Descr *descrs[2]
|
descrs[0] = cnp.PyArray_DESCR(arr1)
|
descrs[1] = cnp.PyArray_DESCR(arr2)
|
|
cnp.NpyString_acquire_allocators(2, descrs, allocators)
|
|
# Write into the first element of each array
|
cdef int ret1 = cnp.NpyString_pack(
|
allocators[0], <cnp.npy_packed_static_string *>cnp.PyArray_DATA(arr1), "Hello world", 11,
|
)
|
cdef int ret2 = cnp.NpyString_pack(
|
allocators[1], <cnp.npy_packed_static_string *>cnp.PyArray_DATA(arr2), "test this", 9,
|
)
|
|
# Write a null string into the last element
|
cdef cnp.npy_intp elsize = cnp.PyArray_ITEMSIZE(arr1)
|
cdef int ret3 = cnp.NpyString_pack_null(
|
allocators[0],
|
<cnp.npy_packed_static_string *>(<char *>cnp.PyArray_DATA(arr1) + 2*elsize),
|
)
|
|
cnp.NpyString_release_allocators(2, allocators)
|
if ret1 == -1 or ret2 == -1 or ret3 == -1:
|
return -1
|
|
return 0
|
|
|
def npystring_allocators_other_types(arr1, arr2):
|
cdef cnp.npy_string_allocator *allocators[2]
|
cdef cnp.PyArray_Descr *descrs[2]
|
descrs[0] = cnp.PyArray_DESCR(arr1)
|
descrs[1] = cnp.PyArray_DESCR(arr2)
|
|
cnp.NpyString_acquire_allocators(2, descrs, allocators)
|
|
# None of the dtypes here are StringDType, so every allocator
|
# should be NULL upon acquisition.
|
cdef int ret = 0
|
for allocator in allocators:
|
if allocator != NULL:
|
ret = -1
|
break
|
|
cnp.NpyString_release_allocators(2, allocators)
|
return ret
|
|
|
def check_npy_uintp_type_enum():
|
# Regression test for gh-27890: cnp.NPY_UINTP was not defined.
|
# Cython would fail to compile this before gh-27890 was fixed.
|
return cnp.NPY_UINTP > 0
|
