annif.suggestion.SparseSuggestionResult.__init__() - Code Metrics - Inspection of "move filter_suggestion to annif.suggestion where i..." - NatLibFi/Annif - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — issue678-refactor-suggestionre... ( d250de...96588c )

by Osma

created 2023-03-20 07:59 UTC

annif.suggestion.SparseSuggestionResult.init() A

↳ Parent: annif.suggestion

Complexity

Conditions

Size

Total Lines	3
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	3
nop	3
dl	0
loc	3
rs	10
c	0
b	0
f	0

"""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 as_list(self):
        """Return the hits as an ordered sequence of 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 filter(self, subject_index, limit=None, threshold=0.0):
        """Return a subset of the hits, filtered by the given limit and
        score threshold, as another SuggestionResult object."""
        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 ListSuggestionResult(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 as_list(self):
        if self._lsr is None:
            self._lsr = self._vector_to_list_suggestion()
        return self._lsr.as_list()

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

    def filter(self, subject_index, limit=None, threshold=0.0):
        mask = self._vector > threshold
        deprecated_ids = subject_index.deprecated_ids()
        if limit is not None:
            limit_mask = np.zeros_like(self._vector, dtype=bool)
            deprecated_set = set(deprecated_ids)
            top_k_subjects = itertools.islice(
                (subj for subj in self.subject_order if subj not in deprecated_set),
                limit,
            )
            limit_mask[list(top_k_subjects)] = True
            mask = mask & limit_mask
        else:
            deprecated_mask = np.ones_like(self._vector, dtype=bool)
            deprecated_mask[deprecated_ids] = False
            mask = mask & deprecated_mask
        vsr = VectorSuggestionResult(self._vector * mask)
        return ListSuggestionResult(vsr.as_list())

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


class ListSuggestionResult(SuggestionResult):
    """SuggestionResult implementation based primarily on lists of hits."""

    def __init__(self, hits):
        self._list = [self._enforce_score_range(hit) for hit in hits if hit.score > 0.0]
        self._vector = None

    @staticmethod
    def _enforce_score_range(hit):
        if hit.score > 1.0:
            return hit._replace(score=1.0)
        return hit

    def _list_to_vector(self, size, destination):
        if destination is None:
            destination = np.zeros(size, dtype=np.float32)

        for hit in self._list:
            if hit.subject_id is not None:
                destination[hit.subject_id] = hit.score
        return destination

    def as_list(self):
        return self._list

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

    def filter(self, subject_index, limit=None, threshold=0.0):
        hits = sorted(self._list, key=lambda hit: hit.score, reverse=True)
        filtered_hits = [
            hit
            for hit in hits
            if hit.score >= threshold and hit.score > 0.0 and hit.subject_id is not None
        ]
        if limit is not None:
            filtered_hits = filtered_hits[:limit]
        return ListSuggestionResult(filtered_hits)

    def __len__(self):
        return len(self._list)


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 as_list(self):
        _, cols = self._array[[self._idx], :].nonzero()
        suggestions = [
            SubjectSuggestion(subject_id=col, score=float(self._array[self._idx, col]))
            for col in cols
        ]
        return 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 filter(self, subject_index, limit=None, threshold=0.0):
        lsr = ListSuggestionResult(self.as_list())
        return lsr.filter(subject_index, limit, threshold)

    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, vocab_size):
        """Create a new SuggestionBatch from a sequence of SuggestionResult objects."""

        # create a dok_array for fast construction
        ar = dok_array((len(suggestion_results), vocab_size), dtype=np.float32)
        for idx, result in enumerate(suggestion_results):
            for suggestion in result.as_list():
                ar[idx, suggestion.subject_id] = suggestion.score
        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))


1			"""Representing suggested subjects."""
2
3			import abc
4			import collections
5			import itertools
6
7			import numpy as np
8			from scipy.sparse import dok_array
9
10			SubjectSuggestion = collections.namedtuple("SubjectSuggestion", "subject_id score")
11			WeightedSuggestionsBatch = collections.namedtuple(
12			"WeightedSuggestionsBatch", "hit_sets weight subjects"
13			)
14
15
16			def filter_suggestion(preds, limit=None, threshold=0.0):
17			"""filter a 2D sparse suggestion array (csr_array), retaining only the
18			top K suggestions with a score above or equal to the threshold for each
19			individual prediction; the rest will be left as zeros"""
20
21			filtered = dok_array(preds.shape, dtype=np.float32)
22			for row in range(preds.shape[0]):
23			arow = preds.getrow(row)
24			top_k = arow.data.argsort()[::-1]
25			if limit is not None:
26			top_k = top_k[:limit]
27			for idx in top_k:
28			val = arow.data[idx]
29			if val < threshold:
30			break
31			filtered[row, arow.indices[idx]] = val
32			return filtered.tocsr()
33
34
35			class SuggestionResult(metaclass=abc.ABCMeta):
36			"""Abstract base class for a set of hits returned by an analysis
37			operation."""
38
39			@abc.abstractmethod
40			def as_list(self):
41			"""Return the hits as an ordered sequence of SubjectSuggestion objects,
42			highest scores first."""
43			pass # pragma: no cover
44
45			@abc.abstractmethod
46			def as_vector(self, size, destination=None):
47			"""Return the hits as a one-dimensional score vector of given size.
48			If destination array is given (not None) it will be used, otherwise a
49			new array will be created."""
50			pass # pragma: no cover
51
52			@abc.abstractmethod
53			def filter(self, subject_index, limit=None, threshold=0.0):
54			"""Return a subset of the hits, filtered by the given limit and
55			score threshold, as another SuggestionResult object."""
56			pass # pragma: no cover
57
58			@abc.abstractmethod
59			def __len__(self):
60			"""Return the number of hits with non-zero scores."""
61			pass # pragma: no cover
62
63
64			class VectorSuggestionResult(SuggestionResult):
65			"""SuggestionResult implementation based primarily on NumPy vectors."""
66
67			def __init__(self, vector):
68			vector_f32 = vector.astype(np.float32)
69			# limit scores to the range 0.0 .. 1.0
70			self._vector = np.minimum(np.maximum(vector_f32, 0.0), 1.0)
71			self._subject_order = None
72			self._lsr = None
73
74			def _vector_to_list_suggestion(self):
75			hits = []
76			for subject_id in self.subject_order:
77			score = self._vector[subject_id]
78			if score <= 0.0:
79			break # we can skip the remaining ones
80			hits.append(SubjectSuggestion(subject_id=subject_id, score=float(score)))
81			return ListSuggestionResult(hits)
82
83			@property
84			def subject_order(self):
85			if self._subject_order is None:
86			self._subject_order = np.argsort(self._vector)[::-1]
87			return self._subject_order
88
89			def as_list(self):
90			if self._lsr is None:
91			self._lsr = self._vector_to_list_suggestion()
92			return self._lsr.as_list()
93
94			def as_vector(self, size, destination=None):
95			if destination is not None:
96			np.copyto(destination, self._vector)
97			return destination
98			return self._vector
99
100			def filter(self, subject_index, limit=None, threshold=0.0):
101			mask = self._vector > threshold
102			deprecated_ids = subject_index.deprecated_ids()
103			if limit is not None:
104			limit_mask = np.zeros_like(self._vector, dtype=bool)
105			deprecated_set = set(deprecated_ids)
106			top_k_subjects = itertools.islice(
107			(subj for subj in self.subject_order if subj not in deprecated_set),
108			limit,
109			)
110			limit_mask[list(top_k_subjects)] = True
111			mask = mask & limit_mask
112			else:
113			deprecated_mask = np.ones_like(self._vector, dtype=bool)
114			deprecated_mask[deprecated_ids] = False
115			mask = mask & deprecated_mask
116			vsr = VectorSuggestionResult(self._vector * mask)
117			return ListSuggestionResult(vsr.as_list())
118
119			def __len__(self):
120			return (self._vector > 0.0).sum()
121
122
123			class ListSuggestionResult(SuggestionResult):
124			"""SuggestionResult implementation based primarily on lists of hits."""
125
126			def __init__(self, hits):
127			self._list = [self._enforce_score_range(hit) for hit in hits if hit.score > 0.0]
128			self._vector = None
129
130			@staticmethod
131			def _enforce_score_range(hit):
132			if hit.score > 1.0:
133			return hit._replace(score=1.0)
134			return hit
135
136			def _list_to_vector(self, size, destination):
137			if destination is None:
138			destination = np.zeros(size, dtype=np.float32)
139
140			for hit in self._list:
141			if hit.subject_id is not None:
142			destination[hit.subject_id] = hit.score
143			return destination
144
145			def as_list(self):
146			return self._list
147
148			def as_vector(self, size, destination=None):
149			if self._vector is None:
150			self._vector = self._list_to_vector(size, destination)
151			return self._vector
152
153			def filter(self, subject_index, limit=None, threshold=0.0):
154			hits = sorted(self._list, key=lambda hit: hit.score, reverse=True)
155			filtered_hits = [
156			hit
157			for hit in hits
158			if hit.score >= threshold and hit.score > 0.0 and hit.subject_id is not None
159			]
160			if limit is not None:
161			filtered_hits = filtered_hits[:limit]
162			return ListSuggestionResult(filtered_hits)
163
164			def __len__(self):
165			return len(self._list)
166
167
168			class SparseSuggestionResult(SuggestionResult):
169			"""SuggestionResult implementation backed by a single row of a sparse array."""
170
171			def __init__(self, array, idx):
172			self._array = array
173			self._idx = idx
174
175			def as_list(self):
176			_, cols = self._array[[self._idx], :].nonzero()
177			suggestions = [
178			SubjectSuggestion(subject_id=col, score=float(self._array[self._idx, col]))
179			for col in cols
180			]
181			return sorted(
182			suggestions, key=lambda suggestion: suggestion.score, reverse=True
183			)
184
185			def as_vector(self, size, destination=None):
186			if destination is not None:
187			print("as_vector called with destination not None")
188			return None
189			return self._array[[self._idx], :].toarray()[0]
190
191			def filter(self, subject_index, limit=None, threshold=0.0):
192			lsr = ListSuggestionResult(self.as_list())
193			return lsr.filter(subject_index, limit, threshold)
194
195			def __len__(self):
196			_, cols = self._array[[self._idx], :].nonzero()
197			return len(cols)
198
199
200			class SuggestionBatch:
201			"""Subject suggestions for a batch of documents."""
202
203			def __init__(self, array):
204			"""Create a new SuggestionBatch from a csr_array"""
205			self.array = array
206
207			@classmethod
208			def from_sequence(cls, suggestion_results, vocab_size):
209			"""Create a new SuggestionBatch from a sequence of SuggestionResult objects."""
210
211			# create a dok_array for fast construction
212			ar = dok_array((len(suggestion_results), vocab_size), dtype=np.float32)
213			for idx, result in enumerate(suggestion_results):
214			for suggestion in result.as_list():
215			ar[idx, suggestion.subject_id] = suggestion.score
216			return cls(ar.tocsr())
217
218			def filter(self, limit=None, threshold=0.0):
219			"""Return a subset of the hits, filtered by the given limit and
220			score threshold, as another SuggestionBatch object."""
221
222			return SuggestionBatch(filter_suggestion(self.array, limit, threshold))
223
224			def __getitem__(self, idx):
225			if idx < 0 or idx >= len(self):
226			raise IndexError
227			return SparseSuggestionResult(self.array, idx)
228
229			def __len__(self):
230			return self.array.shape[0]
231
232
233			class SuggestionResults:
234			"""Subject suggestions for a potentially very large number of documents."""
235
236			def __init__(self, batches):
237			"""Initialize a new SuggestionResults from an iterable that provides
238			SuggestionBatch objects."""
239
240			self.batches = batches
241
242			def filter(self, limit=None, threshold=0.0):
243			"""Return a view of these suggestions, filtered by the given limit
244			and/or threshold, as another SuggestionResults object."""
245
246			return SuggestionResults(
247			(batch.filter(limit, threshold) for batch in self.batches)
248			)
249
250			def __iter__(self):
251			return iter(itertools.chain.from_iterable(self.batches))
252

NatLibFi / Annif

Push — issue678-refactor-suggestionre... ( d250de...96588c )

annif.suggestion.SparseSuggestionResult.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

annif.suggestion.SparseSuggestionResult.init() A