Make an __(cuda)_array_interface__ from a pointer.
(
ptr: Union[CNumericPtr, int],
shape: Tuple[int, ...],
dtype: Type[np.number],
is_cuda: bool,
)
| 192 | # Typing is not strict as there are subtle differences between CUDA array interface and |
| 193 | # array interface. We handle them uniformly for now. |
| 194 | def make_array_interface( |
| 195 | ptr: Union[CNumericPtr, int], |
| 196 | shape: Tuple[int, ...], |
| 197 | dtype: Type[np.number], |
| 198 | is_cuda: bool, |
| 199 | ) -> ArrayInf: |
| 200 | """Make an __(cuda)_array_interface__ from a pointer.""" |
| 201 | # Use an empty array to handle typestr and descr |
| 202 | if is_cuda: |
| 203 | empty = import_cupy().empty(shape=(0,), dtype=dtype) |
| 204 | array = empty.__cuda_array_interface__ # pylint: disable=no-member |
| 205 | else: |
| 206 | empty = np.empty(shape=(0,), dtype=dtype) |
| 207 | array = empty.__array_interface__ # pylint: disable=no-member |
| 208 | |
| 209 | if not isinstance(ptr, int): |
| 210 | addr = ctypes.cast(ptr, ctypes.c_void_p).value |
| 211 | else: |
| 212 | addr = ptr |
| 213 | length = int(np.prod(shape)) |
| 214 | # Handle empty dataset. |
| 215 | assert addr is not None or length == 0 |
| 216 | |
| 217 | if addr is None: |
| 218 | return array |
| 219 | |
| 220 | array["data"] = (addr, True) |
| 221 | if is_cuda and "stream" not in array: |
| 222 | array["stream"] = STREAM_PER_THREAD |
| 223 | array["shape"] = shape |
| 224 | array["strides"] = None |
| 225 | return array |
| 226 | |
| 227 | |
| 228 | def is_arrow_dict(data: Any) -> TypeGuard["pa.DictionaryArray"]: |
no test coverage detected