29360990938

Committed 14 Jul 2026 07:13PM UTC coverage: 97.307% (-0.04%) from 97.343%

Build # 29360990938

Build Type

Pull #232

github

Committed by

web-flow

Commit Message

Merge 760452ad3 into 3edf80d95

Pull Request Pull Request #232: Improve modularisation

Coverage Stats

189 of 192 new or added lines in 17 files covered. (98.44%)

19 existing lines in 3 files now uncovered.

1409 of 1448 relevant lines covered (97.31%)

0.97 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.57

/pelutils/_misc/array.py

import ctypes
from typing import TypeVar, cast

import _pelutils_c as _c
import numpy as np
import numpy.typing as npt

import pelutils._c as c_utils
from pelutils._misc.conditional_import import import_pandas, import_torch
from pelutils.types import AnyArray

pd = import_pandas()
torch = import_torch()

_ArrayT = TypeVar("_ArrayT", bound=npt.ArrayLike)


def array_ptr(arr: npt.ArrayLike) -> ctypes.c_void_p:
    """Return a pointer to a numpy array or torch tensor which can be used to interact with it in low-level languages like C/C++/Rust.

    This function is mostly useful when not using Python's C api and instead interfacing with .so files directly with ctypes.
    """
    if torch is not None and isinstance(arr, torch.Tensor):
        return ctypes.c_void_p(arr.data_ptr())
    if not isinstance(arr, np.ndarray):
        raise TypeError(f"Array should be of type np.ndarray or torch.Tensor, not {type(arr)}")
    if not arr.flags.c_contiguous:
        raise ValueError("Array must be C-contiguous")
    return ctypes.c_void_p(arr.ctypes.data)


def array_bytes(x: npt.ArrayLike) -> int:
    """Calculate the size of a numpy array or torch tensor in bytes."""
    if torch is not None and isinstance(x, torch.Tensor):
        x = x.numpy()
    if isinstance(x, np.ndarray):
        return x.nbytes
    raise TypeError(f"`x` of type {type(x)} is not a numpy array or torch tensor")


def unique(  # noqa: PLR0912
    array: _ArrayT,
    *,
    return_index: bool = False,
    return_inverse: bool = False,
    return_counts: bool = False,
    axis: int = 0,
) -> _ArrayT | tuple[_ArrayT, ...]:
    """Return unique elements in a given numpy array (or torch tensor or pandas series).

    This function works very similar to np.unique, but it runs in linear time, making it significantly faster
    for large arrays. The returned unique elements are unsorted, however.
    """
    is_tensor = False
    if torch is not None and isinstance(array, torch.Tensor):
        is_tensor = True
        np_array = array.numpy()
    elif pd is not None and isinstance(array, pd.Series):
        np_array = array.values
    elif isinstance(array, np.ndarray):
        if np.issubdtype(array.dtype, np.object_):
            raise TypeError(f"Unsupported array dtype {array.dtype}")
        np_array = array
    else:
        raise TypeError(f"Unsupported array type {type(array)}, must be numpy array, torch tensor, or pandas dataframe.")

    if np_array.ndim == 0:
        raise ValueError("unique does not work for shape-less arrays")

    np_array = cast(AnyArray, np_array)
    del array  # Prevent reuse - underlying tensor already referenced by np_array, which should be used from here

    if axis:
        axes = list(range(len(np_array.shape)))
        axes[0] = axis
        axes[axis] = 0
        np_array = np_array.transpose(axes)
    else:
        axes = None
    if not np_array.flags["C_CONTIGUOUS"]:
        np_array = np.ascontiguousarray(np_array)

    index = np.empty(len(np_array), dtype=np.int64)
    inverse = np.empty(len(np_array), dtype=np.int64) if return_inverse else None
    counts = np.empty(len(np_array), dtype=np.int64) if return_counts else None

    c = _c.unique(
        *c_utils.get_array_c_args(np_array),
        index.ctypes.data,
        inverse.ctypes.data if inverse is not None else 0,
        counts.ctypes.data if counts is not None else 0,
    )

    index = index[:c]
    if axis:
        np_array = np_array[index]
        np_array = np.ascontiguousarray(np_array.transpose(axes))
    else:
        np_array = np_array[index]
    ret = [np_array]
    if return_index:
        ret.append(index)
    if return_inverse:
        assert inverse is not None
        ret.append(inverse)
    if return_counts:
        assert counts is not None
        ret.append(counts[index])

    if torch is not None and is_tensor:
        ret = [torch.from_numpy(x) for x in ret]

    return tuple(ret) if len(ret) > 1 else ret[0]  # pyright: ignore[reportReturnType]

1	import ctypes	1✔
2	from typing import TypeVar, cast	1✔
3
4	import _pelutils_c as _c	1✔
5	import numpy as np	1✔
6	import numpy.typing as npt	1✔
7
8	import pelutils._c as c_utils	1✔
9	from pelutils._misc.conditional_import import import_pandas, import_torch	1✔
10	from pelutils.types import AnyArray	1✔
11
12	pd = import_pandas()	1✔
13	torch = import_torch()	1✔
14
15	_ArrayT = TypeVar("_ArrayT", bound=npt.ArrayLike)	1✔
16
17
18	def array_ptr(arr: npt.ArrayLike) -> ctypes.c_void_p:	1✔
19	"""Return a pointer to a numpy array or torch tensor which can be used to interact with it in low-level languages like C/C++/Rust.
20
21	This function is mostly useful when not using Python's C api and instead interfacing with .so files directly with ctypes.
22	"""
23	if torch is not None and isinstance(arr, torch.Tensor):	1✔
24	return ctypes.c_void_p(arr.data_ptr())	1✔
25	if not isinstance(arr, np.ndarray):	1✔
26	raise TypeError(f"Array should be of type np.ndarray or torch.Tensor, not {type(arr)}")	1✔
27	if not arr.flags.c_contiguous:	1✔
28	raise ValueError("Array must be C-contiguous")	1✔
29	return ctypes.c_void_p(arr.ctypes.data)	1✔
30
31
32	def array_bytes(x: npt.ArrayLike) -> int:	1✔
33	"""Calculate the size of a numpy array or torch tensor in bytes."""
34	if torch is not None and isinstance(x, torch.Tensor):	1✔
35	x = x.numpy()	1✔
36	if isinstance(x, np.ndarray):	1✔
37	return x.nbytes	1✔
NEW 38	raise TypeError(f"`x` of type {type(x)} is not a numpy array or torch tensor")	×
39
40
41	def unique( # noqa: PLR0912	1✔
42	array: _ArrayT,
43	*,
44	return_index: bool = False,
45	return_inverse: bool = False,
46	return_counts: bool = False,
47	axis: int = 0,
48	) -> _ArrayT \| tuple[_ArrayT, ...]:
49	"""Return unique elements in a given numpy array (or torch tensor or pandas series).
50
51	This function works very similar to np.unique, but it runs in linear time, making it significantly faster
52	for large arrays. The returned unique elements are unsorted, however.
53	"""
54	is_tensor = False	1✔
55	if torch is not None and isinstance(array, torch.Tensor):	1✔
56	is_tensor = True	1✔
57	np_array = array.numpy()	1✔
58	elif pd is not None and isinstance(array, pd.Series):	1✔
59	np_array = array.values	1✔
60	elif isinstance(array, np.ndarray):	1✔
61	if np.issubdtype(array.dtype, np.object_):	1✔
62	raise TypeError(f"Unsupported array dtype {array.dtype}")	1✔
63	np_array = array	1✔
64	else:
65	raise TypeError(f"Unsupported array type {type(array)}, must be numpy array, torch tensor, or pandas dataframe.")	1✔
66
67	if np_array.ndim == 0:	1✔
68	raise ValueError("unique does not work for shape-less arrays")	1✔
69
70	np_array = cast(AnyArray, np_array)	1✔
71	del array # Prevent reuse - underlying tensor already referenced by np_array, which should be used from here	1✔
72
73	if axis:	1✔
74	axes = list(range(len(np_array.shape)))	1✔
75	axes[0] = axis	1✔
76	axes[axis] = 0	1✔
77	np_array = np_array.transpose(axes)	1✔
78	else:
79	axes = None	1✔
80	if not np_array.flags["C_CONTIGUOUS"]:	1✔
81	np_array = np.ascontiguousarray(np_array)	1✔
82
83	index = np.empty(len(np_array), dtype=np.int64)	1✔
84	inverse = np.empty(len(np_array), dtype=np.int64) if return_inverse else None	1✔
85	counts = np.empty(len(np_array), dtype=np.int64) if return_counts else None	1✔
86
87	c = _c.unique(	1✔
88	*c_utils.get_array_c_args(np_array),
89	index.ctypes.data,
90	inverse.ctypes.data if inverse is not None else 0,
91	counts.ctypes.data if counts is not None else 0,
92	)
93
94	index = index[:c]	1✔
95	if axis:	1✔
96	np_array = np_array[index]	1✔
97	np_array = np.ascontiguousarray(np_array.transpose(axes))	1✔
98	else:
99	np_array = np_array[index]	1✔
100	ret = [np_array]	1✔
101	if return_index:	1✔
102	ret.append(index)	1✔
103	if return_inverse:	1✔
104	assert inverse is not None	1✔
105	ret.append(inverse)	1✔
106	if return_counts:	1✔
107	assert counts is not None	1✔
108	ret.append(counts[index])	1✔
109
110	if torch is not None and is_tensor:	1✔
111	ret = [torch.from_numpy(x) for x in ret]	1✔
112
113	return tuple(ret) if len(ret) > 1 else ret[0] # pyright: ignore[reportReturnType]	1✔

peleiden / pelutils / 29360990938

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous