29639696002

Committed 18 Jul 2026 09:44AM UTC coverage: 97.27% (+0.01%) from 97.258%

Build # 29639696002

Build Type

push

github

Committed by

asgerius

Commit Message

Add release section to README

Coverage Stats

1354 of 1392 relevant lines covered (97.27%)

0.97 hits per line

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

98.59

/pelutils/misc/_array.py

from __future__ import annotations

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

peleiden / pelutils / 29639696002

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