feat: add Camera Work AI package

This commit is contained in:
KINDNICK
2026-08-03 04:42:44 +09:00
parent 26c578416f
commit 02d76bc87c
2187 changed files with 119610 additions and 1 deletions
@@ -0,0 +1,268 @@
from libc.math cimport sqrt, exp
from ..utils._typedefs cimport float64_t, float32_t, int32_t, intp_t
cdef class DistanceMetric:
pass
######################################################################
# Inline distance functions
#
# We use these for the default (euclidean) case so that they can be
# inlined. This leads to faster computation for the most common case
cdef inline float64_t euclidean_dist64(
const float64_t* x1,
const float64_t* x2,
intp_t size,
) except -1 nogil:
cdef float64_t tmp, d=0
cdef intp_t j
for j in range(size):
tmp = <float64_t> (x1[j] - x2[j])
d += tmp * tmp
return sqrt(d)
cdef inline float64_t euclidean_rdist64(
const float64_t* x1,
const float64_t* x2,
intp_t size,
) except -1 nogil:
cdef float64_t tmp, d=0
cdef intp_t j
for j in range(size):
tmp = <float64_t>(x1[j] - x2[j])
d += tmp * tmp
return d
cdef inline float64_t euclidean_dist_to_rdist64(const float64_t dist) except -1 nogil:
return dist * dist
cdef inline float64_t euclidean_rdist_to_dist64(const float64_t dist) except -1 nogil:
return sqrt(dist)
######################################################################
# DistanceMetric64 base class
cdef class DistanceMetric64(DistanceMetric):
# The following attributes are required for a few of the subclasses.
# we must define them here so that cython's limited polymorphism will work.
# Because we don't expect to instantiate a lot of these objects, the
# extra memory overhead of this setup should not be an issue.
cdef float64_t p
cdef const float64_t[::1] vec
cdef const float64_t[:, ::1] mat
cdef intp_t size
cdef object func
cdef object kwargs
cdef float64_t dist(
self,
const float64_t* x1,
const float64_t* x2,
intp_t size,
) except -1 nogil
cdef float64_t rdist(
self,
const float64_t* x1,
const float64_t* x2,
intp_t size,
) except -1 nogil
cdef float64_t dist_csr(
self,
const float64_t* x1_data,
const int32_t* x1_indices,
const float64_t* x2_data,
const int32_t* x2_indices,
const int32_t x1_start,
const int32_t x1_end,
const int32_t x2_start,
const int32_t x2_end,
const intp_t size,
) except -1 nogil
cdef float64_t rdist_csr(
self,
const float64_t* x1_data,
const int32_t* x1_indices,
const float64_t* x2_data,
const int32_t* x2_indices,
const int32_t x1_start,
const int32_t x1_end,
const int32_t x2_start,
const int32_t x2_end,
const intp_t size,
) except -1 nogil
cdef int pdist(
self,
const float64_t[:, ::1] X,
float64_t[:, ::1] D,
) except -1
cdef int cdist(
self,
const float64_t[:, ::1] X,
const float64_t[:, ::1] Y,
float64_t[:, ::1] D,
) except -1
cdef int pdist_csr(
self,
const float64_t* x1_data,
const int32_t[::1] x1_indices,
const int32_t[::1] x1_indptr,
const intp_t size,
float64_t[:, ::1] D,
) except -1 nogil
cdef int cdist_csr(
self,
const float64_t* x1_data,
const int32_t[::1] x1_indices,
const int32_t[::1] x1_indptr,
const float64_t* x2_data,
const int32_t[::1] x2_indices,
const int32_t[::1] x2_indptr,
const intp_t size,
float64_t[:, ::1] D,
) except -1 nogil
cdef float64_t _rdist_to_dist(self, float64_t rdist) except -1 nogil
cdef float64_t _dist_to_rdist(self, float64_t dist) except -1 nogil
######################################################################
# Inline distance functions
#
# We use these for the default (euclidean) case so that they can be
# inlined. This leads to faster computation for the most common case
cdef inline float64_t euclidean_dist32(
const float32_t* x1,
const float32_t* x2,
intp_t size,
) except -1 nogil:
cdef float64_t tmp, d=0
cdef intp_t j
for j in range(size):
tmp = <float64_t> (x1[j] - x2[j])
d += tmp * tmp
return sqrt(d)
cdef inline float64_t euclidean_rdist32(
const float32_t* x1,
const float32_t* x2,
intp_t size,
) except -1 nogil:
cdef float64_t tmp, d=0
cdef intp_t j
for j in range(size):
tmp = <float64_t>(x1[j] - x2[j])
d += tmp * tmp
return d
cdef inline float64_t euclidean_dist_to_rdist32(const float32_t dist) except -1 nogil:
return dist * dist
cdef inline float64_t euclidean_rdist_to_dist32(const float32_t dist) except -1 nogil:
return sqrt(dist)
######################################################################
# DistanceMetric32 base class
cdef class DistanceMetric32(DistanceMetric):
# The following attributes are required for a few of the subclasses.
# we must define them here so that cython's limited polymorphism will work.
# Because we don't expect to instantiate a lot of these objects, the
# extra memory overhead of this setup should not be an issue.
cdef float64_t p
cdef const float64_t[::1] vec
cdef const float64_t[:, ::1] mat
cdef intp_t size
cdef object func
cdef object kwargs
cdef float32_t dist(
self,
const float32_t* x1,
const float32_t* x2,
intp_t size,
) except -1 nogil
cdef float32_t rdist(
self,
const float32_t* x1,
const float32_t* x2,
intp_t size,
) except -1 nogil
cdef float32_t dist_csr(
self,
const float32_t* x1_data,
const int32_t* x1_indices,
const float32_t* x2_data,
const int32_t* x2_indices,
const int32_t x1_start,
const int32_t x1_end,
const int32_t x2_start,
const int32_t x2_end,
const intp_t size,
) except -1 nogil
cdef float32_t rdist_csr(
self,
const float32_t* x1_data,
const int32_t* x1_indices,
const float32_t* x2_data,
const int32_t* x2_indices,
const int32_t x1_start,
const int32_t x1_end,
const int32_t x2_start,
const int32_t x2_end,
const intp_t size,
) except -1 nogil
cdef int pdist(
self,
const float32_t[:, ::1] X,
float32_t[:, ::1] D,
) except -1
cdef int cdist(
self,
const float32_t[:, ::1] X,
const float32_t[:, ::1] Y,
float32_t[:, ::1] D,
) except -1
cdef int pdist_csr(
self,
const float32_t* x1_data,
const int32_t[::1] x1_indices,
const int32_t[::1] x1_indptr,
const intp_t size,
float32_t[:, ::1] D,
) except -1 nogil
cdef int cdist_csr(
self,
const float32_t* x1_data,
const int32_t[::1] x1_indices,
const int32_t[::1] x1_indptr,
const float32_t* x2_data,
const int32_t[::1] x2_indices,
const int32_t[::1] x2_indptr,
const intp_t size,
float32_t[:, ::1] D,
) except -1 nogil
cdef float32_t _rdist_to_dist(self, float32_t rdist) except -1 nogil
cdef float32_t _dist_to_rdist(self, float32_t dist) except -1 nogil
@@ -0,0 +1,152 @@
{{py:
implementation_specific_values = [
# Values are the following ones:
#
# name_suffix, INPUT_DTYPE_t, INPUT_DTYPE
('64', 'float64_t', 'np.float64'),
('32', 'float32_t', 'np.float32')
]
}}
from libc.math cimport sqrt, exp
from ..utils._typedefs cimport float64_t, float32_t, int32_t, intp_t
cdef class DistanceMetric:
pass
{{for name_suffix, INPUT_DTYPE_t, INPUT_DTYPE in implementation_specific_values}}
######################################################################
# Inline distance functions
#
# We use these for the default (euclidean) case so that they can be
# inlined. This leads to faster computation for the most common case
cdef inline float64_t euclidean_dist{{name_suffix}}(
const {{INPUT_DTYPE_t}}* x1,
const {{INPUT_DTYPE_t}}* x2,
intp_t size,
) except -1 nogil:
cdef float64_t tmp, d=0
cdef intp_t j
for j in range(size):
tmp = <float64_t> (x1[j] - x2[j])
d += tmp * tmp
return sqrt(d)
cdef inline float64_t euclidean_rdist{{name_suffix}}(
const {{INPUT_DTYPE_t}}* x1,
const {{INPUT_DTYPE_t}}* x2,
intp_t size,
) except -1 nogil:
cdef float64_t tmp, d=0
cdef intp_t j
for j in range(size):
tmp = <float64_t>(x1[j] - x2[j])
d += tmp * tmp
return d
cdef inline float64_t euclidean_dist_to_rdist{{name_suffix}}(const {{INPUT_DTYPE_t}} dist) except -1 nogil:
return dist * dist
cdef inline float64_t euclidean_rdist_to_dist{{name_suffix}}(const {{INPUT_DTYPE_t}} dist) except -1 nogil:
return sqrt(dist)
######################################################################
# DistanceMetric{{name_suffix}} base class
cdef class DistanceMetric{{name_suffix}}(DistanceMetric):
# The following attributes are required for a few of the subclasses.
# we must define them here so that cython's limited polymorphism will work.
# Because we don't expect to instantiate a lot of these objects, the
# extra memory overhead of this setup should not be an issue.
cdef float64_t p
cdef const float64_t[::1] vec
cdef const float64_t[:, ::1] mat
cdef intp_t size
cdef object func
cdef object kwargs
cdef {{INPUT_DTYPE_t}} dist(
self,
const {{INPUT_DTYPE_t}}* x1,
const {{INPUT_DTYPE_t}}* x2,
intp_t size,
) except -1 nogil
cdef {{INPUT_DTYPE_t}} rdist(
self,
const {{INPUT_DTYPE_t}}* x1,
const {{INPUT_DTYPE_t}}* x2,
intp_t size,
) except -1 nogil
cdef {{INPUT_DTYPE_t}} dist_csr(
self,
const {{INPUT_DTYPE_t}}* x1_data,
const int32_t* x1_indices,
const {{INPUT_DTYPE_t}}* x2_data,
const int32_t* x2_indices,
const int32_t x1_start,
const int32_t x1_end,
const int32_t x2_start,
const int32_t x2_end,
const intp_t size,
) except -1 nogil
cdef {{INPUT_DTYPE_t}} rdist_csr(
self,
const {{INPUT_DTYPE_t}}* x1_data,
const int32_t* x1_indices,
const {{INPUT_DTYPE_t}}* x2_data,
const int32_t* x2_indices,
const int32_t x1_start,
const int32_t x1_end,
const int32_t x2_start,
const int32_t x2_end,
const intp_t size,
) except -1 nogil
cdef int pdist(
self,
const {{INPUT_DTYPE_t}}[:, ::1] X,
{{INPUT_DTYPE_t}}[:, ::1] D,
) except -1
cdef int cdist(
self,
const {{INPUT_DTYPE_t}}[:, ::1] X,
const {{INPUT_DTYPE_t}}[:, ::1] Y,
{{INPUT_DTYPE_t}}[:, ::1] D,
) except -1
cdef int pdist_csr(
self,
const {{INPUT_DTYPE_t}}* x1_data,
const int32_t[::1] x1_indices,
const int32_t[::1] x1_indptr,
const intp_t size,
{{INPUT_DTYPE_t}}[:, ::1] D,
) except -1 nogil
cdef int cdist_csr(
self,
const {{INPUT_DTYPE_t}}* x1_data,
const int32_t[::1] x1_indices,
const int32_t[::1] x1_indptr,
const {{INPUT_DTYPE_t}}* x2_data,
const int32_t[::1] x2_indices,
const int32_t[::1] x2_indptr,
const intp_t size,
{{INPUT_DTYPE_t}}[:, ::1] D,
) except -1 nogil
cdef {{INPUT_DTYPE_t}} _rdist_to_dist(self, {{INPUT_DTYPE_t}} rdist) except -1 nogil
cdef {{INPUT_DTYPE_t}} _dist_to_rdist(self, {{INPUT_DTYPE_t}} dist) except -1 nogil
{{endfor}}
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,31 @@
from ...utils._typedefs cimport intp_t, float64_t
{{for name_suffix in ['64', '32']}}
from ._base cimport BaseDistancesReduction{{name_suffix}}
from ._middle_term_computer cimport MiddleTermComputer{{name_suffix}}
cdef class ArgKmin{{name_suffix}}(BaseDistancesReduction{{name_suffix}}):
"""float{{name_suffix}} implementation of the ArgKmin."""
cdef:
intp_t k
intp_t[:, ::1] argkmin_indices
float64_t[:, ::1] argkmin_distances
# Used as array of pointers to private datastructures used in threads.
float64_t ** heaps_r_distances_chunks
intp_t ** heaps_indices_chunks
cdef class EuclideanArgKmin{{name_suffix}}(ArgKmin{{name_suffix}}):
"""EuclideanDistance-specialisation of ArgKmin{{name_suffix}}."""
cdef:
MiddleTermComputer{{name_suffix}} middle_term_computer
const float64_t[::1] X_norm_squared
const float64_t[::1] Y_norm_squared
bint use_squared_distances
{{endfor}}
@@ -0,0 +1,512 @@
from libc.stdlib cimport free, malloc
from libc.float cimport DBL_MAX
from cython cimport final
from cython.parallel cimport parallel, prange
from ...utils._heap cimport heap_push
from ...utils._sorting cimport simultaneous_sort
from ...utils._typedefs cimport intp_t, float64_t
import numpy as np
import warnings
from numbers import Integral
from scipy.sparse import issparse
from ...utils import check_array, check_scalar
from ...utils.fixes import _in_unstable_openblas_configuration
from ...utils.parallel import _get_threadpool_controller
{{for name_suffix in ['64', '32']}}
from ._base cimport (
BaseDistancesReduction{{name_suffix}},
_sqeuclidean_row_norms{{name_suffix}},
)
from ._datasets_pair cimport DatasetsPair{{name_suffix}}
from ._middle_term_computer cimport MiddleTermComputer{{name_suffix}}
cdef class ArgKmin{{name_suffix}}(BaseDistancesReduction{{name_suffix}}):
"""float{{name_suffix}} implementation of the ArgKmin."""
@classmethod
def compute(
cls,
X,
Y,
intp_t k,
metric="euclidean",
chunk_size=None,
dict metric_kwargs=None,
str strategy=None,
bint return_distance=False,
):
"""Compute the argkmin reduction.
This classmethod is responsible for introspecting the arguments
values to dispatch to the most appropriate implementation of
:class:`ArgKmin{{name_suffix}}`.
This allows decoupling the API entirely from the implementation details
whilst maintaining RAII: all temporarily allocated datastructures necessary
for the concrete implementation are therefore freed when this classmethod
returns.
No instance should directly be created outside of this class method.
"""
# Limit the number of threads in second level of nested parallelism for BLAS
# to avoid threads over-subscription (in DOT or GEMM for instance).
with _get_threadpool_controller().limit(limits=1, user_api='blas'):
if metric in ("euclidean", "sqeuclidean"):
# Specialized implementation of ArgKmin for the Euclidean distance
# for the dense-dense and sparse-sparse cases.
# This implementation computes the distances by chunk using
# a decomposition of the Squared Euclidean distance.
# This specialisation has an improved arithmetic intensity for both
# the dense and sparse settings, allowing in most case speed-ups of
# several orders of magnitude compared to the generic ArgKmin
# implementation.
# Note that squared norms of X and Y are precomputed in the
# constructor of this class by issuing BLAS calls that may use
# multithreading (depending on the BLAS implementation), hence calling
# the constructor needs to be protected under the threadpool_limits
# context, along with the main calls to _parallel_on_Y and
# _parallel_on_X.
# For more information see MiddleTermComputer.
use_squared_distances = metric == "sqeuclidean"
pda = EuclideanArgKmin{{name_suffix}}(
X=X, Y=Y, k=k,
use_squared_distances=use_squared_distances,
chunk_size=chunk_size,
strategy=strategy,
metric_kwargs=metric_kwargs,
)
else:
# Fall back on a generic implementation that handles most scipy
# metrics by computing the distances between 2 vectors at a time.
pda = ArgKmin{{name_suffix}}(
datasets_pair=DatasetsPair{{name_suffix}}.get_for(X, Y, metric, metric_kwargs),
k=k,
chunk_size=chunk_size,
strategy=strategy,
)
if pda.execute_in_parallel_on_Y:
pda._parallel_on_Y()
else:
pda._parallel_on_X()
return pda._finalize_results(return_distance)
def __init__(
self,
DatasetsPair{{name_suffix}} datasets_pair,
chunk_size=None,
strategy=None,
intp_t k=1,
):
super().__init__(
datasets_pair=datasets_pair,
chunk_size=chunk_size,
strategy=strategy,
)
self.k = check_scalar(k, "k", Integral, min_val=1)
# Allocating pointers to datastructures but not the datastructures themselves.
# There are as many pointers as effective threads.
#
# For the sake of explicitness:
# - when parallelizing on X, the pointers of those heaps are referencing
# (with proper offsets) addresses of the two main heaps (see below)
# - when parallelizing on Y, the pointers of those heaps are referencing
# small heaps which are thread-wise-allocated and whose content will be
# merged with the main heaps'.
self.heaps_r_distances_chunks = <float64_t **> malloc(
sizeof(float64_t *) * self.chunks_n_threads
)
self.heaps_indices_chunks = <intp_t **> malloc(
sizeof(intp_t *) * self.chunks_n_threads
)
# Main heaps which will be returned as results by `ArgKmin{{name_suffix}}.compute`.
self.argkmin_indices = np.full((self.n_samples_X, self.k), 0, dtype=np.intp)
self.argkmin_distances = np.full((self.n_samples_X, self.k), DBL_MAX, dtype=np.float64)
def __dealloc__(self):
if self.heaps_indices_chunks is not NULL:
free(self.heaps_indices_chunks)
if self.heaps_r_distances_chunks is not NULL:
free(self.heaps_r_distances_chunks)
cdef void _compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil:
cdef:
intp_t i, j
intp_t n_samples_X = X_end - X_start
intp_t n_samples_Y = Y_end - Y_start
float64_t *heaps_r_distances = self.heaps_r_distances_chunks[thread_num]
intp_t *heaps_indices = self.heaps_indices_chunks[thread_num]
# Pushing the distances and their associated indices on a heap
# which by construction will keep track of the argkmin.
for i in range(n_samples_X):
for j in range(n_samples_Y):
heap_push(
values=heaps_r_distances + i * self.k,
indices=heaps_indices + i * self.k,
size=self.k,
val=self.datasets_pair.surrogate_dist(X_start + i, Y_start + j),
val_idx=Y_start + j,
)
cdef void _parallel_on_X_init_chunk(
self,
intp_t thread_num,
intp_t X_start,
intp_t X_end,
) noexcept nogil:
# As this strategy is embarrassingly parallel, we can set each
# thread's heaps pointer to the proper position on the main heaps.
self.heaps_r_distances_chunks[thread_num] = &self.argkmin_distances[X_start, 0]
self.heaps_indices_chunks[thread_num] = &self.argkmin_indices[X_start, 0]
cdef void _parallel_on_X_prange_iter_finalize(
self,
intp_t thread_num,
intp_t X_start,
intp_t X_end,
) noexcept nogil:
cdef:
intp_t idx
# Sorting the main heaps portion associated to `X[X_start:X_end]`
# in ascending order w.r.t the distances.
for idx in range(X_end - X_start):
simultaneous_sort(
self.heaps_r_distances_chunks[thread_num] + idx * self.k,
self.heaps_indices_chunks[thread_num] + idx * self.k,
self.k
)
cdef void _parallel_on_Y_init(
self,
) noexcept nogil:
cdef:
# Maximum number of scalar elements (the last chunks can be smaller)
intp_t heaps_size = self.X_n_samples_chunk * self.k
intp_t thread_num
# The allocation is done in parallel for data locality purposes: this way
# the heaps used in each threads are allocated in pages which are closer
# to the CPU core used by the thread.
# See comments about First Touch Placement Policy:
# https://www.openmp.org/wp-content/uploads/openmp-webinar-vanderPas-20210318.pdf #noqa
for thread_num in prange(self.chunks_n_threads, schedule='static', nogil=True,
num_threads=self.chunks_n_threads):
# As chunks of X are shared across threads, so must their
# heaps. To solve this, each thread has its own heaps
# which are then synchronised back in the main ones.
self.heaps_r_distances_chunks[thread_num] = <float64_t *> malloc(
heaps_size * sizeof(float64_t)
)
self.heaps_indices_chunks[thread_num] = <intp_t *> malloc(
heaps_size * sizeof(intp_t)
)
cdef void _parallel_on_Y_parallel_init(
self,
intp_t thread_num,
intp_t X_start,
intp_t X_end,
) noexcept nogil:
# Initialising heaps (memset can't be used here)
for idx in range(self.X_n_samples_chunk * self.k):
self.heaps_r_distances_chunks[thread_num][idx] = DBL_MAX
self.heaps_indices_chunks[thread_num][idx] = -1
@final
cdef void _parallel_on_Y_synchronize(
self,
intp_t X_start,
intp_t X_end,
) noexcept nogil:
cdef:
intp_t idx, jdx, thread_num
with nogil, parallel(num_threads=self.effective_n_threads):
# Synchronising the thread heaps with the main heaps.
# This is done in parallel sample-wise (no need for locks).
#
# This might break each thread's data locality as each heap which
# was allocated in a thread is being now being used in several threads.
#
# Still, this parallel pattern has shown to be efficient in practice.
for idx in prange(X_end - X_start, schedule="static"):
for thread_num in range(self.chunks_n_threads):
for jdx in range(self.k):
heap_push(
values=&self.argkmin_distances[X_start + idx, 0],
indices=&self.argkmin_indices[X_start + idx, 0],
size=self.k,
val=self.heaps_r_distances_chunks[thread_num][idx * self.k + jdx],
val_idx=self.heaps_indices_chunks[thread_num][idx * self.k + jdx],
)
cdef void _parallel_on_Y_finalize(
self,
) noexcept nogil:
cdef:
intp_t idx, thread_num
with nogil, parallel(num_threads=self.chunks_n_threads):
# Deallocating temporary datastructures
for thread_num in prange(self.chunks_n_threads, schedule='static'):
free(self.heaps_r_distances_chunks[thread_num])
free(self.heaps_indices_chunks[thread_num])
# Sorting the main in ascending order w.r.t the distances.
# This is done in parallel sample-wise (no need for locks).
for idx in prange(self.n_samples_X, schedule='static'):
simultaneous_sort(
&self.argkmin_distances[idx, 0],
&self.argkmin_indices[idx, 0],
self.k,
)
return
cdef void compute_exact_distances(self) noexcept nogil:
cdef:
intp_t i, j
float64_t[:, ::1] distances = self.argkmin_distances
for i in prange(self.n_samples_X, schedule='static', nogil=True,
num_threads=self.effective_n_threads):
for j in range(self.k):
distances[i, j] = self.datasets_pair.distance_metric._rdist_to_dist(
# Guard against potential -0., causing nan production.
max(distances[i, j], 0.)
)
def _finalize_results(self, bint return_distance=False):
if return_distance:
# We need to recompute distances because we relied on
# surrogate distances for the reduction.
self.compute_exact_distances()
# Values are returned identically to the way `KNeighborsMixin.kneighbors`
# returns values. This is counter-intuitive but this allows not using
# complex adaptations where `ArgKmin.compute` is called.
return np.asarray(self.argkmin_distances), np.asarray(self.argkmin_indices)
return np.asarray(self.argkmin_indices)
cdef class EuclideanArgKmin{{name_suffix}}(ArgKmin{{name_suffix}}):
"""EuclideanDistance-specialisation of ArgKmin{{name_suffix}}."""
@classmethod
def is_usable_for(cls, X, Y, metric) -> bool:
return (ArgKmin{{name_suffix}}.is_usable_for(X, Y, metric) and
not _in_unstable_openblas_configuration())
def __init__(
self,
X,
Y,
intp_t k,
bint use_squared_distances=False,
chunk_size=None,
strategy=None,
metric_kwargs=None,
):
if (
isinstance(metric_kwargs, dict) and
(metric_kwargs.keys() - {"X_norm_squared", "Y_norm_squared"})
):
warnings.warn(
f"Some metric_kwargs have been passed ({metric_kwargs}) but aren't "
f"usable for this case (EuclideanArgKmin64) and will be ignored.",
UserWarning,
stacklevel=3,
)
super().__init__(
# The datasets pair here is used for exact distances computations
datasets_pair=DatasetsPair{{name_suffix}}.get_for(X, Y, metric="euclidean"),
chunk_size=chunk_size,
strategy=strategy,
k=k,
)
cdef:
intp_t dist_middle_terms_chunks_size = self.Y_n_samples_chunk * self.X_n_samples_chunk
self.middle_term_computer = MiddleTermComputer{{name_suffix}}.get_for(
X,
Y,
self.effective_n_threads,
self.chunks_n_threads,
dist_middle_terms_chunks_size,
n_features=X.shape[1],
chunk_size=self.chunk_size,
)
if metric_kwargs is not None and "Y_norm_squared" in metric_kwargs:
self.Y_norm_squared = check_array(
metric_kwargs.pop("Y_norm_squared"),
ensure_2d=False,
input_name="Y_norm_squared",
dtype=np.float64,
)
else:
self.Y_norm_squared = _sqeuclidean_row_norms{{name_suffix}}(
Y,
self.effective_n_threads,
)
if metric_kwargs is not None and "X_norm_squared" in metric_kwargs:
self.X_norm_squared = check_array(
metric_kwargs.pop("X_norm_squared"),
ensure_2d=False,
input_name="X_norm_squared",
dtype=np.float64,
)
else:
# Do not recompute norms if datasets are identical.
self.X_norm_squared = (
self.Y_norm_squared if X is Y else
_sqeuclidean_row_norms{{name_suffix}}(
X,
self.effective_n_threads,
)
)
self.use_squared_distances = use_squared_distances
@final
cdef void compute_exact_distances(self) noexcept nogil:
if not self.use_squared_distances:
ArgKmin{{name_suffix}}.compute_exact_distances(self)
@final
cdef void _parallel_on_X_parallel_init(
self,
intp_t thread_num,
) noexcept nogil:
ArgKmin{{name_suffix}}._parallel_on_X_parallel_init(self, thread_num)
self.middle_term_computer._parallel_on_X_parallel_init(thread_num)
@final
cdef void _parallel_on_X_init_chunk(
self,
intp_t thread_num,
intp_t X_start,
intp_t X_end,
) noexcept nogil:
ArgKmin{{name_suffix}}._parallel_on_X_init_chunk(self, thread_num, X_start, X_end)
self.middle_term_computer._parallel_on_X_init_chunk(thread_num, X_start, X_end)
@final
cdef void _parallel_on_X_pre_compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil:
ArgKmin{{name_suffix}}._parallel_on_X_pre_compute_and_reduce_distances_on_chunks(
self,
X_start, X_end,
Y_start, Y_end,
thread_num,
)
self.middle_term_computer._parallel_on_X_pre_compute_and_reduce_distances_on_chunks(
X_start, X_end, Y_start, Y_end, thread_num,
)
@final
cdef void _parallel_on_Y_init(
self,
) noexcept nogil:
ArgKmin{{name_suffix}}._parallel_on_Y_init(self)
self.middle_term_computer._parallel_on_Y_init()
@final
cdef void _parallel_on_Y_parallel_init(
self,
intp_t thread_num,
intp_t X_start,
intp_t X_end,
) noexcept nogil:
ArgKmin{{name_suffix}}._parallel_on_Y_parallel_init(self, thread_num, X_start, X_end)
self.middle_term_computer._parallel_on_Y_parallel_init(thread_num, X_start, X_end)
@final
cdef void _parallel_on_Y_pre_compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil:
ArgKmin{{name_suffix}}._parallel_on_Y_pre_compute_and_reduce_distances_on_chunks(
self,
X_start, X_end,
Y_start, Y_end,
thread_num,
)
self.middle_term_computer._parallel_on_Y_pre_compute_and_reduce_distances_on_chunks(
X_start, X_end, Y_start, Y_end, thread_num
)
@final
cdef void _compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil:
cdef:
intp_t i, j
float64_t sqeuclidean_dist_i_j
intp_t n_X = X_end - X_start
intp_t n_Y = Y_end - Y_start
float64_t * dist_middle_terms = self.middle_term_computer._compute_dist_middle_terms(
X_start, X_end, Y_start, Y_end, thread_num
)
float64_t * heaps_r_distances = self.heaps_r_distances_chunks[thread_num]
intp_t * heaps_indices = self.heaps_indices_chunks[thread_num]
# Pushing the distance and their associated indices on heaps
# which keep tracks of the argkmin.
for i in range(n_X):
for j in range(n_Y):
sqeuclidean_dist_i_j = (
self.X_norm_squared[i + X_start] +
dist_middle_terms[i * n_Y + j] +
self.Y_norm_squared[j + Y_start]
)
# Catastrophic cancellation might cause -0. to be present,
# e.g. when computing d(x_i, y_i) when X is Y.
sqeuclidean_dist_i_j = max(0., sqeuclidean_dist_i_j)
heap_push(
values=heaps_r_distances + i * self.k,
indices=heaps_indices + i * self.k,
size=self.k,
val=sqeuclidean_dist_i_j,
val_idx=j + Y_start,
)
{{endfor}}
@@ -0,0 +1,182 @@
from cython cimport floating, integral
from cython.parallel cimport parallel, prange
from libcpp.map cimport map as cpp_map, pair as cpp_pair
from libc.stdlib cimport free
from ...utils._typedefs cimport intp_t, float64_t
from ...utils.parallel import _get_threadpool_controller
import numpy as np
from scipy.sparse import issparse
from ._classmode cimport WeightingStrategy
{{for name_suffix in ["32", "64"]}}
from ._argkmin cimport ArgKmin{{name_suffix}}
from ._datasets_pair cimport DatasetsPair{{name_suffix}}
cdef class ArgKminClassMode{{name_suffix}}(ArgKmin{{name_suffix}}):
"""
{{name_suffix}}bit implementation of ArgKminClassMode.
"""
cdef:
const intp_t[:] Y_labels,
const intp_t[:] unique_Y_labels
float64_t[:, :] class_scores
cpp_map[intp_t, intp_t] labels_to_index
WeightingStrategy weight_type
@classmethod
def compute(
cls,
X,
Y,
intp_t k,
weights,
Y_labels,
unique_Y_labels,
str metric="euclidean",
chunk_size=None,
dict metric_kwargs=None,
str strategy=None,
):
"""Compute the argkmin reduction with Y_labels.
This classmethod is responsible for introspecting the arguments
values to dispatch to the most appropriate implementation of
:class:`ArgKminClassMode{{name_suffix}}`.
This allows decoupling the API entirely from the implementation details
whilst maintaining RAII: all temporarily allocated datastructures necessary
for the concrete implementation are therefore freed when this classmethod
returns.
No instance _must_ directly be created outside of this class method.
"""
# Use a generic implementation that handles most scipy
# metrics by computing the distances between 2 vectors at a time.
pda = ArgKminClassMode{{name_suffix}}(
datasets_pair=DatasetsPair{{name_suffix}}.get_for(X, Y, metric, metric_kwargs),
k=k,
chunk_size=chunk_size,
strategy=strategy,
weights=weights,
Y_labels=Y_labels,
unique_Y_labels=unique_Y_labels,
)
# Limit the number of threads in second level of nested parallelism for BLAS
# to avoid threads over-subscription (in GEMM for instance).
with _get_threadpool_controller().limit(limits=1, user_api="blas"):
if pda.execute_in_parallel_on_Y:
pda._parallel_on_Y()
else:
pda._parallel_on_X()
return pda._finalize_results()
def __init__(
self,
DatasetsPair{{name_suffix}} datasets_pair,
const intp_t[:] Y_labels,
const intp_t[:] unique_Y_labels,
chunk_size=None,
strategy=None,
intp_t k=1,
weights=None,
):
super().__init__(
datasets_pair=datasets_pair,
chunk_size=chunk_size,
strategy=strategy,
k=k,
)
if weights == "uniform":
self.weight_type = WeightingStrategy.uniform
elif weights == "distance":
self.weight_type = WeightingStrategy.distance
else:
self.weight_type = WeightingStrategy.callable
self.Y_labels = Y_labels
self.unique_Y_labels = unique_Y_labels
cdef intp_t idx, neighbor_class_idx
# Map from set of unique labels to their indices in `class_scores`
# Buffer used in building a histogram for one-pass weighted mode
self.class_scores = np.zeros(
(self.n_samples_X, unique_Y_labels.shape[0]), dtype=np.float64,
)
def _finalize_results(self):
probabilities = np.asarray(self.class_scores)
probabilities /= probabilities.sum(axis=1, keepdims=True)
return probabilities
cdef inline void weighted_histogram_mode(
self,
intp_t sample_index,
intp_t* indices,
float64_t* distances,
) noexcept nogil:
cdef:
intp_t neighbor_idx, neighbor_class_idx, label_index, multi_output_index
float64_t score_incr = 1
# TODO: Implement other WeightingStrategy values
bint use_distance_weighting = (
self.weight_type == WeightingStrategy.distance
)
# Iterate through the sample k-nearest neighbours
for neighbor_rank in range(self.k):
# Absolute indice of the neighbor_rank-th Nearest Neighbors
# in range [0, n_samples_Y)
# TODO: inspect if it worth permuting this condition
# and the for-loop above for improved branching.
if use_distance_weighting:
score_incr = 1 / distances[neighbor_rank]
neighbor_idx = indices[neighbor_rank]
neighbor_class_idx = self.Y_labels[neighbor_idx]
self.class_scores[sample_index][neighbor_class_idx] += score_incr
return
cdef void _parallel_on_X_prange_iter_finalize(
self,
intp_t thread_num,
intp_t X_start,
intp_t X_end,
) noexcept nogil:
cdef:
intp_t idx, sample_index
for idx in range(X_end - X_start):
# One-pass top-one weighted mode
# Compute the absolute index in [0, n_samples_X)
sample_index = X_start + idx
self.weighted_histogram_mode(
sample_index,
&self.heaps_indices_chunks[thread_num][idx * self.k],
&self.heaps_r_distances_chunks[thread_num][idx * self.k],
)
return
cdef void _parallel_on_Y_finalize(
self,
) noexcept nogil:
cdef:
intp_t sample_index, thread_num
with nogil, parallel(num_threads=self.chunks_n_threads):
# Deallocating temporary datastructures
for thread_num in prange(self.chunks_n_threads, schedule='static'):
free(self.heaps_r_distances_chunks[thread_num])
free(self.heaps_indices_chunks[thread_num])
for sample_index in prange(self.n_samples_X, schedule='static'):
self.weighted_histogram_mode(
sample_index,
&self.argkmin_indices[sample_index][0],
&self.argkmin_distances[sample_index][0],
)
return
{{endfor}}
@@ -0,0 +1,135 @@
from cython cimport final
from ...utils._typedefs cimport intp_t, float64_t
{{for name_suffix in ['64', '32']}}
from ._datasets_pair cimport DatasetsPair{{name_suffix}}
cpdef float64_t[::1] _sqeuclidean_row_norms{{name_suffix}}(
X,
intp_t num_threads,
)
cdef class BaseDistancesReduction{{name_suffix}}:
"""
Base float{{name_suffix}} implementation template of the pairwise-distances
reduction backends.
Implementations inherit from this template and may override the several
defined hooks as needed in order to easily extend functionality with
minimal redundant code.
"""
cdef:
readonly DatasetsPair{{name_suffix}} datasets_pair
# The number of threads that can be used is stored in effective_n_threads.
#
# The number of threads to use in the parallelization strategy
# (i.e. parallel_on_X or parallel_on_Y) can be smaller than effective_n_threads:
# for small datasets, fewer threads might be needed to loop over pair of chunks.
#
# Hence, the number of threads that _will_ be used for looping over chunks
# is stored in chunks_n_threads, allowing solely using what we need.
#
# Thus, an invariant is:
#
# chunks_n_threads <= effective_n_threads
#
intp_t effective_n_threads
intp_t chunks_n_threads
intp_t n_samples_chunk, chunk_size
intp_t n_samples_X, X_n_samples_chunk, X_n_chunks, X_n_samples_last_chunk
intp_t n_samples_Y, Y_n_samples_chunk, Y_n_chunks, Y_n_samples_last_chunk
bint execute_in_parallel_on_Y
@final
cdef void _parallel_on_X(self) noexcept nogil
@final
cdef void _parallel_on_Y(self) noexcept nogil
# Placeholder methods which have to be implemented
cdef void _compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil
# Placeholder methods which can be implemented
cdef void compute_exact_distances(self) noexcept nogil
cdef void _parallel_on_X_parallel_init(
self,
intp_t thread_num,
) noexcept nogil
cdef void _parallel_on_X_init_chunk(
self,
intp_t thread_num,
intp_t X_start,
intp_t X_end,
) noexcept nogil
cdef void _parallel_on_X_pre_compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil
cdef void _parallel_on_X_prange_iter_finalize(
self,
intp_t thread_num,
intp_t X_start,
intp_t X_end,
) noexcept nogil
cdef void _parallel_on_X_parallel_finalize(
self,
intp_t thread_num
) noexcept nogil
cdef void _parallel_on_Y_init(
self,
) noexcept nogil
cdef void _parallel_on_Y_parallel_init(
self,
intp_t thread_num,
intp_t X_start,
intp_t X_end,
) noexcept nogil
cdef void _parallel_on_Y_pre_compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil
cdef void _parallel_on_Y_synchronize(
self,
intp_t X_start,
intp_t X_end,
) noexcept nogil
cdef void _parallel_on_Y_finalize(
self,
) noexcept nogil
{{endfor}}
@@ -0,0 +1,504 @@
from cython cimport final
from cython.operator cimport dereference as deref
from cython.parallel cimport parallel, prange
from libcpp.vector cimport vector
from ...utils._cython_blas cimport _dot
from ...utils._openmp_helpers cimport omp_get_thread_num
from ...utils._typedefs cimport intp_t, float32_t, float64_t, int32_t
import numpy as np
from scipy.sparse import issparse
from numbers import Integral
from sklearn import get_config
from sklearn.utils import check_scalar
from ...utils._openmp_helpers import _openmp_effective_n_threads
#####################
cdef float64_t[::1] _sqeuclidean_row_norms64_dense(
const float64_t[:, ::1] X,
intp_t num_threads,
):
"""Compute the squared euclidean norm of the rows of X in parallel.
This is faster than using np.einsum("ij, ij->i") even when using a single thread.
"""
cdef:
# Casting for X to remove the const qualifier is needed because APIs
# exposed via scipy.linalg.cython_blas aren't reflecting the arguments'
# const qualifier.
# See: https://github.com/scipy/scipy/issues/14262
float64_t * X_ptr = <float64_t *> &X[0, 0]
intp_t idx = 0
intp_t n = X.shape[0]
intp_t d = X.shape[1]
float64_t[::1] squared_row_norms = np.empty(n, dtype=np.float64)
for idx in prange(n, schedule='static', nogil=True, num_threads=num_threads):
squared_row_norms[idx] = _dot(d, X_ptr + idx * d, 1, X_ptr + idx * d, 1)
return squared_row_norms
cdef float64_t[::1] _sqeuclidean_row_norms32_dense(
const float32_t[:, ::1] X,
intp_t num_threads,
):
"""Compute the squared euclidean norm of the rows of X in parallel.
This is faster than using np.einsum("ij, ij->i") even when using a single thread.
"""
cdef:
# Casting for X to remove the const qualifier is needed because APIs
# exposed via scipy.linalg.cython_blas aren't reflecting the arguments'
# const qualifier.
# See: https://github.com/scipy/scipy/issues/14262
float32_t * X_ptr = <float32_t *> &X[0, 0]
intp_t i = 0, j = 0
intp_t thread_num
intp_t n = X.shape[0]
intp_t d = X.shape[1]
float64_t[::1] squared_row_norms = np.empty(n, dtype=np.float64)
# To upcast the i-th row of X from float32 to float64
vector[vector[float64_t]] X_i_upcast = vector[vector[float64_t]](
num_threads, vector[float64_t](d)
)
with nogil, parallel(num_threads=num_threads):
thread_num = omp_get_thread_num()
for i in prange(n, schedule='static'):
# Upcasting the i-th row of X from float32 to float64
for j in range(d):
X_i_upcast[thread_num][j] = <float64_t> deref(X_ptr + i * d + j)
squared_row_norms[i] = _dot(
d, X_i_upcast[thread_num].data(), 1,
X_i_upcast[thread_num].data(), 1,
)
return squared_row_norms
cdef float64_t[::1] _sqeuclidean_row_norms64_sparse(
const float64_t[:] X_data,
const int32_t[:] X_indptr,
intp_t num_threads,
):
cdef:
intp_t n = X_indptr.shape[0] - 1
int32_t X_i_ptr, idx = 0
float64_t[::1] squared_row_norms = np.zeros(n, dtype=np.float64)
for idx in prange(n, schedule='static', nogil=True, num_threads=num_threads):
for X_i_ptr in range(X_indptr[idx], X_indptr[idx+1]):
squared_row_norms[idx] += X_data[X_i_ptr] * X_data[X_i_ptr]
return squared_row_norms
{{for name_suffix in ["64", "32"]}}
from ._datasets_pair cimport DatasetsPair{{name_suffix}}
cpdef float64_t[::1] _sqeuclidean_row_norms{{name_suffix}}(
X,
intp_t num_threads,
):
if issparse(X):
# TODO: remove this instruction which is a cast in the float32 case
# by moving squared row norms computations in MiddleTermComputer.
X_data = np.asarray(X.data, dtype=np.float64)
X_indptr = np.asarray(X.indptr, dtype=np.int32)
return _sqeuclidean_row_norms64_sparse(X_data, X_indptr, num_threads)
else:
return _sqeuclidean_row_norms{{name_suffix}}_dense(X, num_threads)
cdef class BaseDistancesReduction{{name_suffix}}:
"""
Base float{{name_suffix}} implementation template of the pairwise-distances
reduction backends.
Implementations inherit from this template and may override the several
defined hooks as needed in order to easily extend functionality with
minimal redundant code.
"""
def __init__(
self,
DatasetsPair{{name_suffix}} datasets_pair,
chunk_size=None,
strategy=None,
):
cdef:
intp_t X_n_full_chunks, Y_n_full_chunks
if chunk_size is None:
chunk_size = get_config().get("pairwise_dist_chunk_size", 256)
self.chunk_size = check_scalar(chunk_size, "chunk_size", Integral, min_val=20)
self.effective_n_threads = _openmp_effective_n_threads()
self.datasets_pair = datasets_pair
self.n_samples_X = datasets_pair.n_samples_X()
self.X_n_samples_chunk = min(self.n_samples_X, self.chunk_size)
X_n_full_chunks = self.n_samples_X // self.X_n_samples_chunk
X_n_samples_remainder = self.n_samples_X % self.X_n_samples_chunk
self.X_n_chunks = X_n_full_chunks + (X_n_samples_remainder != 0)
if X_n_samples_remainder != 0:
self.X_n_samples_last_chunk = X_n_samples_remainder
else:
self.X_n_samples_last_chunk = self.X_n_samples_chunk
self.n_samples_Y = datasets_pair.n_samples_Y()
self.Y_n_samples_chunk = min(self.n_samples_Y, self.chunk_size)
Y_n_full_chunks = self.n_samples_Y // self.Y_n_samples_chunk
Y_n_samples_remainder = self.n_samples_Y % self.Y_n_samples_chunk
self.Y_n_chunks = Y_n_full_chunks + (Y_n_samples_remainder != 0)
if Y_n_samples_remainder != 0:
self.Y_n_samples_last_chunk = Y_n_samples_remainder
else:
self.Y_n_samples_last_chunk = self.Y_n_samples_chunk
if strategy is None:
strategy = get_config().get("pairwise_dist_parallel_strategy", 'auto')
if strategy not in ('parallel_on_X', 'parallel_on_Y', 'auto'):
raise RuntimeError(f"strategy must be 'parallel_on_X, 'parallel_on_Y', "
f"or 'auto', but currently strategy='{self.strategy}'.")
if strategy == 'auto':
# This is a simple heuristic whose constant for the
# comparison has been chosen based on experiments.
# parallel_on_X has less synchronization overhead than
# parallel_on_Y and should therefore be used whenever
# n_samples_X is large enough to not starve any of the
# available hardware threads.
if self.n_samples_Y < self.n_samples_X:
# No point to even consider parallelizing on Y in this case. This
# is in particular important to do this on machines with a large
# number of hardware threads.
strategy = 'parallel_on_X'
elif 4 * self.chunk_size * self.effective_n_threads < self.n_samples_X:
# If Y is larger than X, but X is still large enough to allow for
# parallelism, we might still want to favor parallelizing on X.
strategy = 'parallel_on_X'
else:
strategy = 'parallel_on_Y'
self.execute_in_parallel_on_Y = strategy == "parallel_on_Y"
# Not using less, not using more.
self.chunks_n_threads = min(
self.Y_n_chunks if self.execute_in_parallel_on_Y else self.X_n_chunks,
self.effective_n_threads,
)
@final
cdef void _parallel_on_X(self) noexcept nogil:
"""Perform computation and reduction in parallel on chunks of X.
This strategy dispatches tasks statically on threads. Each task
processes exactly only one chunk of X, computing and reducing
distances matrices between vectors of this chunk and vectors of all
chunks of Y, one chunk of Y at a time.
This strategy is embarrassingly parallel with no intermediate data
structures synchronization at all.
Private datastructures are modified internally by threads.
Private template methods can be implemented on subclasses to
interact with those datastructures at various stages.
"""
cdef:
intp_t Y_start, Y_end, X_start, X_end, X_chunk_idx, Y_chunk_idx
intp_t thread_num
with nogil, parallel(num_threads=self.chunks_n_threads):
thread_num = omp_get_thread_num()
# Allocating thread datastructures
self._parallel_on_X_parallel_init(thread_num)
for X_chunk_idx in prange(self.X_n_chunks, schedule='static'):
X_start = X_chunk_idx * self.X_n_samples_chunk
if X_chunk_idx == self.X_n_chunks - 1:
X_end = X_start + self.X_n_samples_last_chunk
else:
X_end = X_start + self.X_n_samples_chunk
# Reinitializing thread datastructures for the new X chunk
self._parallel_on_X_init_chunk(thread_num, X_start, X_end)
for Y_chunk_idx in range(self.Y_n_chunks):
Y_start = Y_chunk_idx * self.Y_n_samples_chunk
if Y_chunk_idx == self.Y_n_chunks - 1:
Y_end = Y_start + self.Y_n_samples_last_chunk
else:
Y_end = Y_start + self.Y_n_samples_chunk
self._parallel_on_X_pre_compute_and_reduce_distances_on_chunks(
X_start, X_end,
Y_start, Y_end,
thread_num,
)
self._compute_and_reduce_distances_on_chunks(
X_start, X_end,
Y_start, Y_end,
thread_num,
)
# Adjusting thread datastructures on the full pass on Y
self._parallel_on_X_prange_iter_finalize(thread_num, X_start, X_end)
# end: for X_chunk_idx
# Deallocating thread datastructures
self._parallel_on_X_parallel_finalize(thread_num)
# end: with nogil, parallel
return
@final
cdef void _parallel_on_Y(self) noexcept nogil:
"""Perform computation and reduction in parallel on chunks of Y.
This strategy is a sequence of embarrassingly parallel subtasks:
chunks of X are iterated over sequentially, and for each chunk of X,
tasks are dispatched statically on threads. Each task processes one
and only one chunk of Y, computing and reducing distances matrices
between vectors of the chunk of X and vectors of the Y.
It comes with lock-free and parallelized intermediate data structures
that synchronize at each iteration of the sequential outer loop on X
chunks.
Private datastructures are modified internally by threads.
Private template methods can be implemented on subclasses to
interact with those datastructures at various stages.
"""
cdef:
intp_t Y_start, Y_end, X_start, X_end, X_chunk_idx, Y_chunk_idx
intp_t thread_num
# Allocating datastructures shared by all threads
self._parallel_on_Y_init()
for X_chunk_idx in range(self.X_n_chunks):
X_start = X_chunk_idx * self.X_n_samples_chunk
if X_chunk_idx == self.X_n_chunks - 1:
X_end = X_start + self.X_n_samples_last_chunk
else:
X_end = X_start + self.X_n_samples_chunk
with nogil, parallel(num_threads=self.chunks_n_threads):
thread_num = omp_get_thread_num()
# Initializing datastructures used in this thread
self._parallel_on_Y_parallel_init(thread_num, X_start, X_end)
for Y_chunk_idx in prange(self.Y_n_chunks, schedule='static'):
Y_start = Y_chunk_idx * self.Y_n_samples_chunk
if Y_chunk_idx == self.Y_n_chunks - 1:
Y_end = Y_start + self.Y_n_samples_last_chunk
else:
Y_end = Y_start + self.Y_n_samples_chunk
self._parallel_on_Y_pre_compute_and_reduce_distances_on_chunks(
X_start, X_end,
Y_start, Y_end,
thread_num,
)
self._compute_and_reduce_distances_on_chunks(
X_start, X_end,
Y_start, Y_end,
thread_num,
)
# end: prange
# end: with nogil, parallel
# Synchronizing the thread datastructures with the main ones
self._parallel_on_Y_synchronize(X_start, X_end)
# end: for X_chunk_idx
# Deallocating temporary datastructures and adjusting main datastructures
self._parallel_on_Y_finalize()
return
# Placeholder methods which have to be implemented
cdef void _compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil:
"""Compute the pairwise distances on two chunks of X and Y and reduce them.
This is THE core computational method of BaseDistancesReduction{{name_suffix}}.
This must be implemented in subclasses agnostically from the parallelization
strategies.
"""
return
def _finalize_results(self, bint return_distance):
"""Callback adapting datastructures before returning results.
This must be implemented in subclasses.
"""
return None
# Placeholder methods which can be implemented
cdef void compute_exact_distances(self) noexcept nogil:
"""Convert rank-preserving distances to exact distances or recompute them."""
return
cdef void _parallel_on_X_parallel_init(
self,
intp_t thread_num,
) noexcept nogil:
"""Allocate datastructures used in a thread given its number."""
return
cdef void _parallel_on_X_init_chunk(
self,
intp_t thread_num,
intp_t X_start,
intp_t X_end,
) noexcept nogil:
"""Initialize datastructures used in a thread given its number.
In this method, EuclideanDistance specialisations of subclass of
BaseDistancesReduction _must_ call:
self.middle_term_computer._parallel_on_X_init_chunk(
thread_num, X_start, X_end,
)
to ensure the proper upcast of X[X_start:X_end] to float64 prior
to the reduction with float64 accumulator buffers when X.dtype is
float32.
"""
return
cdef void _parallel_on_X_pre_compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil:
"""Initialize datastructures just before the _compute_and_reduce_distances_on_chunks.
In this method, EuclideanDistance specialisations of subclass of
BaseDistancesReduction _must_ call:
self.middle_term_computer._parallel_on_X_pre_compute_and_reduce_distances_on_chunks(
X_start, X_end, Y_start, Y_end, thread_num,
)
to ensure the proper upcast of Y[Y_start:Y_end] to float64 prior
to the reduction with float64 accumulator buffers when Y.dtype is
float32.
"""
return
cdef void _parallel_on_X_prange_iter_finalize(
self,
intp_t thread_num,
intp_t X_start,
intp_t X_end,
) noexcept nogil:
"""Interact with datastructures after a reduction on chunks."""
return
cdef void _parallel_on_X_parallel_finalize(
self,
intp_t thread_num
) noexcept nogil:
"""Interact with datastructures after executing all the reductions."""
return
cdef void _parallel_on_Y_init(
self,
) noexcept nogil:
"""Allocate datastructures used in all threads."""
return
cdef void _parallel_on_Y_parallel_init(
self,
intp_t thread_num,
intp_t X_start,
intp_t X_end,
) noexcept nogil:
"""Initialize datastructures used in a thread given its number.
In this method, EuclideanDistance specialisations of subclass of
BaseDistancesReduction _must_ call:
self.middle_term_computer._parallel_on_Y_parallel_init(
thread_num, X_start, X_end,
)
to ensure the proper upcast of X[X_start:X_end] to float64 prior
to the reduction with float64 accumulator buffers when X.dtype is
float32.
"""
return
cdef void _parallel_on_Y_pre_compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil:
"""Initialize datastructures just before the _compute_and_reduce_distances_on_chunks.
In this method, EuclideanDistance specialisations of subclass of
BaseDistancesReduction _must_ call:
self.middle_term_computer._parallel_on_Y_pre_compute_and_reduce_distances_on_chunks(
X_start, X_end, Y_start, Y_end, thread_num,
)
to ensure the proper upcast of Y[Y_start:Y_end] to float64 prior
to the reduction with float64 accumulator buffers when Y.dtype is
float32.
"""
return
cdef void _parallel_on_Y_synchronize(
self,
intp_t X_start,
intp_t X_end,
) noexcept nogil:
"""Update thread datastructures before leaving a parallel region."""
return
cdef void _parallel_on_Y_finalize(
self,
) noexcept nogil:
"""Update datastructures after executing all the reductions."""
return
{{endfor}}
@@ -0,0 +1,5 @@
cpdef enum WeightingStrategy:
uniform = 0
# TODO: Implement the following options in weighted_histogram_mode
distance = 1
callable = 2
@@ -0,0 +1,67 @@
{{py:
implementation_specific_values = [
# Values are the following ones:
#
# name_suffix, INPUT_DTYPE_t, INPUT_DTYPE
('64', 'DistanceMetric64', 'float64_t'),
('32', 'DistanceMetric32', 'float32_t')
]
}}
from ...utils._typedefs cimport float64_t, float32_t, int32_t, intp_t
from ...metrics._dist_metrics cimport DistanceMetric64, DistanceMetric32, DistanceMetric
{{for name_suffix, DistanceMetric, INPUT_DTYPE_t in implementation_specific_values}}
cdef class DatasetsPair{{name_suffix}}:
cdef:
{{DistanceMetric}} distance_metric
intp_t n_features
cdef intp_t n_samples_X(self) noexcept nogil
cdef intp_t n_samples_Y(self) noexcept nogil
cdef float64_t dist(self, intp_t i, intp_t j) noexcept nogil
cdef float64_t surrogate_dist(self, intp_t i, intp_t j) noexcept nogil
cdef class DenseDenseDatasetsPair{{name_suffix}}(DatasetsPair{{name_suffix}}):
cdef:
const {{INPUT_DTYPE_t}}[:, ::1] X
const {{INPUT_DTYPE_t}}[:, ::1] Y
cdef class SparseSparseDatasetsPair{{name_suffix}}(DatasetsPair{{name_suffix}}):
cdef:
const {{INPUT_DTYPE_t}}[:] X_data
const int32_t[::1] X_indices
const int32_t[::1] X_indptr
const {{INPUT_DTYPE_t}}[:] Y_data
const int32_t[::1] Y_indices
const int32_t[::1] Y_indptr
cdef class SparseDenseDatasetsPair{{name_suffix}}(DatasetsPair{{name_suffix}}):
cdef:
const {{INPUT_DTYPE_t}}[:] X_data
const int32_t[::1] X_indices
const int32_t[::1] X_indptr
const {{INPUT_DTYPE_t}}[:] Y_data
const int32_t[::1] Y_indices
intp_t n_Y
cdef class DenseSparseDatasetsPair{{name_suffix}}(DatasetsPair{{name_suffix}}):
cdef:
# As distance metrics are commutative, we can simply rely
# on the implementation of SparseDenseDatasetsPair and
# swap arguments.
DatasetsPair{{name_suffix}} datasets_pair
{{endfor}}
@@ -0,0 +1,406 @@
import copy
{{py:
implementation_specific_values = [
# Values are the following ones:
#
# name_suffix, INPUT_DTYPE_t, INPUT_DTYPE
('64', 'DistanceMetric64', 'float64_t', 'np.float64'),
('32', 'DistanceMetric32', 'float32_t', 'np.float32')
]
}}
import numpy as np
from cython cimport final
from ...utils._typedefs cimport float64_t, float32_t, intp_t
from scipy.sparse import issparse, csr_matrix
{{for name_suffix, DistanceMetric, INPUT_DTYPE_t, INPUT_DTYPE in implementation_specific_values}}
cdef class DatasetsPair{{name_suffix}}:
"""Abstract class which wraps a pair of datasets (X, Y).
This class allows computing distances between a single pair of rows of
of X and Y at a time given the pair of their indices (i, j). This class is
specialized for each metric thanks to the :func:`get_for` factory classmethod.
The handling of parallelization over chunks to compute the distances
and aggregation for several rows at a time is done in dedicated
subclasses of :class:`BaseDistancesReductionDispatcher` that in-turn rely on
subclasses of :class:`DatasetsPair` for each pair of rows in the data. The
goal is to make it possible to decouple the generic parallelization and
aggregation logic from metric-specific computation as much as possible.
X and Y can be stored as C-contiguous np.ndarrays or CSR matrices
in subclasses.
This class avoids the overhead of dispatching distance computations
to :class:`sklearn.metrics.DistanceMetric` based on the physical
representation of the vectors (sparse vs. dense). It makes use of
cython.final to remove the overhead of dispatching method calls.
Parameters
----------
distance_metric: {{DistanceMetric}}
The distance metric responsible for computing distances
between two vectors of (X, Y).
"""
@classmethod
def get_for(
cls,
X,
Y,
metric="euclidean",
dict metric_kwargs=None,
) -> DatasetsPair{{name_suffix}}:
"""Return the DatasetsPair implementation for the given arguments.
Parameters
----------
X : {ndarray, sparse matrix} of shape (n_samples_X, n_features)
Input data.
If provided as a ndarray, it must be C-contiguous.
If provided as a sparse matrix, it must be in CSR format.
Y : {ndarray, sparse matrix} of shape (n_samples_Y, n_features)
Input data.
If provided as a ndarray, it must be C-contiguous.
If provided as a sparse matrix, it must be in CSR format.
metric : str or DistanceMetric object, default='euclidean'
The distance metric to compute between rows of X and Y.
The default metric is a fast implementation of the Euclidean
metric. For a list of available metrics, see the documentation
of :class:`~sklearn.metrics.DistanceMetric`.
metric_kwargs : dict, default=None
Keyword arguments to pass to specified metric function.
Returns
-------
datasets_pair: DatasetsPair{{name_suffix}}
The suited DatasetsPair{{name_suffix}} implementation.
"""
# X_norm_squared and Y_norm_squared might be propagated
# down to DatasetsPairs via metrics_kwargs when the Euclidean
# specialisations can't be used.
# To prevent X_norm_squared and Y_norm_squared to be passed
# down to DistanceMetrics (whose constructors would raise
# a RuntimeError), we pop them here.
if metric_kwargs is not None:
# Copying metric_kwargs not to pop "X_norm_squared"
# and "Y_norm_squared" where they are used
metric_kwargs = copy.copy(metric_kwargs)
metric_kwargs.pop("X_norm_squared", None)
metric_kwargs.pop("Y_norm_squared", None)
cdef:
{{DistanceMetric}} distance_metric = DistanceMetric.get_metric(
metric,
{{INPUT_DTYPE}},
**(metric_kwargs or {})
)
# Metric-specific checks that do not replace nor duplicate `check_array`.
distance_metric._validate_data(X)
distance_metric._validate_data(Y)
X_is_sparse = issparse(X)
Y_is_sparse = issparse(Y)
if not X_is_sparse and not Y_is_sparse:
return DenseDenseDatasetsPair{{name_suffix}}(X, Y, distance_metric)
if X_is_sparse and Y_is_sparse:
return SparseSparseDatasetsPair{{name_suffix}}(X, Y, distance_metric)
if X_is_sparse and not Y_is_sparse:
return SparseDenseDatasetsPair{{name_suffix}}(X, Y, distance_metric)
return DenseSparseDatasetsPair{{name_suffix}}(X, Y, distance_metric)
@classmethod
def unpack_csr_matrix(cls, X: csr_matrix):
"""Ensure that the CSR matrix is indexed with np.int32."""
X_data = np.asarray(X.data, dtype={{INPUT_DTYPE}})
X_indices = np.asarray(X.indices, dtype=np.int32)
X_indptr = np.asarray(X.indptr, dtype=np.int32)
return X_data, X_indices, X_indptr
def __init__(self, {{DistanceMetric}} distance_metric, intp_t n_features):
self.distance_metric = distance_metric
self.n_features = n_features
cdef intp_t n_samples_X(self) noexcept nogil:
"""Number of samples in X."""
# This is a abstract method.
# This _must_ always be overwritten in subclasses.
# TODO: add "with gil: raise" here when supporting Cython 3.0
return -999
cdef intp_t n_samples_Y(self) noexcept nogil:
"""Number of samples in Y."""
# This is a abstract method.
# This _must_ always be overwritten in subclasses.
# TODO: add "with gil: raise" here when supporting Cython 3.0
return -999
cdef float64_t surrogate_dist(self, intp_t i, intp_t j) noexcept nogil:
return self.dist(i, j)
cdef float64_t dist(self, intp_t i, intp_t j) noexcept nogil:
# This is a abstract method.
# This _must_ always be overwritten in subclasses.
# TODO: add "with gil: raise" here when supporting Cython 3.0
return -1
@final
cdef class DenseDenseDatasetsPair{{name_suffix}}(DatasetsPair{{name_suffix}}):
"""Compute distances between row vectors of two arrays.
Parameters
----------
X: ndarray of shape (n_samples_X, n_features)
Rows represent vectors. Must be C-contiguous.
Y: ndarray of shape (n_samples_Y, n_features)
Rows represent vectors. Must be C-contiguous.
distance_metric: DistanceMetric
The distance metric responsible for computing distances
between two row vectors of (X, Y).
"""
def __init__(
self,
const {{INPUT_DTYPE_t}}[:, ::1] X,
const {{INPUT_DTYPE_t}}[:, ::1] Y,
{{DistanceMetric}} distance_metric,
):
super().__init__(distance_metric, n_features=X.shape[1])
# Arrays have already been checked
self.X = X
self.Y = Y
@final
cdef intp_t n_samples_X(self) noexcept nogil:
return self.X.shape[0]
@final
cdef intp_t n_samples_Y(self) noexcept nogil:
return self.Y.shape[0]
@final
cdef float64_t surrogate_dist(self, intp_t i, intp_t j) noexcept nogil:
return self.distance_metric.rdist(&self.X[i, 0], &self.Y[j, 0], self.n_features)
@final
cdef float64_t dist(self, intp_t i, intp_t j) noexcept nogil:
return self.distance_metric.dist(&self.X[i, 0], &self.Y[j, 0], self.n_features)
@final
cdef class SparseSparseDatasetsPair{{name_suffix}}(DatasetsPair{{name_suffix}}):
"""Compute distances between vectors of two CSR matrices.
Parameters
----------
X: sparse matrix of shape (n_samples_X, n_features)
Rows represent vectors. Must be in CSR format.
Y: sparse matrix of shape (n_samples_Y, n_features)
Rows represent vectors. Must be in CSR format.
distance_metric: DistanceMetric
The distance metric responsible for computing distances
between two vectors of (X, Y).
"""
def __init__(self, X, Y, {{DistanceMetric}} distance_metric):
super().__init__(distance_metric, n_features=X.shape[1])
self.X_data, self.X_indices, self.X_indptr = self.unpack_csr_matrix(X)
self.Y_data, self.Y_indices, self.Y_indptr = self.unpack_csr_matrix(Y)
@final
cdef intp_t n_samples_X(self) noexcept nogil:
return self.X_indptr.shape[0] - 1
@final
cdef intp_t n_samples_Y(self) noexcept nogil:
return self.Y_indptr.shape[0] - 1
@final
cdef float64_t surrogate_dist(self, intp_t i, intp_t j) noexcept nogil:
return self.distance_metric.rdist_csr(
x1_data=&self.X_data[0],
x1_indices=&self.X_indices[0],
x2_data=&self.Y_data[0],
x2_indices=&self.Y_indices[0],
x1_start=self.X_indptr[i],
x1_end=self.X_indptr[i + 1],
x2_start=self.Y_indptr[j],
x2_end=self.Y_indptr[j + 1],
size=self.n_features,
)
@final
cdef float64_t dist(self, intp_t i, intp_t j) noexcept nogil:
return self.distance_metric.dist_csr(
x1_data=&self.X_data[0],
x1_indices=&self.X_indices[0],
x2_data=&self.Y_data[0],
x2_indices=&self.Y_indices[0],
x1_start=self.X_indptr[i],
x1_end=self.X_indptr[i + 1],
x2_start=self.Y_indptr[j],
x2_end=self.Y_indptr[j + 1],
size=self.n_features,
)
@final
cdef class SparseDenseDatasetsPair{{name_suffix}}(DatasetsPair{{name_suffix}}):
"""Compute distances between vectors of a CSR matrix and a dense array.
Parameters
----------
X: sparse matrix of shape (n_samples_X, n_features)
Rows represent vectors. Must be in CSR format.
Y: ndarray of shape (n_samples_Y, n_features)
Rows represent vectors. Must be C-contiguous.
distance_metric: DistanceMetric
The distance metric responsible for computing distances
between two vectors of (X, Y).
"""
def __init__(self, X, Y, {{DistanceMetric}} distance_metric):
super().__init__(distance_metric, n_features=X.shape[1])
self.X_data, self.X_indices, self.X_indptr = self.unpack_csr_matrix(X)
# We support the sparse-dense case by using the sparse-sparse interfaces
# of `DistanceMetric` (namely `DistanceMetric.{dist_csr,rdist_csr}`) to
# avoid introducing a new complex set of interfaces. In this case, we
# need to convert `Y` (the dense array) into a CSR matrix.
#
# Here we motive using another simpler CSR representation to use for `Y`.
#
# If we were to use the usual CSR representation for `Y`, storing all
# the columns indices in `indices` would have required allocating an
# array of n_samples × n_features elements with repeated contiguous
# integers from 0 to n_features - 1. This would have been very wasteful
# from a memory point of view. This alternative representation just uses
# the necessary amount of information needed and only necessitates
# shifting the address of `data` before calling the CSR × CSR routines.
#
# In this representation:
#
# - the `data` array is the original dense array, `Y`, whose first
# element's address is shifted before calling the CSR × CSR routine
#
# - the `indices` array is a single row of `n_features` elements:
#
# [0, 1, ..., n_features-1]
#
# - the `indptr` array is not materialised as the indices pointers'
# offset is constant (the offset equals `n_features`). Moreover, as
# `data` is shifted, constant `start` and `end` indices pointers
# respectively equalling 0 and n_features are used.
# Y array already has been checked here
self.n_Y = Y.shape[0]
self.Y_data = np.ravel(Y)
self.Y_indices = np.arange(self.n_features, dtype=np.int32)
@final
cdef intp_t n_samples_X(self) noexcept nogil:
return self.X_indptr.shape[0] - 1
@final
cdef intp_t n_samples_Y(self) noexcept nogil:
return self.n_Y
@final
cdef float64_t surrogate_dist(self, intp_t i, intp_t j) noexcept nogil:
return self.distance_metric.rdist_csr(
x1_data=&self.X_data[0],
x1_indices=&self.X_indices[0],
# Increment the data pointer such that x2_start=0 is aligned with the
# j-th row
x2_data=&self.Y_data[0] + j * self.n_features,
x2_indices=&self.Y_indices[0],
x1_start=self.X_indptr[i],
x1_end=self.X_indptr[i + 1],
x2_start=0,
x2_end=self.n_features,
size=self.n_features,
)
@final
cdef float64_t dist(self, intp_t i, intp_t j) noexcept nogil:
return self.distance_metric.dist_csr(
x1_data=&self.X_data[0],
x1_indices=&self.X_indices[0],
# Increment the data pointer such that x2_start=0 is aligned with the
# j-th row
x2_data=&self.Y_data[0] + j * self.n_features,
x2_indices=&self.Y_indices[0],
x1_start=self.X_indptr[i],
x1_end=self.X_indptr[i + 1],
x2_start=0,
x2_end=self.n_features,
size=self.n_features,
)
@final
cdef class DenseSparseDatasetsPair{{name_suffix}}(DatasetsPair{{name_suffix}}):
"""Compute distances between vectors of a dense array and a CSR matrix.
Parameters
----------
X: ndarray of shape (n_samples_X, n_features)
Rows represent vectors. Must be C-contiguous.
Y: sparse matrix of shape (n_samples_Y, n_features)
Rows represent vectors. Must be in CSR format.
distance_metric: DistanceMetric
The distance metric responsible for computing distances
between two vectors of (X, Y).
"""
def __init__(self, X, Y, {{DistanceMetric}} distance_metric):
super().__init__(distance_metric, n_features=X.shape[1])
# Swapping arguments on the constructor
self.datasets_pair = SparseDenseDatasetsPair{{name_suffix}}(Y, X, distance_metric)
@final
cdef intp_t n_samples_X(self) noexcept nogil:
# Swapping interface
return self.datasets_pair.n_samples_Y()
@final
cdef intp_t n_samples_Y(self) noexcept nogil:
# Swapping interface
return self.datasets_pair.n_samples_X()
@final
cdef float64_t surrogate_dist(self, intp_t i, intp_t j) noexcept nogil:
# Swapping arguments on the same interface
return self.datasets_pair.surrogate_dist(j, i)
@final
cdef float64_t dist(self, intp_t i, intp_t j) noexcept nogil:
# Swapping arguments on the same interface
return self.datasets_pair.dist(j, i)
{{endfor}}
@@ -0,0 +1,228 @@
{{py:
implementation_specific_values = [
# Values are the following ones:
#
# name_suffix, upcast_to_float64, INPUT_DTYPE_t, INPUT_DTYPE
#
# We also use the float64 dtype and C-type names as defined in
# `sklearn.utils._typedefs` to maintain consistency.
#
('64', False, 'float64_t', 'np.float64'),
('32', True, 'float32_t', 'np.float32')
]
}}
from libcpp.vector cimport vector
from ...utils._typedefs cimport float64_t, float32_t, int32_t, intp_t
cdef void _middle_term_sparse_sparse_64(
const float64_t[:] X_data,
const int32_t[:] X_indices,
const int32_t[:] X_indptr,
intp_t X_start,
intp_t X_end,
const float64_t[:] Y_data,
const int32_t[:] Y_indices,
const int32_t[:] Y_indptr,
intp_t Y_start,
intp_t Y_end,
float64_t * D,
) noexcept nogil
{{for name_suffix, upcast_to_float64, INPUT_DTYPE_t, INPUT_DTYPE in implementation_specific_values}}
cdef class MiddleTermComputer{{name_suffix}}:
cdef:
intp_t effective_n_threads
intp_t chunks_n_threads
intp_t dist_middle_terms_chunks_size
intp_t n_features
intp_t chunk_size
# Buffers for the `-2 * X_c @ Y_c.T` term computed via GEMM
vector[vector[float64_t]] dist_middle_terms_chunks
cdef void _parallel_on_X_pre_compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil
cdef void _parallel_on_X_parallel_init(self, intp_t thread_num) noexcept nogil
cdef void _parallel_on_X_init_chunk(
self,
intp_t thread_num,
intp_t X_start,
intp_t X_end,
) noexcept nogil
cdef void _parallel_on_Y_init(self) noexcept nogil
cdef void _parallel_on_Y_parallel_init(
self,
intp_t thread_num,
intp_t X_start,
intp_t X_end,
) noexcept nogil
cdef void _parallel_on_Y_pre_compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num
) noexcept nogil
cdef float64_t * _compute_dist_middle_terms(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil
cdef class DenseDenseMiddleTermComputer{{name_suffix}}(MiddleTermComputer{{name_suffix}}):
cdef:
const {{INPUT_DTYPE_t}}[:, ::1] X
const {{INPUT_DTYPE_t}}[:, ::1] Y
{{if upcast_to_float64}}
# Buffers for upcasting chunks of X and Y from 32bit to 64bit
vector[vector[float64_t]] X_c_upcast
vector[vector[float64_t]] Y_c_upcast
{{endif}}
cdef void _parallel_on_X_pre_compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil
cdef void _parallel_on_X_init_chunk(
self,
intp_t thread_num,
intp_t X_start,
intp_t X_end,
) noexcept nogil
cdef void _parallel_on_Y_parallel_init(
self,
intp_t thread_num,
intp_t X_start,
intp_t X_end,
) noexcept nogil
cdef void _parallel_on_Y_pre_compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num
) noexcept nogil
cdef float64_t * _compute_dist_middle_terms(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil
cdef class SparseSparseMiddleTermComputer{{name_suffix}}(MiddleTermComputer{{name_suffix}}):
cdef:
const float64_t[:] X_data
const int32_t[:] X_indices
const int32_t[:] X_indptr
const float64_t[:] Y_data
const int32_t[:] Y_indices
const int32_t[:] Y_indptr
cdef void _parallel_on_X_pre_compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num
) noexcept nogil
cdef void _parallel_on_Y_pre_compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num
) noexcept nogil
cdef float64_t * _compute_dist_middle_terms(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil
cdef class SparseDenseMiddleTermComputer{{name_suffix}}(MiddleTermComputer{{name_suffix}}):
cdef:
const float64_t[:] X_data
const int32_t[:] X_indices
const int32_t[:] X_indptr
const {{INPUT_DTYPE_t}}[:, ::1] Y
# We treat the dense-sparse case with the sparse-dense case by simply
# treating the dist_middle_terms as F-ordered and by swapping arguments.
# This attribute is meant to encode the case and adapt the logic
# accordingly.
bint c_ordered_middle_term
cdef void _parallel_on_X_pre_compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num
) noexcept nogil
cdef void _parallel_on_Y_pre_compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num
) noexcept nogil
cdef float64_t * _compute_dist_middle_terms(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil
{{endfor}}
@@ -0,0 +1,633 @@
{{py:
implementation_specific_values = [
# Values are the following ones:
#
# name_suffix, upcast_to_float64, INPUT_DTYPE_t, INPUT_DTYPE
#
# We also use the float64 dtype and C-type names as defined in
# `sklearn.utils._typedefs` to maintain consistency.
#
('64', False, 'float64_t', 'np.float64'),
('32', True, 'float32_t', 'np.float32')
]
}}
from libcpp.vector cimport vector
from libcpp.algorithm cimport fill
from ...utils._cython_blas cimport (
BLAS_Order,
BLAS_Trans,
NoTrans,
RowMajor,
Trans,
_gemm,
)
from ...utils._typedefs cimport float64_t, float32_t, int32_t, intp_t
import numpy as np
from scipy.sparse import issparse, csr_matrix
cdef void _middle_term_sparse_sparse_64(
const float64_t[:] X_data,
const int32_t[:] X_indices,
const int32_t[:] X_indptr,
intp_t X_start,
intp_t X_end,
const float64_t[:] Y_data,
const int32_t[:] Y_indices,
const int32_t[:] Y_indptr,
intp_t Y_start,
intp_t Y_end,
float64_t * D,
) noexcept nogil:
# This routine assumes that D points to the first element of a
# zeroed buffer of length at least equal to n_X × n_Y, conceptually
# representing a 2-d C-ordered array.
cdef:
intp_t i, j, k
intp_t n_X = X_end - X_start
intp_t n_Y = Y_end - Y_start
intp_t x_col, x_ptr, y_col, y_ptr
for i in range(n_X):
for x_ptr in range(X_indptr[X_start+i], X_indptr[X_start+i+1]):
x_col = X_indices[x_ptr]
for j in range(n_Y):
k = i * n_Y + j
for y_ptr in range(Y_indptr[Y_start+j], Y_indptr[Y_start+j+1]):
y_col = Y_indices[y_ptr]
if x_col == y_col:
D[k] += -2 * X_data[x_ptr] * Y_data[y_ptr]
{{for name_suffix, upcast_to_float64, INPUT_DTYPE_t, INPUT_DTYPE in implementation_specific_values}}
cdef void _middle_term_sparse_dense_{{name_suffix}}(
const float64_t[:] X_data,
const int32_t[:] X_indices,
const int32_t[:] X_indptr,
intp_t X_start,
intp_t X_end,
const {{INPUT_DTYPE_t}}[:, ::1] Y,
intp_t Y_start,
intp_t Y_end,
bint c_ordered_middle_term,
float64_t * dist_middle_terms,
) noexcept nogil:
# This routine assumes that dist_middle_terms is a pointer to the first element
# of a buffer filled with zeros of length at least equal to n_X × n_Y, conceptually
# representing a 2-d C-ordered of F-ordered array.
cdef:
intp_t i, j, k
intp_t n_X = X_end - X_start
intp_t n_Y = Y_end - Y_start
intp_t X_i_col_idx, X_i_ptr, Y_j_col_idx, Y_j_ptr
for i in range(n_X):
for j in range(n_Y):
k = i * n_Y + j if c_ordered_middle_term else j * n_X + i
for X_i_ptr in range(X_indptr[X_start+i], X_indptr[X_start+i+1]):
X_i_col_idx = X_indices[X_i_ptr]
dist_middle_terms[k] += -2 * X_data[X_i_ptr] * Y[Y_start + j, X_i_col_idx]
cdef class MiddleTermComputer{{name_suffix}}:
"""Helper class to compute a Euclidean distance matrix in chunks.
This is an abstract base class that is further specialized depending
on the type of data (dense or sparse).
`EuclideanDistance` subclasses relies on the squared Euclidean
distances between chunks of vectors X_c and Y_c using the
following decomposition for the (i,j) pair :
||X_c_i - Y_c_j||² = ||X_c_i||² - 2 X_c_i.Y_c_j^T + ||Y_c_j||²
This helper class is in charge of wrapping the common logic to compute
the middle term, i.e. `- 2 X_c_i.Y_c_j^T`.
"""
@classmethod
def get_for(
cls,
X,
Y,
effective_n_threads,
chunks_n_threads,
dist_middle_terms_chunks_size,
n_features,
chunk_size,
) -> MiddleTermComputer{{name_suffix}}:
"""Return the MiddleTermComputer implementation for the given arguments.
Parameters
----------
X : ndarray or CSR sparse matrix of shape (n_samples_X, n_features)
Input data.
If provided as a ndarray, it must be C-contiguous.
Y : ndarray or CSR sparse matrix of shape (n_samples_Y, n_features)
Input data.
If provided as a ndarray, it must be C-contiguous.
Returns
-------
middle_term_computer: MiddleTermComputer{{name_suffix}}
The suited MiddleTermComputer{{name_suffix}} implementation.
"""
X_is_sparse = issparse(X)
Y_is_sparse = issparse(Y)
if not X_is_sparse and not Y_is_sparse:
return DenseDenseMiddleTermComputer{{name_suffix}}(
X,
Y,
effective_n_threads,
chunks_n_threads,
dist_middle_terms_chunks_size,
n_features,
chunk_size,
)
if X_is_sparse and Y_is_sparse:
return SparseSparseMiddleTermComputer{{name_suffix}}(
X,
Y,
effective_n_threads,
chunks_n_threads,
dist_middle_terms_chunks_size,
n_features,
chunk_size,
)
if X_is_sparse and not Y_is_sparse:
return SparseDenseMiddleTermComputer{{name_suffix}}(
X,
Y,
effective_n_threads,
chunks_n_threads,
dist_middle_terms_chunks_size,
n_features,
chunk_size,
c_ordered_middle_term=True
)
if not X_is_sparse and Y_is_sparse:
# NOTE: The Dense-Sparse case is implement via the Sparse-Dense case.
#
# To do so:
# - X (dense) and Y (sparse) are swapped
# - the distance middle term is seen as F-ordered for consistency
# (c_ordered_middle_term = False)
return SparseDenseMiddleTermComputer{{name_suffix}}(
# Mind that X and Y are swapped here.
Y,
X,
effective_n_threads,
chunks_n_threads,
dist_middle_terms_chunks_size,
n_features,
chunk_size,
c_ordered_middle_term=False,
)
raise NotImplementedError(
"X and Y must be CSR sparse matrices or numpy arrays."
)
@classmethod
def unpack_csr_matrix(cls, X: csr_matrix):
"""Ensure that the CSR matrix is indexed with np.int32."""
X_data = np.asarray(X.data, dtype=np.float64)
X_indices = np.asarray(X.indices, dtype=np.int32)
X_indptr = np.asarray(X.indptr, dtype=np.int32)
return X_data, X_indices, X_indptr
def __init__(
self,
intp_t effective_n_threads,
intp_t chunks_n_threads,
intp_t dist_middle_terms_chunks_size,
intp_t n_features,
intp_t chunk_size,
):
self.effective_n_threads = effective_n_threads
self.chunks_n_threads = chunks_n_threads
self.dist_middle_terms_chunks_size = dist_middle_terms_chunks_size
self.n_features = n_features
self.chunk_size = chunk_size
self.dist_middle_terms_chunks = vector[vector[float64_t]](self.effective_n_threads)
cdef void _parallel_on_X_pre_compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil:
return
cdef void _parallel_on_X_parallel_init(self, intp_t thread_num) noexcept nogil:
self.dist_middle_terms_chunks[thread_num].resize(self.dist_middle_terms_chunks_size)
cdef void _parallel_on_X_init_chunk(
self,
intp_t thread_num,
intp_t X_start,
intp_t X_end,
) noexcept nogil:
return
cdef void _parallel_on_Y_init(self) noexcept nogil:
for thread_num in range(self.chunks_n_threads):
self.dist_middle_terms_chunks[thread_num].resize(
self.dist_middle_terms_chunks_size
)
cdef void _parallel_on_Y_parallel_init(
self,
intp_t thread_num,
intp_t X_start,
intp_t X_end,
) noexcept nogil:
return
cdef void _parallel_on_Y_pre_compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num
) noexcept nogil:
return
cdef float64_t * _compute_dist_middle_terms(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil:
return NULL
cdef class DenseDenseMiddleTermComputer{{name_suffix}}(MiddleTermComputer{{name_suffix}}):
"""Computes the middle term of the Euclidean distance between two chunked dense matrices
X_c and Y_c.
dist_middle_terms = - 2 X_c_i.Y_c_j^T
This class use the BLAS gemm routine to perform the dot product of each chunks
of the distance matrix with improved arithmetic intensity and vector instruction (SIMD).
"""
def __init__(
self,
const {{INPUT_DTYPE_t}}[:, ::1] X,
const {{INPUT_DTYPE_t}}[:, ::1] Y,
intp_t effective_n_threads,
intp_t chunks_n_threads,
intp_t dist_middle_terms_chunks_size,
intp_t n_features,
intp_t chunk_size,
):
super().__init__(
effective_n_threads,
chunks_n_threads,
dist_middle_terms_chunks_size,
n_features,
chunk_size,
)
self.X = X
self.Y = Y
{{if upcast_to_float64}}
# We populate the buffer for upcasting chunks of X and Y from float32 to float64.
self.X_c_upcast = vector[vector[float64_t]](self.effective_n_threads)
self.Y_c_upcast = vector[vector[float64_t]](self.effective_n_threads)
upcast_buffer_n_elements = self.chunk_size * n_features
for thread_num in range(self.effective_n_threads):
self.X_c_upcast[thread_num].resize(upcast_buffer_n_elements)
self.Y_c_upcast[thread_num].resize(upcast_buffer_n_elements)
{{endif}}
cdef void _parallel_on_X_pre_compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil:
{{if upcast_to_float64}}
cdef:
intp_t i, j
intp_t n_chunk_samples = Y_end - Y_start
# Upcasting Y_c=Y[Y_start:Y_end, :] from float32 to float64
for i in range(n_chunk_samples):
for j in range(self.n_features):
self.Y_c_upcast[thread_num][i * self.n_features + j] = <float64_t> self.Y[Y_start + i, j]
{{else}}
return
{{endif}}
cdef void _parallel_on_X_init_chunk(
self,
intp_t thread_num,
intp_t X_start,
intp_t X_end,
) noexcept nogil:
{{if upcast_to_float64}}
cdef:
intp_t i, j
intp_t n_chunk_samples = X_end - X_start
# Upcasting X_c=X[X_start:X_end, :] from float32 to float64
for i in range(n_chunk_samples):
for j in range(self.n_features):
self.X_c_upcast[thread_num][i * self.n_features + j] = <float64_t> self.X[X_start + i, j]
{{else}}
return
{{endif}}
cdef void _parallel_on_Y_parallel_init(
self,
intp_t thread_num,
intp_t X_start,
intp_t X_end,
) noexcept nogil:
{{if upcast_to_float64}}
cdef:
intp_t i, j
intp_t n_chunk_samples = X_end - X_start
# Upcasting X_c=X[X_start:X_end, :] from float32 to float64
for i in range(n_chunk_samples):
for j in range(self.n_features):
self.X_c_upcast[thread_num][i * self.n_features + j] = <float64_t> self.X[X_start + i, j]
{{else}}
return
{{endif}}
cdef void _parallel_on_Y_pre_compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num
) noexcept nogil:
{{if upcast_to_float64}}
cdef:
intp_t i, j
intp_t n_chunk_samples = Y_end - Y_start
# Upcasting Y_c=Y[Y_start:Y_end, :] from float32 to float64
for i in range(n_chunk_samples):
for j in range(self.n_features):
self.Y_c_upcast[thread_num][i * self.n_features + j] = <float64_t> self.Y[Y_start + i, j]
{{else}}
return
{{endif}}
cdef float64_t * _compute_dist_middle_terms(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil:
cdef:
float64_t *dist_middle_terms = self.dist_middle_terms_chunks[thread_num].data()
# Careful: LDA, LDB and LDC are given for F-ordered arrays
# in BLAS documentations, for instance:
# https://www.netlib.org/lapack/explore-html/db/dc9/group__single__blas__level3_gafe51bacb54592ff5de056acabd83c260.html #noqa
#
# Here, we use their counterpart values to work with C-ordered arrays.
BLAS_Order order = RowMajor
BLAS_Trans ta = NoTrans
BLAS_Trans tb = Trans
intp_t m = X_end - X_start
intp_t n = Y_end - Y_start
intp_t K = self.n_features
float64_t alpha = - 2.
{{if upcast_to_float64}}
float64_t * A = self.X_c_upcast[thread_num].data()
float64_t * B = self.Y_c_upcast[thread_num].data()
{{else}}
# Casting for A and B to remove the const is needed because APIs exposed via
# scipy.linalg.cython_blas aren't reflecting the arguments' const qualifier.
# See: https://github.com/scipy/scipy/issues/14262
float64_t * A = <float64_t *> &self.X[X_start, 0]
float64_t * B = <float64_t *> &self.Y[Y_start, 0]
{{endif}}
intp_t lda = self.n_features
intp_t ldb = self.n_features
float64_t beta = 0.
intp_t ldc = Y_end - Y_start
# dist_middle_terms = `-2 * X[X_start:X_end] @ Y[Y_start:Y_end].T`
_gemm(order, ta, tb, m, n, K, alpha, A, lda, B, ldb, beta, dist_middle_terms, ldc)
return dist_middle_terms
cdef class SparseSparseMiddleTermComputer{{name_suffix}}(MiddleTermComputer{{name_suffix}}):
"""Middle term of the Euclidean distance between two chunked CSR matrices.
The result is return as a contiguous array.
dist_middle_terms = - 2 X_c_i.Y_c_j^T
The logic of the computation is wrapped in the routine _middle_term_sparse_sparse_64.
This routine iterates over the data, indices and indptr arrays of the sparse matrices without
densifying them.
"""
def __init__(
self,
X,
Y,
intp_t effective_n_threads,
intp_t chunks_n_threads,
intp_t dist_middle_terms_chunks_size,
intp_t n_features,
intp_t chunk_size,
):
super().__init__(
effective_n_threads,
chunks_n_threads,
dist_middle_terms_chunks_size,
n_features,
chunk_size,
)
self.X_data, self.X_indices, self.X_indptr = self.unpack_csr_matrix(X)
self.Y_data, self.Y_indices, self.Y_indptr = self.unpack_csr_matrix(Y)
cdef void _parallel_on_X_pre_compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil:
# Flush the thread dist_middle_terms_chunks to 0.0
fill(
self.dist_middle_terms_chunks[thread_num].begin(),
self.dist_middle_terms_chunks[thread_num].end(),
0.0,
)
cdef void _parallel_on_Y_pre_compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil:
# Flush the thread dist_middle_terms_chunks to 0.0
fill(
self.dist_middle_terms_chunks[thread_num].begin(),
self.dist_middle_terms_chunks[thread_num].end(),
0.0,
)
cdef float64_t * _compute_dist_middle_terms(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil:
cdef:
float64_t *dist_middle_terms = (
self.dist_middle_terms_chunks[thread_num].data()
)
_middle_term_sparse_sparse_64(
self.X_data,
self.X_indices,
self.X_indptr,
X_start,
X_end,
self.Y_data,
self.Y_indices,
self.Y_indptr,
Y_start,
Y_end,
dist_middle_terms,
)
return dist_middle_terms
cdef class SparseDenseMiddleTermComputer{{name_suffix}}(MiddleTermComputer{{name_suffix}}):
"""Middle term of the Euclidean distance between chunks of a CSR matrix and a np.ndarray.
The logic of the computation is wrapped in the routine _middle_term_sparse_dense_{{name_suffix}}.
This routine iterates over the data, indices and indptr arrays of the sparse matrices
without densifying them.
"""
def __init__(
self,
X,
Y,
intp_t effective_n_threads,
intp_t chunks_n_threads,
intp_t dist_middle_terms_chunks_size,
intp_t n_features,
intp_t chunk_size,
bint c_ordered_middle_term,
):
super().__init__(
effective_n_threads,
chunks_n_threads,
dist_middle_terms_chunks_size,
n_features,
chunk_size,
)
self.X_data, self.X_indices, self.X_indptr = self.unpack_csr_matrix(X)
self.Y = Y
self.c_ordered_middle_term = c_ordered_middle_term
cdef void _parallel_on_X_pre_compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil:
# Fill the thread's dist_middle_terms_chunks with 0.0 before
# computing its elements in _compute_dist_middle_terms.
fill(
self.dist_middle_terms_chunks[thread_num].begin(),
self.dist_middle_terms_chunks[thread_num].end(),
0.0,
)
cdef void _parallel_on_Y_pre_compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil:
# Fill the thread's dist_middle_terms_chunks with 0.0 before
# computing its elements in _compute_dist_middle_terms.
fill(
self.dist_middle_terms_chunks[thread_num].begin(),
self.dist_middle_terms_chunks[thread_num].end(),
0.0,
)
cdef float64_t * _compute_dist_middle_terms(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil:
cdef:
float64_t *dist_middle_terms = (
self.dist_middle_terms_chunks[thread_num].data()
)
# For the dense-sparse case, we use the sparse-dense case
# with dist_middle_terms seen as F-ordered.
# Hence we swap indices pointers here.
if not self.c_ordered_middle_term:
X_start, Y_start = Y_start, X_start
X_end, Y_end = Y_end, X_end
_middle_term_sparse_dense_{{name_suffix}}(
self.X_data,
self.X_indices,
self.X_indptr,
X_start,
X_end,
self.Y,
Y_start,
Y_end,
self.c_ordered_middle_term,
dist_middle_terms,
)
return dist_middle_terms
{{endfor}}
@@ -0,0 +1,90 @@
cimport numpy as cnp
from libcpp.memory cimport shared_ptr
from libcpp.vector cimport vector
from cython cimport final
from ...utils._typedefs cimport intp_t, float64_t
cnp.import_array()
######################
## std::vector to np.ndarray coercion
# As type covariance is not supported for C++ containers via Cython,
# we need to redefine fused types.
ctypedef fused vector_double_intp_t:
vector[intp_t]
vector[float64_t]
ctypedef fused vector_vector_double_intp_t:
vector[vector[intp_t]]
vector[vector[float64_t]]
cdef cnp.ndarray[object, ndim=1] coerce_vectors_to_nd_arrays(
shared_ptr[vector_vector_double_intp_t] vecs
)
#####################
{{for name_suffix in ['64', '32']}}
from ._base cimport BaseDistancesReduction{{name_suffix}}
from ._middle_term_computer cimport MiddleTermComputer{{name_suffix}}
cdef class RadiusNeighbors{{name_suffix}}(BaseDistancesReduction{{name_suffix}}):
"""float{{name_suffix}} implementation of the RadiusNeighbors."""
cdef:
float64_t radius
# DistanceMetric{{name_suffix}} compute rank-preserving surrogate distance via rdist
# which are proxies necessitating less computations.
# We get the equivalent for the radius to be able to compare it against
# vectors' rank-preserving surrogate distances.
float64_t r_radius
# Neighbors indices and distances are returned as np.ndarrays of np.ndarrays.
#
# For this implementation, we want resizable buffers which we will wrap
# into numpy arrays at the end. std::vector comes as a handy container
# for interacting efficiently with resizable buffers.
#
# Though it is possible to access their buffer address with
# std::vector::data, they can't be stolen: buffers lifetime
# is tied to their std::vector and are deallocated when
# std::vectors are.
#
# To solve this, we dynamically allocate std::vectors and then
# encapsulate them in a StdVectorSentinel responsible for
# freeing them when the associated np.ndarray is freed.
#
# Shared pointers (defined via shared_ptr) are use for safer memory management.
# Unique pointers (defined via unique_ptr) can't be used as datastructures
# are shared across threads for parallel_on_X; see _parallel_on_X_init_chunk.
shared_ptr[vector[vector[intp_t]]] neigh_indices
shared_ptr[vector[vector[float64_t]]] neigh_distances
# Used as array of pointers to private datastructures used in threads.
vector[shared_ptr[vector[vector[intp_t]]]] neigh_indices_chunks
vector[shared_ptr[vector[vector[float64_t]]]] neigh_distances_chunks
bint sort_results
@final
cdef void _merge_vectors(
self,
intp_t idx,
intp_t num_threads,
) noexcept nogil
cdef class EuclideanRadiusNeighbors{{name_suffix}}(RadiusNeighbors{{name_suffix}}):
"""EuclideanDistance-specialisation of RadiusNeighbors{{name_suffix}}."""
cdef:
MiddleTermComputer{{name_suffix}} middle_term_computer
const float64_t[::1] X_norm_squared
const float64_t[::1] Y_norm_squared
bint use_squared_distances
{{endfor}}
@@ -0,0 +1,514 @@
cimport numpy as cnp
import numpy as np
import warnings
from libcpp.memory cimport shared_ptr, make_shared
from libcpp.vector cimport vector
from libcpp.algorithm cimport move
from cython cimport final
from cython.operator cimport dereference as deref
from cython.parallel cimport parallel, prange
from ...utils._sorting cimport simultaneous_sort
from ...utils._typedefs cimport intp_t, float64_t
from ...utils._vector_sentinel cimport vector_to_nd_array
from numbers import Real
from scipy.sparse import issparse
from ...utils import check_array, check_scalar
from ...utils.fixes import _in_unstable_openblas_configuration
from ...utils.parallel import _get_threadpool_controller
cnp.import_array()
######################
cdef cnp.ndarray[object, ndim=1] coerce_vectors_to_nd_arrays(
shared_ptr[vector_vector_double_intp_t] vecs
):
"""Coerce a std::vector of std::vector to a ndarray of ndarray."""
cdef:
intp_t n = deref(vecs).size()
cnp.ndarray[object, ndim=1] nd_arrays_of_nd_arrays = np.empty(n, dtype=np.ndarray)
for i in range(n):
nd_arrays_of_nd_arrays[i] = vector_to_nd_array(&(deref(vecs)[i]))
return nd_arrays_of_nd_arrays
#####################
{{for name_suffix in ['64', '32']}}
from ._base cimport (
BaseDistancesReduction{{name_suffix}},
_sqeuclidean_row_norms{{name_suffix}}
)
from ._datasets_pair cimport DatasetsPair{{name_suffix}}
from ._middle_term_computer cimport MiddleTermComputer{{name_suffix}}
cdef class RadiusNeighbors{{name_suffix}}(BaseDistancesReduction{{name_suffix}}):
"""float{{name_suffix}} implementation of the RadiusNeighbors."""
@classmethod
def compute(
cls,
X,
Y,
float64_t radius,
str metric="euclidean",
chunk_size=None,
dict metric_kwargs=None,
str strategy=None,
bint return_distance=False,
bint sort_results=False,
):
"""Compute the radius-neighbors reduction.
This classmethod is responsible for introspecting the arguments
values to dispatch to the most appropriate implementation of
:class:`RadiusNeighbors{{name_suffix}}`.
This allows decoupling the API entirely from the implementation details
whilst maintaining RAII: all temporarily allocated datastructures necessary
for the concrete implementation are therefore freed when this classmethod
returns.
No instance should directly be created outside of this class method.
"""
if metric in ("euclidean", "sqeuclidean"):
# Specialized implementation of RadiusNeighbors for the Euclidean
# distance for the dense-dense and sparse-sparse cases.
# This implementation computes the distances by chunk using
# a decomposition of the Squared Euclidean distance.
# This specialisation has an improved arithmetic intensity for both
# the dense and sparse settings, allowing in most case speed-ups of
# several orders of magnitude compared to the generic RadiusNeighbors
# implementation.
# For more information see MiddleTermComputer.
use_squared_distances = metric == "sqeuclidean"
pda = EuclideanRadiusNeighbors{{name_suffix}}(
X=X, Y=Y, radius=radius,
use_squared_distances=use_squared_distances,
chunk_size=chunk_size,
strategy=strategy,
sort_results=sort_results,
metric_kwargs=metric_kwargs,
)
else:
# Fall back on a generic implementation that handles most scipy
# metrics by computing the distances between 2 vectors at a time.
pda = RadiusNeighbors{{name_suffix}}(
datasets_pair=DatasetsPair{{name_suffix}}.get_for(X, Y, metric, metric_kwargs),
radius=radius,
chunk_size=chunk_size,
strategy=strategy,
sort_results=sort_results,
)
# Limit the number of threads in second level of nested parallelism for BLAS
# to avoid threads over-subscription (in GEMM for instance).
with _get_threadpool_controller().limit(limits=1, user_api="blas"):
if pda.execute_in_parallel_on_Y:
pda._parallel_on_Y()
else:
pda._parallel_on_X()
return pda._finalize_results(return_distance)
def __init__(
self,
DatasetsPair{{name_suffix}} datasets_pair,
float64_t radius,
chunk_size=None,
strategy=None,
sort_results=False,
):
super().__init__(
datasets_pair=datasets_pair,
chunk_size=chunk_size,
strategy=strategy,
)
self.radius = check_scalar(radius, "radius", Real, min_val=0)
self.r_radius = self.datasets_pair.distance_metric._dist_to_rdist(radius)
self.sort_results = sort_results
# Allocating pointers to datastructures but not the datastructures themselves.
# There are as many pointers as effective threads.
#
# For the sake of explicitness:
# - when parallelizing on X, the pointers of those heaps are referencing
# self.neigh_distances and self.neigh_indices
# - when parallelizing on Y, the pointers of those heaps are referencing
# std::vectors of std::vectors which are thread-wise-allocated and whose
# content will be merged into self.neigh_distances and self.neigh_indices.
self.neigh_distances_chunks = vector[shared_ptr[vector[vector[float64_t]]]](
self.chunks_n_threads
)
self.neigh_indices_chunks = vector[shared_ptr[vector[vector[intp_t]]]](
self.chunks_n_threads
)
# Temporary datastructures which will be coerced to numpy arrays on before
# RadiusNeighbors.compute "return" and will be then freed.
self.neigh_distances = make_shared[vector[vector[float64_t]]](self.n_samples_X)
self.neigh_indices = make_shared[vector[vector[intp_t]]](self.n_samples_X)
cdef void _compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil:
cdef:
intp_t i, j
float64_t r_dist_i_j
for i in range(X_start, X_end):
for j in range(Y_start, Y_end):
r_dist_i_j = self.datasets_pair.surrogate_dist(i, j)
if r_dist_i_j <= self.r_radius:
deref(self.neigh_distances_chunks[thread_num])[i].push_back(r_dist_i_j)
deref(self.neigh_indices_chunks[thread_num])[i].push_back(j)
def _finalize_results(self, bint return_distance=False):
if return_distance:
# We need to recompute distances because we relied on
# surrogate distances for the reduction.
self.compute_exact_distances()
return (
coerce_vectors_to_nd_arrays(self.neigh_distances),
coerce_vectors_to_nd_arrays(self.neigh_indices),
)
return coerce_vectors_to_nd_arrays(self.neigh_indices)
cdef void _parallel_on_X_init_chunk(
self,
intp_t thread_num,
intp_t X_start,
intp_t X_end,
) noexcept nogil:
# As this strategy is embarrassingly parallel, we can set the
# thread vectors' pointers to the main vectors'.
self.neigh_distances_chunks[thread_num] = self.neigh_distances
self.neigh_indices_chunks[thread_num] = self.neigh_indices
@final
cdef void _parallel_on_X_prange_iter_finalize(
self,
intp_t thread_num,
intp_t X_start,
intp_t X_end,
) noexcept nogil:
cdef:
intp_t idx
# Sorting neighbors for each query vector of X
if self.sort_results:
for idx in range(X_start, X_end):
simultaneous_sort(
deref(self.neigh_distances)[idx].data(),
deref(self.neigh_indices)[idx].data(),
deref(self.neigh_indices)[idx].size()
)
cdef void _parallel_on_Y_init(
self,
) noexcept nogil:
cdef:
intp_t thread_num
# As chunks of X are shared across threads, so must datastructures to avoid race
# conditions: each thread has its own vectors of n_samples_X vectors which are
# then merged back in the main n_samples_X vectors.
for thread_num in range(self.chunks_n_threads):
self.neigh_distances_chunks[thread_num] = make_shared[vector[vector[float64_t]]](self.n_samples_X)
self.neigh_indices_chunks[thread_num] = make_shared[vector[vector[intp_t]]](self.n_samples_X)
@final
cdef void _merge_vectors(
self,
intp_t idx,
intp_t num_threads,
) noexcept nogil:
cdef:
intp_t thread_num
intp_t idx_n_elements = 0
intp_t last_element_idx = deref(self.neigh_indices)[idx].size()
# Resizing buffers only once for the given number of elements.
for thread_num in range(num_threads):
idx_n_elements += deref(self.neigh_distances_chunks[thread_num])[idx].size()
deref(self.neigh_distances)[idx].resize(last_element_idx + idx_n_elements)
deref(self.neigh_indices)[idx].resize(last_element_idx + idx_n_elements)
# Moving the elements by range using the range first element
# as the reference for the insertion.
for thread_num in range(num_threads):
move(
deref(self.neigh_distances_chunks[thread_num])[idx].begin(),
deref(self.neigh_distances_chunks[thread_num])[idx].end(),
deref(self.neigh_distances)[idx].begin() + last_element_idx
)
move(
deref(self.neigh_indices_chunks[thread_num])[idx].begin(),
deref(self.neigh_indices_chunks[thread_num])[idx].end(),
deref(self.neigh_indices)[idx].begin() + last_element_idx
)
last_element_idx += deref(self.neigh_distances_chunks[thread_num])[idx].size()
cdef void _parallel_on_Y_finalize(
self,
) noexcept nogil:
cdef:
intp_t idx
with nogil, parallel(num_threads=self.effective_n_threads):
# Merge vectors used in threads into the main ones.
# This is done in parallel sample-wise (no need for locks).
for idx in prange(self.n_samples_X, schedule='static'):
self._merge_vectors(idx, self.chunks_n_threads)
# The content of the vector have been std::moved.
# Hence they can't be used anymore and can be deleted.
# Their deletion is carried out automatically as the
# implementation relies on shared pointers.
# Sort in parallel in ascending order w.r.t the distances if requested.
if self.sort_results:
for idx in prange(self.n_samples_X, schedule='static'):
simultaneous_sort(
deref(self.neigh_distances)[idx].data(),
deref(self.neigh_indices)[idx].data(),
deref(self.neigh_indices)[idx].size()
)
return
cdef void compute_exact_distances(self) noexcept nogil:
"""Convert rank-preserving distances to pairwise distances in parallel."""
cdef:
intp_t i
vector[intp_t].size_type j
for i in prange(self.n_samples_X, nogil=True, schedule='static',
num_threads=self.effective_n_threads):
for j in range(deref(self.neigh_indices)[i].size()):
deref(self.neigh_distances)[i][j] = (
self.datasets_pair.distance_metric._rdist_to_dist(
# Guard against potential -0., causing nan production.
max(deref(self.neigh_distances)[i][j], 0.)
)
)
cdef class EuclideanRadiusNeighbors{{name_suffix}}(RadiusNeighbors{{name_suffix}}):
"""EuclideanDistance-specialisation of RadiusNeighbors{{name_suffix}}."""
@classmethod
def is_usable_for(cls, X, Y, metric) -> bool:
return (RadiusNeighbors{{name_suffix}}.is_usable_for(X, Y, metric)
and not _in_unstable_openblas_configuration())
def __init__(
self,
X,
Y,
float64_t radius,
bint use_squared_distances=False,
chunk_size=None,
strategy=None,
sort_results=False,
metric_kwargs=None,
):
if (
isinstance(metric_kwargs, dict) and
(metric_kwargs.keys() - {"X_norm_squared", "Y_norm_squared"})
):
warnings.warn(
f"Some metric_kwargs have been passed ({metric_kwargs}) but aren't "
f"usable for this case (EuclideanRadiusNeighbors64) and will be ignored.",
UserWarning,
stacklevel=3,
)
super().__init__(
# The datasets pair here is used for exact distances computations
datasets_pair=DatasetsPair{{name_suffix}}.get_for(X, Y, metric="euclidean"),
radius=radius,
chunk_size=chunk_size,
strategy=strategy,
sort_results=sort_results,
)
cdef:
intp_t dist_middle_terms_chunks_size = self.Y_n_samples_chunk * self.X_n_samples_chunk
self.middle_term_computer = MiddleTermComputer{{name_suffix}}.get_for(
X,
Y,
self.effective_n_threads,
self.chunks_n_threads,
dist_middle_terms_chunks_size,
n_features=X.shape[1],
chunk_size=self.chunk_size,
)
if metric_kwargs is not None and "Y_norm_squared" in metric_kwargs:
self.Y_norm_squared = check_array(
metric_kwargs.pop("Y_norm_squared"),
ensure_2d=False,
input_name="Y_norm_squared",
dtype=np.float64,
)
else:
self.Y_norm_squared = _sqeuclidean_row_norms{{name_suffix}}(
Y,
self.effective_n_threads,
)
if metric_kwargs is not None and "X_norm_squared" in metric_kwargs:
self.X_norm_squared = check_array(
metric_kwargs.pop("X_norm_squared"),
ensure_2d=False,
input_name="X_norm_squared",
dtype=np.float64,
)
else:
# Do not recompute norms if datasets are identical.
self.X_norm_squared = (
self.Y_norm_squared if X is Y else
_sqeuclidean_row_norms{{name_suffix}}(
X,
self.effective_n_threads,
)
)
self.use_squared_distances = use_squared_distances
if use_squared_distances:
# In this specialisation and this setup, the value passed to the radius is
# already considered to be the adapted radius, so we overwrite it.
self.r_radius = radius
@final
cdef void _parallel_on_X_parallel_init(
self,
intp_t thread_num,
) noexcept nogil:
RadiusNeighbors{{name_suffix}}._parallel_on_X_parallel_init(self, thread_num)
self.middle_term_computer._parallel_on_X_parallel_init(thread_num)
@final
cdef void _parallel_on_X_init_chunk(
self,
intp_t thread_num,
intp_t X_start,
intp_t X_end,
) noexcept nogil:
RadiusNeighbors{{name_suffix}}._parallel_on_X_init_chunk(self, thread_num, X_start, X_end)
self.middle_term_computer._parallel_on_X_init_chunk(thread_num, X_start, X_end)
@final
cdef void _parallel_on_X_pre_compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil:
RadiusNeighbors{{name_suffix}}._parallel_on_X_pre_compute_and_reduce_distances_on_chunks(
self,
X_start, X_end,
Y_start, Y_end,
thread_num,
)
self.middle_term_computer._parallel_on_X_pre_compute_and_reduce_distances_on_chunks(
X_start, X_end, Y_start, Y_end, thread_num,
)
@final
cdef void _parallel_on_Y_init(
self,
) noexcept nogil:
RadiusNeighbors{{name_suffix}}._parallel_on_Y_init(self)
self.middle_term_computer._parallel_on_Y_init()
@final
cdef void _parallel_on_Y_parallel_init(
self,
intp_t thread_num,
intp_t X_start,
intp_t X_end,
) noexcept nogil:
RadiusNeighbors{{name_suffix}}._parallel_on_Y_parallel_init(self, thread_num, X_start, X_end)
self.middle_term_computer._parallel_on_Y_parallel_init(thread_num, X_start, X_end)
@final
cdef void _parallel_on_Y_pre_compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil:
RadiusNeighbors{{name_suffix}}._parallel_on_Y_pre_compute_and_reduce_distances_on_chunks(
self,
X_start, X_end,
Y_start, Y_end,
thread_num,
)
self.middle_term_computer._parallel_on_Y_pre_compute_and_reduce_distances_on_chunks(
X_start, X_end, Y_start, Y_end, thread_num
)
@final
cdef void compute_exact_distances(self) noexcept nogil:
if not self.use_squared_distances:
RadiusNeighbors{{name_suffix}}.compute_exact_distances(self)
@final
cdef void _compute_and_reduce_distances_on_chunks(
self,
intp_t X_start,
intp_t X_end,
intp_t Y_start,
intp_t Y_end,
intp_t thread_num,
) noexcept nogil:
cdef:
intp_t i, j
float64_t sqeuclidean_dist_i_j
intp_t n_X = X_end - X_start
intp_t n_Y = Y_end - Y_start
float64_t *dist_middle_terms = self.middle_term_computer._compute_dist_middle_terms(
X_start, X_end, Y_start, Y_end, thread_num
)
# Pushing the distance and their associated indices in vectors.
for i in range(n_X):
for j in range(n_Y):
sqeuclidean_dist_i_j = (
self.X_norm_squared[i + X_start]
+ dist_middle_terms[i * n_Y + j]
+ self.Y_norm_squared[j + Y_start]
)
# Catastrophic cancellation might cause -0. to be present,
# e.g. when computing d(x_i, y_i) when X is Y.
sqeuclidean_dist_i_j = max(0., sqeuclidean_dist_i_j)
if sqeuclidean_dist_i_j <= self.r_radius:
deref(self.neigh_distances_chunks[thread_num])[i + X_start].push_back(sqeuclidean_dist_i_j)
deref(self.neigh_indices_chunks[thread_num])[i + X_start].push_back(j + Y_start)
{{endfor}}
@@ -0,0 +1,217 @@
import warnings
from cython cimport floating, final, integral
from cython.operator cimport dereference as deref
from cython.parallel cimport parallel, prange
from ._classmode cimport WeightingStrategy
from ...utils._typedefs cimport intp_t, float64_t, uint8_t
import numpy as np
from scipy.sparse import issparse
from ...utils.parallel import _get_threadpool_controller
{{for name_suffix in ["32", "64"]}}
from ._radius_neighbors cimport RadiusNeighbors{{name_suffix}}
from ._datasets_pair cimport DatasetsPair{{name_suffix}}
cdef class RadiusNeighborsClassMode{{name_suffix}}(RadiusNeighbors{{name_suffix}}):
"""
{{name_suffix}}bit implementation of RadiusNeighborsClassMode.
"""
cdef:
const intp_t[::1] Y_labels
const intp_t[::1] unique_Y_labels
intp_t outlier_label_index
bint outlier_label_exists
bint outliers_exist
uint8_t[::1] outliers
object outlier_label
float64_t[:, ::1] class_scores
WeightingStrategy weight_type
@classmethod
def compute(
cls,
X,
Y,
float64_t radius,
weights,
Y_labels,
unique_Y_labels,
outlier_label=None,
str metric="euclidean",
chunk_size=None,
dict metric_kwargs=None,
str strategy=None,
):
# Use a generic implementation that handles most scipy
# metrics by computing the distances between 2 vectors at a time.
pda = RadiusNeighborsClassMode{{name_suffix}}(
datasets_pair=DatasetsPair{{name_suffix}}.get_for(X, Y, metric, metric_kwargs),
radius=radius,
chunk_size=chunk_size,
strategy=strategy,
weights=weights,
Y_labels=Y_labels,
unique_Y_labels=unique_Y_labels,
outlier_label=outlier_label,
)
# Limit the number of threads in second level of nested parallelism for BLAS
# to avoid threads over-subscription (in GEMM for instance).
with _get_threadpool_controller().limit(limits=1, user_api="blas"):
if pda.execute_in_parallel_on_Y:
pda._parallel_on_Y()
else:
pda._parallel_on_X()
return pda._finalize_results()
def __init__(
self,
DatasetsPair{{name_suffix}} datasets_pair,
const intp_t[::1] Y_labels,
const intp_t[::1] unique_Y_labels,
float64_t radius,
chunk_size=None,
strategy=None,
weights=None,
outlier_label=None,
):
super().__init__(
datasets_pair=datasets_pair,
chunk_size=chunk_size,
strategy=strategy,
radius=radius,
)
if weights == "uniform":
self.weight_type = WeightingStrategy.uniform
elif weights == "distance":
self.weight_type = WeightingStrategy.distance
else:
self.weight_type = WeightingStrategy.callable
self.Y_labels = Y_labels
self.unique_Y_labels = unique_Y_labels
self.outlier_label_index = -1
self.outliers_exist = False
self.outlier_label = outlier_label
self.outliers = np.zeros(self.n_samples_X, dtype=np.bool_)
cdef intp_t idx
if self.outlier_label is not None:
for idx in range(self.unique_Y_labels.shape[0]):
if self.unique_Y_labels[idx] == outlier_label:
self.outlier_label_index = idx
# Map from set of unique labels to their indices in `class_scores`
# Buffer used in building a histogram for one-pass weighted mode
self.class_scores = np.zeros(
(self.n_samples_X, unique_Y_labels.shape[0]), dtype=np.float64,
)
cdef inline void weighted_histogram_mode(
self,
intp_t sample_index,
intp_t sample_n_neighbors,
intp_t* indices,
float64_t* distances,
) noexcept nogil:
cdef:
intp_t neighbor_idx, neighbor_class_idx, label_index
float64_t score_incr = 1
bint use_distance_weighting = (
self.weight_type == WeightingStrategy.distance
)
if sample_n_neighbors == 0:
self.outliers_exist = True
self.outliers[sample_index] = True
if self.outlier_label_index >= 0:
self.class_scores[sample_index][self.outlier_label_index] = score_incr
return
# Iterate over the neighbors. This can be different for
# each of the samples as they are based on the radius.
for neighbor_rank in range(sample_n_neighbors):
if use_distance_weighting:
score_incr = 1 / distances[neighbor_rank]
neighbor_idx = indices[neighbor_rank]
neighbor_class_idx = self.Y_labels[neighbor_idx]
self.class_scores[sample_index][neighbor_class_idx] += score_incr
return
@final
cdef void _parallel_on_X_prange_iter_finalize(
self,
intp_t thread_num,
intp_t X_start,
intp_t X_end,
) noexcept nogil:
cdef:
intp_t idx
for idx in range(X_start, X_end):
self.weighted_histogram_mode(
sample_index=idx,
sample_n_neighbors=deref(self.neigh_indices)[idx].size(),
indices=deref(self.neigh_indices)[idx].data(),
distances=deref(self.neigh_distances)[idx].data(),
)
return
@final
cdef void _parallel_on_Y_finalize(
self,
) noexcept nogil:
cdef:
intp_t idx
with nogil, parallel(num_threads=self.effective_n_threads):
# Merge vectors used in threads into the main ones.
# This is done in parallel sample-wise (no need for locks).
for idx in prange(self.n_samples_X, schedule='static'):
self._merge_vectors(idx, self.chunks_n_threads)
for idx in prange(self.n_samples_X, schedule='static'):
self.weighted_histogram_mode(
sample_index=idx,
sample_n_neighbors=deref(self.neigh_indices)[idx].size(),
indices=deref(self.neigh_indices)[idx].data(),
distances=deref(self.neigh_distances)[idx].data(),
)
return
def _finalize_results(self):
if self.outliers_exist and self.outlier_label is None:
raise ValueError(
"No neighbors found for test samples %r, "
"you can try using larger radius, "
"giving a label for outliers, "
"or considering removing them from your dataset."
% np.where(self.outliers)[0]
)
if self.outliers_exist and self.outlier_label_index < 0:
warnings.warn(
"Outlier label %s is not in training "
"classes. All class probabilities of "
"outliers will be assigned with 0."
% self.outlier_label
)
probabilities = np.asarray(self.class_scores)
normalizer = probabilities.sum(axis=1, keepdims=True)
normalizer[normalizer == 0.0] = 1.0
probabilities /= normalizer
return probabilities
{{endfor}}
@@ -0,0 +1,193 @@
# Note: the dependencies between different Cython files in
# _pairwise_distances_reduction is probably one of the most involved in
# scikit-learn. If you change this file make sure you build from scratch:
# rm -rf build; make dev-meson
# run a command like this:
# ninja -C build/cp312 -t missingdeps
# and make sure that the output is something like:
# No missing dependencies on generated files found.
# _pairwise_distances_reduction is cimported from other subpackages so this is
# needed for the cimport to work
_pairwise_distances_reduction_cython_tree = [
fs.copyfile('__init__.py'),
# We are in a sub-module of metrics, so we always need to have
# sklearn/metrics/__init__.py copied to the build directory to avoid the
# error:
# relative cimport beyond main package is not allowed
metrics_cython_tree
]
_classmode_pxd = fs.copyfile('_classmode.pxd')
_datasets_pair_pxd = custom_target(
'_datasets_pair_pxd',
output: '_datasets_pair.pxd',
input: '_datasets_pair.pxd.tp',
command: [tempita, '@INPUT@', '-o', '@OUTDIR@']
)
_datasets_pair_pyx = custom_target(
'_datasets_pair_pyx',
output: '_datasets_pair.pyx',
input: '_datasets_pair.pyx.tp',
command: [tempita, '@INPUT@', '-o', '@OUTDIR@'],
# TODO in principle this should go in py.exension_module below. This is
# temporary work-around for dependency issue with .pyx.tp files. For more
# details, see https://github.com/mesonbuild/meson/issues/13212
depends: [_datasets_pair_pxd, _pairwise_distances_reduction_cython_tree, utils_cython_tree],
)
_datasets_pair = py.extension_module(
'_datasets_pair',
cython_gen_cpp.process(_datasets_pair_pyx),
dependencies: [np_dep],
subdir: 'sklearn/metrics/_pairwise_distances_reduction',
install: true
)
_base_pxd = custom_target(
'_base_pxd',
output: '_base.pxd',
input: '_base.pxd.tp',
command: [tempita, '@INPUT@', '-o', '@OUTDIR@']
)
_base_pyx = custom_target(
'_base_pyx',
output: '_base.pyx',
input: '_base.pyx.tp',
command: [tempita, '@INPUT@', '-o', '@OUTDIR@'],
# TODO in principle this should go in py.exension_module below. This is
# temporary work-around for dependency issue with .pyx.tp files. For more
# details, see https://github.com/mesonbuild/meson/issues/13212
depends: [_base_pxd, _pairwise_distances_reduction_cython_tree,
_datasets_pair_pxd, utils_cython_tree],
)
_base = py.extension_module(
'_base',
cython_gen_cpp.process(_base_pyx),
dependencies: [np_dep, openmp_dep],
subdir: 'sklearn/metrics/_pairwise_distances_reduction',
install: true
)
_middle_term_computer_pxd = custom_target(
'_middle_term_computer_pxd',
output: '_middle_term_computer.pxd',
input: '_middle_term_computer.pxd.tp',
command: [tempita, '@INPUT@', '-o', '@OUTDIR@']
)
_middle_term_computer_pyx = custom_target(
'_middle_term_computer_pyx',
output: '_middle_term_computer.pyx',
input: '_middle_term_computer.pyx.tp',
command: [tempita, '@INPUT@', '-o', '@OUTDIR@'],
# TODO in principle this should go in py.exension_module below. This is
# temporary work-around for dependency issue with .pyx.tp files. For more
# details, see https://github.com/mesonbuild/meson/issues/13212
depends: [_middle_term_computer_pxd,
_pairwise_distances_reduction_cython_tree,
utils_cython_tree],
)
_middle_term_computer = py.extension_module(
'_middle_term_computer',
cython_gen_cpp.process(_middle_term_computer_pyx),
dependencies: [np_dep],
subdir: 'sklearn/metrics/_pairwise_distances_reduction',
install: true
)
_argkmin_pxd = custom_target(
'_argkmin_pxd',
output: '_argkmin.pxd',
input: '_argkmin.pxd.tp',
command: [tempita, '@INPUT@', '-o', '@OUTDIR@']
)
_argkmin_pyx = custom_target(
'_argkmin_pyx',
output: '_argkmin.pyx',
input: '_argkmin.pyx.tp',
command: [tempita, '@INPUT@', '-o', '@OUTDIR@'],
# TODO in principle this should go in py.exension_module below. This is
# temporary work-around for dependency issue with .pyx.tp files. For more
# details, see https://github.com/mesonbuild/meson/issues/13212
depends: [_argkmin_pxd,
_pairwise_distances_reduction_cython_tree,
_datasets_pair_pxd, _base_pxd, _middle_term_computer_pxd],
)
_argkmin = py.extension_module(
'_argkmin',
cython_gen_cpp.process(_argkmin_pyx),
dependencies: [np_dep, openmp_dep],
subdir: 'sklearn/metrics/_pairwise_distances_reduction',
install: true
)
_radius_neighbors_pxd = custom_target(
'_radius_neighbors_pxd',
output: '_radius_neighbors.pxd',
input: '_radius_neighbors.pxd.tp',
command: [tempita, '@INPUT@', '-o', '@OUTDIR@']
)
_radius_neighbors_pyx = custom_target(
'_radius_neighbors_pyx',
output: '_radius_neighbors.pyx',
input: '_radius_neighbors.pyx.tp',
command: [tempita, '@INPUT@', '-o', '@OUTDIR@'],
# TODO in principle this should go in py.exension_module below. This is
# temporary work-around for dependency issue with .pyx.tp files. For more
# details, see https://github.com/mesonbuild/meson/issues/13212
depends: [_radius_neighbors_pxd,
_datasets_pair_pxd, _base_pxd, _middle_term_computer_pxd,
_pairwise_distances_reduction_cython_tree, utils_cython_tree],
)
_radius_neighbors = py.extension_module(
'_radius_neighbors',
cython_gen_cpp.process(_radius_neighbors_pyx),
dependencies: [np_dep, openmp_dep],
subdir: 'sklearn/metrics/_pairwise_distances_reduction',
install: true
)
_argkmin_classmode_pyx = custom_target(
'_argkmin_classmode_pyx',
output: '_argkmin_classmode.pyx',
input: '_argkmin_classmode.pyx.tp',
command: [tempita, '@INPUT@', '-o', '@OUTDIR@'],
# TODO in principle this should go in py.exension_module below. This is
# temporary work-around for dependency issue with .pyx.tp files. For more
# details, see https://github.com/mesonbuild/meson/issues/13212
depends: [_classmode_pxd,
_argkmin_pxd, _pairwise_distances_reduction_cython_tree,
_datasets_pair_pxd, _base_pxd, _middle_term_computer_pxd, utils_cython_tree],
)
_argkmin_classmode = py.extension_module(
'_argkmin_classmode',
cython_gen_cpp.process(_argkmin_classmode_pyx),
dependencies: [np_dep, openmp_dep],
# XXX: for some reason -fno-sized-deallocation is needed otherwise there is
# an error with undefined symbol _ZdlPv at import time in manylinux wheels.
# See https://github.com/scikit-learn/scikit-learn/issues/28596 for more details.
cpp_args: ['-fno-sized-deallocation'],
subdir: 'sklearn/metrics/_pairwise_distances_reduction',
install: true
)
_radius_neighbors_classmode_pyx = custom_target(
'_radius_neighbors_classmode_pyx',
output: '_radius_neighbors_classmode.pyx',
input: '_radius_neighbors_classmode.pyx.tp',
command: [tempita, '@INPUT@', '-o', '@OUTDIR@'],
# TODO in principle this should go in py.exension_module below. This is
# temporary work-around for dependency issue with .pyx.tp files. For more
# details, see https://github.com/mesonbuild/meson/issues/13212
depends: [_classmode_pxd,
_middle_term_computer_pxd, _radius_neighbors_pxd,
_pairwise_distances_reduction_cython_tree,
_datasets_pair_pxd, _base_pxd, utils_cython_tree],
)
_radius_neighbors_classmode = py.extension_module(
'_radius_neighbors_classmode',
cython_gen_cpp.process(_radius_neighbors_classmode_pyx),
dependencies: [np_dep, openmp_dep],
subdir: 'sklearn/metrics/_pairwise_distances_reduction',
install: true
)
@@ -0,0 +1,107 @@
# Authors: The scikit-learn developers
# SPDX-License-Identifier: BSD-3-Clause
from cython cimport floating
from cython.parallel cimport prange
from libc.math cimport fabs
from ..utils._typedefs cimport intp_t
from ..utils._openmp_helpers import _openmp_effective_n_threads
def _chi2_kernel_fast(floating[:, :] X,
floating[:, :] Y,
floating[:, :] result):
cdef intp_t i, j, k
cdef intp_t n_samples_X = X.shape[0]
cdef intp_t n_samples_Y = Y.shape[0]
cdef intp_t n_features = X.shape[1]
cdef double res, nom, denom
with nogil:
for i in range(n_samples_X):
for j in range(n_samples_Y):
res = 0
for k in range(n_features):
denom = (X[i, k] - Y[j, k])
nom = (X[i, k] + Y[j, k])
if nom != 0:
res += denom * denom / nom
result[i, j] = -res
def _sparse_manhattan(
const floating[::1] X_data,
const int[:] X_indices,
const int[:] X_indptr,
const floating[::1] Y_data,
const int[:] Y_indices,
const int[:] Y_indptr,
double[:, ::1] D,
):
"""Pairwise L1 distances for CSR matrices.
Usage:
>>> D = np.zeros(X.shape[0], Y.shape[0])
>>> _sparse_manhattan(X.data, X.indices, X.indptr,
... Y.data, Y.indices, Y.indptr,
... D)
"""
cdef intp_t px, py, i, j, ix, iy
cdef double d = 0.0
cdef int m = D.shape[0]
cdef int n = D.shape[1]
cdef int X_indptr_end = 0
cdef int Y_indptr_end = 0
cdef int num_threads = _openmp_effective_n_threads()
# We scan the matrices row by row.
# Given row px in X and row py in Y, we find the positions (i and j
# respectively), in .indices where the indices for the two rows start.
# If the indices (ix and iy) are the same, the corresponding data values
# are processed and the cursors i and j are advanced.
# If not, the lowest index is considered. Its associated data value is
# processed and its cursor is advanced.
# We proceed like this until one of the cursors hits the end for its row.
# Then we process all remaining data values in the other row.
# Below the avoidance of inplace operators is intentional.
# When prange is used, the inplace operator has a special meaning, i.e. it
# signals a "reduction"
for px in prange(m, nogil=True, num_threads=num_threads):
X_indptr_end = X_indptr[px + 1]
for py in range(n):
Y_indptr_end = Y_indptr[py + 1]
i = X_indptr[px]
j = Y_indptr[py]
d = 0.0
while i < X_indptr_end and j < Y_indptr_end:
ix = X_indices[i]
iy = Y_indices[j]
if ix == iy:
d = d + fabs(X_data[i] - Y_data[j])
i = i + 1
j = j + 1
elif ix < iy:
d = d + fabs(X_data[i])
i = i + 1
else:
d = d + fabs(Y_data[j])
j = j + 1
if i == X_indptr_end:
while j < Y_indptr_end:
d = d + fabs(Y_data[j])
j = j + 1
else:
while i < X_indptr_end:
d = d + fabs(X_data[i])
i = i + 1
D[px, py] = d
@@ -0,0 +1,69 @@
# Authors: The scikit-learn developers
# SPDX-License-Identifier: BSD-3-Clause
from libc.math cimport exp, lgamma
from ...utils._typedefs cimport float64_t, int64_t
import numpy as np
from scipy.special import gammaln
def expected_mutual_information(contingency, int64_t n_samples):
"""Calculate the expected mutual information for two labelings."""
cdef:
float64_t emi = 0
int64_t n_rows, n_cols
float64_t term2, term3, gln
int64_t[::1] a_view, b_view
float64_t[::1] term1
float64_t[::1] gln_a, gln_b, gln_Na, gln_Nb, gln_Nnij, log_Nnij
float64_t[::1] log_a, log_b
Py_ssize_t i, j, nij
int64_t start, end
n_rows, n_cols = contingency.shape
a = np.ravel(contingency.sum(axis=1).astype(np.int64, copy=False))
b = np.ravel(contingency.sum(axis=0).astype(np.int64, copy=False))
a_view = a
b_view = b
# any labelling with zero entropy implies EMI = 0
if a.size == 1 or b.size == 1:
return 0.0
# There are three major terms to the EMI equation, which are multiplied to
# and then summed over varying nij values.
# While nijs[0] will never be used, having it simplifies the indexing.
nijs = np.arange(0, max(np.max(a), np.max(b)) + 1, dtype='float')
nijs[0] = 1 # Stops divide by zero warnings. As its not used, no issue.
# term1 is nij / N
term1 = nijs / n_samples
# term2 is log((N*nij) / (a * b)) == log(N * nij) - log(a * b)
log_a = np.log(a)
log_b = np.log(b)
# term2 uses log(N * nij) = log(N) + log(nij)
log_Nnij = np.log(n_samples) + np.log(nijs)
# term3 is large, and involved many factorials. Calculate these in log
# space to stop overflows.
gln_a = gammaln(a + 1)
gln_b = gammaln(b + 1)
gln_Na = gammaln(n_samples - a + 1)
gln_Nb = gammaln(n_samples - b + 1)
gln_Nnij = gammaln(nijs + 1) + gammaln(n_samples + 1)
# emi itself is a summation over the various values.
for i in range(n_rows):
for j in range(n_cols):
start = max(1, a_view[i] - n_samples + b_view[j])
end = min(a_view[i], b_view[j]) + 1
for nij in range(start, end):
term2 = log_Nnij[nij] - log_a[i] - log_b[j]
# Numerators are positive, denominators are negative.
gln = (gln_a[i] + gln_b[j] + gln_Na[i] + gln_Nb[j]
- gln_Nnij[nij] - lgamma(a_view[i] - nij + 1)
- lgamma(b_view[j] - nij + 1)
- lgamma(n_samples - a_view[i] - b_view[j] + nij + 1))
term3 = exp(gln)
emi += (term1[nij] * term2 * term3)
return emi
@@ -0,0 +1,6 @@
py.extension_module(
'_expected_mutual_info_fast',
cython_gen.process('_expected_mutual_info_fast.pyx'),
subdir: 'sklearn/metrics/cluster',
install: true
)
@@ -0,0 +1,49 @@
# Metrics is cimported from other subpackages so this is needed for the cimport
# to work
metrics_cython_tree = [
fs.copyfile('__init__.py')
]
# Some metrics code cimports code from utils, we may as well copy all the necessary files
metrics_cython_tree += utils_cython_tree
_dist_metrics_pxd = custom_target(
'_dist_metrics_pxd',
output: '_dist_metrics.pxd',
input: '_dist_metrics.pxd.tp',
command: [tempita, '@INPUT@', '-o', '@OUTDIR@'],
# Need to install the generated pxd because it is needed in other subpackages
# Cython code, e.g. sklearn.cluster
install_dir: sklearn_dir / 'metrics',
install: true,
)
metrics_cython_tree += [_dist_metrics_pxd]
_dist_metrics_pyx = custom_target(
'_dist_metrics_pyx',
output: '_dist_metrics.pyx',
input: '_dist_metrics.pyx.tp',
command: [tempita, '@INPUT@', '-o', '@OUTDIR@'],
# TODO in principle this should go in py.exension_module below. This is
# temporary work-around for dependency issue with .pyx.tp files. For more
# details, see https://github.com/mesonbuild/meson/issues/13212
depends: metrics_cython_tree,
)
_dist_metrics = py.extension_module(
'_dist_metrics',
cython_gen.process(_dist_metrics_pyx),
dependencies: [np_dep],
subdir: 'sklearn/metrics',
install: true
)
py.extension_module(
'_pairwise_fast',
[cython_gen.process('_pairwise_fast.pyx'), metrics_cython_tree],
dependencies: [openmp_dep],
subdir: 'sklearn/metrics',
install: true
)
subdir('_pairwise_distances_reduction')
subdir('cluster')