AnnifProject

Complexity

Total Complexity

22

Size/Duplication

Total Lines	85
Duplicated Lines	0 %

Importance

Changes	7
Bugs	1	Features	0

8 Methods

Rating	Name	Duplication	Size	Complexity
A	_initialize_backends()	0	12	3
A	__init__()	0	4	1
A	_merge_hits()	0	9	3
B	_analyze_with_backends()	0	14	6
A	analyze()	0	12	1
A	dump()	0	6	2
A	load_subjects()	0	6	2
A	_filter_hits()	0	12	4

"""Project management functionality for Annif"""

import collections
import configparser
import logging

The import logging seems to be unused.

from flask import current_app
import annif
import annif.hit
import annif.backend
from annif import logger


class AnnifProject:

The variable __class__ seems to be unused.

    """Class representing the configuration of a single Annif project."""

    def __init__(self, project_id, config):
        self.project_id = project_id
        self.language = config['language']
        self.backends = self._initialize_backends(config['backends'])

    def _initialize_backends(self, backends_configuration):

This method could be written as a function/class method.

        backends = []
        for backenddef in backends_configuration.split(','):
            bedefs = backenddef.strip().split(':')
            backend_id = bedefs[0]
            if len(bedefs) > 1:
                weight = float(bedefs[1])
            else:
                weight = 1.0
            backend = annif.backend.get_backend(backend_id)
            backends.append((backend, weight))
        return backends

    def _analyze_with_backends(self, text, backend_params):
        if backend_params is None:
            backend_params = {}
        hits_by_uri = collections.defaultdict(list)
        for backend, weight in self.backends:
            beparams = backend_params.get(backend.backend_id, {})
            hits = [hit for hit in backend.analyze(text, params=beparams)
                    if hit.score > 0.0]
            logger.debug(
                'Got {} hits from backend {}'.format(

Use formatting in logging functions and pass the parameters as arguments

                    len(hits), backend.backend_id))
            for hit in hits:
                hits_by_uri[hit.uri].append((hit.score * weight, hit))
        return hits_by_uri

    @classmethod
    def _merge_hits(cls, hits_by_uri):
        merged_hits = []
        for score_hits in hits_by_uri.values():
            total = sum([sh[0] for sh in score_hits])
            hit = annif.hit.AnalysisHit(
                score_hits[0][1].uri, score_hits[0][1].label, total)
            merged_hits.append(hit)
        return merged_hits

    @classmethod
    def _filter_hits(cls, hits, limit, threshold):
        hits.sort(key=lambda hit: hit.score, reverse=True)
        hits = hits[:limit]
        logger.debug(
            '{} hits after applying limit {}'.format(

Use formatting in logging functions and pass the parameters as arguments

                len(hits), limit))
        hits = [hit for hit in hits if hit.score >= threshold]
        logger.debug(
            '{} hits after applying threshold {}'.format(

Use formatting in logging functions and pass the parameters as arguments

                len(hits), threshold))
        return hits

    def analyze(self, text, limit=10, threshold=0.0, backend_params=None):
        """Analyze the given text by passing it to backends and joining the
        results. Returns a list of AnalysisHit objects ordered by decreasing
        score. The limit parameter defines the maximum number of hits to return.
        Only hits whose score is over the threshold are returned."""

        logger.debug('Analyzing text "{}..." (len={})'.format(

Use formatting in logging functions and pass the parameters as arguments

            text[:20], len(text)))
        hits_by_uri = self._analyze_with_backends(text, backend_params)
        merged_hits = self._merge_hits(hits_by_uri)
        logger.debug('{} hits after merging'.format(len(merged_hits)))

Use formatting in logging functions and pass the parameters as arguments

        return self._filter_hits(merged_hits, limit, threshold)

    def load_subjects(self, subjects):

This method should have a docstring.

        for backend, weight in self.backends:

The variable weight seems to be unused.

            logger.debug(
                'Loading subjects for backend {}'.format(

Use formatting in logging functions and pass the parameters as arguments

                    backend.backend_id))
            backend.load_subjects(subjects)

    def dump(self):
        """return this project as a dict"""
        return {'project_id': self.project_id,
                'language': self.language,
                'backends': [{'backend_id': be[0].backend_id,
                              'weight': be[1]} for be in self.backends]
                }

Wrong continued indentation (remove 1 space).

def get_projects():
    """return the available projects as a dict of project_id -> AnnifProject"""
    projects_file = current_app.config['PROJECTS_FILE']
    config = configparser.ConfigParser()
    with open(projects_file) as projf:
        config.read_file(projf)

    # create AnnifProject objects from the configuration file
    projects = {}
    for project_id in config.sections():
        projects[project_id] = AnnifProject(project_id, config[project_id])
    return projects


def get_project(project_id):
    """return the definition of a single Project by project_id"""
    projects = get_projects()
    try:
        return projects[project_id]
    except KeyError:
        raise ValueError("No such project {}".format(project_id))