TFIDFBackend

Complexity

Total Complexity

26

Size/Duplication

Total Lines	127
Duplicated Lines	0 %

Importance

Changes	6
Bugs	0	Features	0

11 Methods

Rating	Name	Duplication	Size	Complexity
A	_initialize_analyzer()	0	4	2
A	_initialize_tfidf()	0	5	2
A	_initialize_dictionary()	0	5	2
A	load_subjects()	0	18	2
A	_initialize_subjects()	0	5	2
A	_analyze_chunks()	0	9	3
A	_initialize_index()	0	5	2
A	_atomic_save()	0	9	2
A	initialize()	0	6	1
B	_merge_chunk_results()	0	22	6
A	_analyze()	0	16	2

"""Backend that returns most similar subjects based on similarity in sparse
TF-IDF normalized bag-of-words vector space"""

import collections
import glob
import os
import os.path
import tempfile
import gensim.corpora
import gensim.models
import gensim.similarities
import annif.analyzer
import annif.corpus
from annif.hit import AnalysisHit
from . import backend


class VectorCorpus:

The variable __class__ seems to be unused.

    """A class that wraps a subject corpus so it can be iterated as lists of
    vectors, by using a dictionary to map words to integers."""

    def __init__(self, corpus, dictionary, analyzer):
        self.corpus = corpus
        self.dictionary = dictionary
        self.analyzer = analyzer

    def __iter__(self):
        """Iterate through the subject directory, yielding vectors that are
        derived from subjects using the given analyzer and dictionary."""

        for subject in self.corpus:
            yield self.dictionary.doc2bow(
                self.analyzer.tokenize_words(subject.text))


class SubjectIndex:

The variable __class__ seems to be unused.

    """A class that remembers the associations between integers subject IDs
    and their URIs and labels."""

    def __init__(self, corpus):
        """Initialize the subject index from a subject corpus."""
        self._uris = []
        self._labels = []
        for subject_id, subject in enumerate(corpus):

The variable subject_id seems to be unused.

            self._uris.append(subject.uri)
            self._labels.append(subject.label)

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

    def __getitem__(self, subject_id):
        return (self._uris[subject_id], self._labels[subject_id])

    def save(self, path):
        """Save this subject index into a file."""

        with open(path, 'w') as subjfile:
            for subject_id in range(len(self)):
                line = "<{}>\t{}".format(
                    self._uris[subject_id], self._labels[subject_id])
                print(line, file=subjfile)

    @classmethod
    def load(cls, path):
        """Load a subject index from a file and return it."""

        def file_as_corpus(path):

This function should have a docstring.

            with open(path) as subjfile:
                for line in subjfile:
                    uri, label = line.strip().split(None, 1)
                    uri = uri[1:-1]
                    yield annif.corpus.Subject(uri, label, None)

        return cls(file_as_corpus(path))


class TFIDFBackend(backend.AnnifBackend):

This class should have a docstring.

The variable __class__ seems to be unused.

    name = "tfidf"

    # top K subjects per chunk to consider
    MAX_CHUNK_SUBJECTS = 100

    # defaults for uninitialized instances
    _subjects = None
    _analyzer = None
    _dictionary = None
    _tfidf = None
    _index = None

    def _atomic_save(self, obj, dirname, filename):
        tempfd, tempfilename = tempfile.mkstemp(prefix=filename, dir=dirname)
        os.close(tempfd)
        self.debug('saving {} to temporary file {}'.format(obj, tempfilename))
        obj.save(tempfilename)
        for fn in glob.glob(tempfilename + '*'):

The name fn does not conform to the variable naming conventions ((([a-z][a-z0-9_]{2,30})|(_[a-z0-9_]*))$).

            newname = fn.replace(tempfilename, os.path.join(dirname, filename))
            self.debug('renaming temporary file {} to {}'.format(fn, newname))
            os.rename(fn, newname)

    def _initialize_subjects(self):
        if self._subjects is None:
            path = os.path.join(self._get_datadir(), 'subjects')
            self.debug('loading subjects from {}'.format(path))
            self._subjects = SubjectIndex.load(path)

    def _initialize_analyzer(self):
        if self._analyzer is None:
            self._analyzer = annif.analyzer.get_analyzer(
                self.params['analyzer'])

    def _initialize_dictionary(self):
        if self._dictionary is None:
            path = os.path.join(self._get_datadir(), 'dictionary')
            self.debug('loading dictionary from {}'.format(path))
            self._dictionary = gensim.corpora.Dictionary.load(path)

    def _initialize_tfidf(self):
        if self._tfidf is None:
            path = os.path.join(self._get_datadir(), 'tfidf')
            self.debug('loading TF-IDF model from {}'.format(path))
            self._tfidf = gensim.models.TfidfModel.load(path)

    def _initialize_index(self):
        if self._index is None:
            path = os.path.join(self._get_datadir(), 'index')
            self.debug('loading similarity index from {}'.format(path))
            self._index = gensim.similarities.SparseMatrixSimilarity.load(path)

    def initialize(self):

This method should have a docstring.

        self._initialize_subjects()
        self._initialize_analyzer()
        self._initialize_dictionary()
        self._initialize_tfidf()
        self._initialize_index()

    def load_subjects(self, subjects):

Parameters differ from overridden 'load_subjects' method

        self.info('Backend {}: creating subject index'.format(self.backend_id))
        self._subjects = SubjectIndex(subjects)
        self._atomic_save(self._subjects, self._get_datadir(), 'subjects')
        self._initialize_analyzer()
        self.info('creating dictionary')
        self._dictionary = gensim.corpora.Dictionary(
            (self._analyzer.tokenize_words(subject.text)
             for subject in subjects))
        self._atomic_save(self._dictionary, self._get_datadir(), 'dictionary')
        veccorpus = VectorCorpus(subjects, self._dictionary, self._analyzer)
        self.info('creating TF-IDF model')
        self._tfidf = gensim.models.TfidfModel(veccorpus)
        self._atomic_save(self._tfidf, self._get_datadir(), 'tfidf')
        self.info('creating similarity index')
        self._index = gensim.similarities.SparseMatrixSimilarity(
            self._tfidf[veccorpus], num_features=len(self._dictionary))
        self._atomic_save(self._index, self._get_datadir(), 'index')

    def _analyze_chunks(self, chunks):
        results = []
        for docsim in self._index[chunks]:
            sims = sorted(
                enumerate(docsim),
                key=lambda item: item[1],
                reverse=True)
            results.append(sims[:self.MAX_CHUNK_SUBJECTS])
        return results

    def _merge_chunk_results(self, chunk_results):
        subject_scores = collections.defaultdict(float)
        for result in chunk_results:
            for subject_id, score in result:
                subject_scores[subject_id] += score
        best_subjects = sorted([(score,
                                 subject_id) for subject_id,
                                score in subject_scores.items()],
                               reverse=True)
        limit = int(self.params['limit'])
        results = []
        for score, subject_id in best_subjects[:limit]:
            if score <= 0.0:
                continue
            subject = self._subjects[subject_id]
            results.append(
                AnalysisHit(
                    subject[0],
                    subject[1],
                    score /
                    len(chunk_results)))
        return results

    def _analyze(self, text, params):
        self.initialize()
        self.debug('Analyzing text "{}..." (len={})'.format(
            text[:20], len(text)))
        sentences = self._analyzer.tokenize_sentences(text)
        self.debug('Found {} sentences'.format(len(sentences)))
        chunksize = int(params['chunksize'])
        chunks = []  # chunks represented as TF-IDF normalized vectors
        for i in range(0, len(sentences), chunksize):
            chunktext = ' '.join(sentences[i:i + chunksize])
            chunkbow = self._dictionary.doc2bow(
                self._analyzer.tokenize_words(chunktext))
            chunks.append(self._tfidf[chunkbow])
        self.debug('Split sentences into {} chunks'.format(len(chunks)))
        chunk_results = self._analyze_chunks(chunks)
        return self._merge_chunk_results(chunk_results)