Source code for matchms.similarity.CosineBlink

from collections.abc import Sequence
import numpy as np
from numba import njit  # TODO: check if numba is necessary/useful here
from scipy.sparse import csr_array
from matchms.Scores import Scores
from matchms.typing import SpectrumType
from .BaseSimilarity import BaseSimilarity


@njit(cache=True, fastmath=True)
def _windowed_sum_numba(
        source_bins: np.ndarray,
        source_vals: np.ndarray,
        query_positions: np.ndarray,
        R: int,
    ) -> np.ndarray:
    """
    Two-pointer windowed sum for sorted integer arrays.

    For each position `pos` in `query_positions`, compute:
        sum(source_vals[j]) where source_bins[j] ∈ [pos - R, pos + R]

    Parameters
    ----------
    source_bins : np.ndarray
        Sorted 1D array of integer bin indices (ascending).
    source_vals : np.ndarray
        1D array of values aligned to `source_bins`.
    query_positions : np.ndarray
        1D array of integer bin positions at which to evaluate the windowed sum.
    R : int
        Window radius in bins (inclusive on both sides).

    Returns
    -------
    np.ndarray
        1D array of windowed sums, same length/order as `query_positions`.
    """
    n_bins = source_bins.size
    m = query_positions.size
    out = np.zeros(m, dtype=np.float32)
    if n_bins == 0 or m == 0:
        return out

    left_idx = 0
    right_idx = 0
    acc = 0.0

    for i in range(m):
        pos = query_positions[i]
        left_bound = pos - R
        right_bound = pos + R

        # expand right edge
        while right_idx < n_bins and source_bins[right_idx] <= right_bound:
            acc += source_vals[right_idx]
            right_idx += 1

        # contract left edge
        while left_idx < right_idx and source_bins[left_idx] < left_bound:
            acc -= source_vals[left_idx]
            left_idx += 1

        out[i] = acc
    return out