annif.suggestion

Complexity

Total Complexity

39

Size/Duplication

Total Lines	182
Duplicated Lines	0 %

Importance

Changes

0

21 Methods

Rating	Name	Duplication	Size	Complexity
A	SuggestionResult.__len__()	0	4	1
A	SuggestionResult.as_vector()	0	6	1
A	SuggestionResult.__iter__()	0	5	1
A	VectorSuggestionResult._vector_to_list_suggestion()	0	8	3
A	VectorSuggestionResult.as_vector()	0	5	2
A	VectorSuggestionResult.__len__()	0	2	1
A	VectorSuggestionResult.subject_order()	0	5	2
A	VectorSuggestionResult.__iter__()	0	4	2
A	VectorSuggestionResult.__init__()	0	6	1
A	SuggestionBatch.__init__()	0	3	1
A	SparseSuggestionResult.as_vector()	0	5	2
A	SuggestionBatch.filter()	0	5	1
A	SuggestionBatch.__len__()	0	2	1
A	SuggestionBatch.__getitem__()	0	4	3
A	SuggestionResults.filter()	0	6	1
A	SparseSuggestionResult.__init__()	0	3	1
A	SuggestionResults.__iter__()	0	2	1
A	SuggestionResults.__init__()	0	5	1
A	SparseSuggestionResult.__iter__()	0	8	2
A	SuggestionBatch.from_sequence()	0	14	5
A	SparseSuggestionResult.__len__()	0	3	1

"""Representing suggested subjects."""

import abc
import collections
import itertools

import numpy as np
from scipy.sparse import dok_array

SubjectSuggestion = collections.namedtuple("SubjectSuggestion", "subject_id score")
WeightedSuggestionsBatch = collections.namedtuple(
    "WeightedSuggestionsBatch", "hit_sets weight subjects"
)


def filter_suggestion(preds, limit=None, threshold=0.0):
    """filter a 2D sparse suggestion array (csr_array), retaining only the
    top K suggestions with a score above or equal to the threshold for each
    individual prediction; the rest will be left as zeros"""

    filtered = dok_array(preds.shape, dtype=np.float32)
    for row in range(preds.shape[0]):
        arow = preds.getrow(row)
        top_k = arow.data.argsort()[::-1]
        if limit is not None:
            top_k = top_k[:limit]
        for idx in top_k:
            val = arow.data[idx]
            if val < threshold:
                break
            filtered[row, arow.indices[idx]] = val
    return filtered.tocsr()


class SuggestionResult(metaclass=abc.ABCMeta):
    """Abstract base class for a set of hits returned by an analysis
    operation."""

    @abc.abstractmethod
    def __iter__(self):
        """Return the hits as an iterator that returns SubjectSuggestion objects,
        highest scores first."""
        pass  # pragma: no cover

    @abc.abstractmethod
    def as_vector(self, size, destination=None):
        """Return the hits as a one-dimensional score vector of given size.
        If destination array is given (not None) it will be used, otherwise a
        new array will be created."""
        pass  # pragma: no cover

    @abc.abstractmethod
    def __len__(self):
        """Return the number of hits with non-zero scores."""
        pass  # pragma: no cover


class VectorSuggestionResult(SuggestionResult):
    """SuggestionResult implementation based primarily on NumPy vectors."""

    def __init__(self, vector):
        vector_f32 = vector.astype(np.float32)
        # limit scores to the range 0.0 .. 1.0
        self._vector = np.minimum(np.maximum(vector_f32, 0.0), 1.0)
        self._subject_order = None
        self._lsr = None

    def _vector_to_list_suggestion(self):
        hits = []
        for subject_id in self.subject_order:
            score = self._vector[subject_id]
            if score <= 0.0:
                break  # we can skip the remaining ones
            hits.append(SubjectSuggestion(subject_id=subject_id, score=float(score)))
        return hits

    @property
    def subject_order(self):
        if self._subject_order is None:
            self._subject_order = np.argsort(self._vector)[::-1]
        return self._subject_order

    def __iter__(self):
        if self._lsr is None:
            self._lsr = self._vector_to_list_suggestion()
        return iter(self._lsr)

    def as_vector(self, size, destination=None):
        if destination is not None:
            np.copyto(destination, self._vector)
            return destination
        return self._vector

    def __len__(self):
        return (self._vector > 0.0).sum()


class SparseSuggestionResult(SuggestionResult):
    """SuggestionResult implementation backed by a single row of a sparse array."""

    def __init__(self, array, idx):
        self._array = array
        self._idx = idx

    def __iter__(self):
        _, cols = self._array[[self._idx], :].nonzero()
        suggestions = [
            SubjectSuggestion(subject_id=col, score=float(self._array[self._idx, col]))
            for col in cols
        ]
        return iter(
            sorted(suggestions, key=lambda suggestion: suggestion.score, reverse=True)
        )

    def as_vector(self, size, destination=None):
        if destination is not None:
            print("as_vector called with destination not None")
            return None
        return self._array[[self._idx], :].toarray()[0]

    def __len__(self):
        _, cols = self._array[[self._idx], :].nonzero()
        return len(cols)


class SuggestionBatch:
    """Subject suggestions for a batch of documents."""

    def __init__(self, array):
        """Create a new SuggestionBatch from a csr_array"""
        self.array = array

    @classmethod
    def from_sequence(cls, suggestion_results, subject_index, limit=None):
        """Create a new SuggestionBatch from a sequence of SuggestionResult objects."""

        deprecated = set(subject_index.deprecated_ids())

        # create a dok_array for fast construction
        ar = dok_array((len(suggestion_results), len(subject_index)), dtype=np.float32)
        for idx, result in enumerate(suggestion_results):
            for suggestion in itertools.islice(result, limit):
                if suggestion.subject_id in deprecated or suggestion.score < 0.0:
                    continue
                ar[idx, suggestion.subject_id] = min(suggestion.score, 1.0)
        return cls(ar.tocsr())

    def filter(self, limit=None, threshold=0.0):
        """Return a subset of the hits, filtered by the given limit and
        score threshold, as another SuggestionBatch object."""

        return SuggestionBatch(filter_suggestion(self.array, limit, threshold))

    def __getitem__(self, idx):
        if idx < 0 or idx >= len(self):
            raise IndexError
        return SparseSuggestionResult(self.array, idx)

    def __len__(self):
        return self.array.shape[0]


class SuggestionResults:
    """Subject suggestions for a potentially very large number of documents."""

    def __init__(self, batches):
        """Initialize a new SuggestionResults from an iterable that provides
        SuggestionBatch objects."""

        self.batches = batches

    def filter(self, limit=None, threshold=0.0):
        """Return a view of these suggestions, filtered by the given limit
        and/or threshold, as another SuggestionResults object."""

        return SuggestionResults(
            (batch.filter(limit, threshold) for batch in self.batches)
        )

    def __iter__(self):
        return iter(itertools.chain.from_iterable(self.batches))