annif.project

Complexity

Total Complexity

59

Size/Duplication

Total Lines	345
Duplicated Lines	0 %

Importance

Changes

0

25 Methods

Rating	Name	Duplication	Size	Complexity
A	AnnifProject._init_access()	0	8	2
A	AnnifProject.subjects()	0	3	1
A	AnnifProject.vocab()	0	9	3
A	AnnifProject._initialize_subjects()	0	8	2
A	AnnifProject.remove_model_data()	0	10	2
A	AnnifProject._initialize_backend()	0	9	3
A	AnnifProject._suggest_with_backend()	0	9	2
A	AnnifProject.analyzer()	0	11	3
A	AnnifProject.backend()	0	18	4
A	AnnifProject.initialize()	0	11	1
A	AnnifProject.dump()	0	8	1
A	AnnifProject.train()	0	8	3
A	AnnifProject.suggest()	0	13	3
A	AnnifProject.learn()	0	13	3
A	AnnifProject.__init__()	0	11	1
A	AnnifProject._initialize_analyzer()	0	7	2
A	AnnifRegistry.get_projects()	0	8	1
A	AnnifRegistry.get_project()	0	8	2
A	AnnifProject._get_info()	0	8	3
A	AnnifProject.modification_time()	0	3	1
A	AnnifProject.is_trained()	0	3	1
B	AnnifRegistry._create_projects()	0	27	5
A	AnnifRegistry.__init__()	0	7	3
A	ProjectSuggestMap.suggest()	0	6	1
A	ProjectSuggestMap.__init__()	0	6	1

"""Project management functionality for Annif"""

import collections
import configparser
import enum
import os.path
from flask import current_app
from shutil import rmtree
import annif
import annif.analyzer
import annif.corpus
import annif.suggestion
import annif.backend
import annif.util
import annif.vocab
from annif.datadir import DatadirMixin
from annif.exception import AnnifException, ConfigurationException, \
    NotSupportedException, NotInitializedException

logger = annif.logger


class Access(enum.IntEnum):
    """Enumeration of access levels for projects"""
    private = 1
    hidden = 2
    public = 3


class AnnifProject(DatadirMixin):
    """Class representing the configuration of a single Annif project."""

    # defaults for uninitialized instances
    _analyzer = None
    _backend = None
    _vocab = None
    initialized = False

    # default values for configuration settings
    DEFAULT_ACCESS = 'public'

    def __init__(self, project_id, config, datadir, registry):
        DatadirMixin.__init__(self, datadir, 'projects', project_id)
        self.project_id = project_id
        self.name = config.get('name', project_id)
        self.language = config['language']
        self.analyzer_spec = config.get('analyzer', None)
        self.vocab_id = config.get('vocab', None)
        self.config = config
        self._base_datadir = datadir
        self.registry = registry
        self._init_access()

    def _init_access(self):
        access = self.config.get('access', self.DEFAULT_ACCESS)
        try:
            self.access = getattr(Access, access)
        except AttributeError:
            raise ConfigurationException(
                "'{}' is not a valid access setting".format(access),
                project_id=self.project_id)

    def _initialize_analyzer(self):
        if not self.analyzer_spec:
            return  # not configured, so assume it's not needed
        analyzer = self.analyzer
        logger.debug("Project '%s': initialized analyzer: %s",
                     self.project_id,
                     str(analyzer))

    def _initialize_subjects(self):
        try:
            subjects = self.subjects
            logger.debug("Project '%s': initialized subjects: %s",
                         self.project_id,
                         str(subjects))
        except AnnifException as err:
            logger.warning(err.format_message())

    def _initialize_backend(self):
        logger.debug("Project '%s': initializing backend", self.project_id)
        try:
            if not self.backend:
                logger.debug("Cannot initialize backend: does not exist")
                return
            self.backend.initialize()
        except AnnifException as err:
            logger.warning(err.format_message())

    def initialize(self):
        """initialize this project and its backend so that they are ready to
        be used"""

        logger.debug("Initializing project '%s'", self.project_id)

        self._initialize_analyzer()
        self._initialize_subjects()
        self._initialize_backend()

        self.initialized = True

    def _suggest_with_backend(self, text, backend_params):
        if backend_params is None:
            backend_params = {}
        beparams = backend_params.get(self.backend.backend_id, {})
        hits = self.backend.suggest(text, beparams)
        logger.debug(
            'Got %d hits from backend %s',
            len(hits), self.backend.backend_id)
        return hits

    @property
    def analyzer(self):
        if self._analyzer is None:
            if self.analyzer_spec:
                self._analyzer = annif.analyzer.get_analyzer(
                    self.analyzer_spec)
            else:
                raise ConfigurationException(
                    "analyzer setting is missing (and needed by the backend)",
                    project_id=self.project_id)
        return self._analyzer

    @property
    def backend(self):
        if self._backend is None:
            if 'backend' not in self.config:
                raise ConfigurationException(
                    "backend setting is missing", project_id=self.project_id)
            backend_id = self.config['backend']
            try:
                backend_class = annif.backend.get_backend(backend_id)
                self._backend = backend_class(
                    backend_id, config_params=self.config,
                    project=self)
            except ValueError:
                logger.warning(
                    "Could not create backend %s, "
                    "make sure you've installed optional dependencies",
                    backend_id)
        return self._backend

    @property
    def vocab(self):
        if self._vocab is None:
            if self.vocab_id is None:
                raise ConfigurationException("vocab setting is missing",
                                             project_id=self.project_id)
            self._vocab = annif.vocab.AnnifVocabulary(self.vocab_id,
                                                      self._base_datadir)
        return self._vocab

    @property
    def subjects(self):
        return self.vocab.subjects

    def _get_info(self, key):
        try:
            be = self.backend
            if be is not None:
                return getattr(be, key)
        except AnnifException as err:
            logger.warning(err.format_message())
            return None

    @property
    def is_trained(self):
        return self._get_info('is_trained')

    @property
    def modification_time(self):
        return self._get_info('modification_time')

    def suggest(self, text, backend_params=None):
        """Suggest subjects the given text by passing it to the backend. Returns a
        list of SubjectSuggestion objects ordered by decreasing score."""
        if not self.is_trained:
            if self.is_trained is None:
                logger.warn('Could not get train state information.')
            else:
                raise NotInitializedException('Project is not trained.')
        logger.debug('Suggesting subjects for text "%s..." (len=%d)',
                     text[:20], len(text))
        hits = self._suggest_with_backend(text, backend_params)
        logger.debug('%d hits from backend', len(hits))
        return hits

    def train(self, corpus, backend_params=None):
        """train the project using documents from a metadata source"""
        if corpus != 'cached':
            corpus.set_subject_index(self.subjects)
        if backend_params is None:
            backend_params = {}
        beparams = backend_params.get(self.backend.backend_id, {})
        self.backend.train(corpus, beparams)

    def learn(self, corpus, backend_params=None):
        """further train the project using documents from a metadata source"""
        corpus.set_subject_index(self.subjects)
        if backend_params is None:
            backend_params = {}
        beparams = backend_params.get(self.backend.backend_id, {})
        if isinstance(
                self.backend,
                annif.backend.backend.AnnifLearningBackend):
            self.backend.learn(corpus, beparams)
        else:
            raise NotSupportedException("Learning not supported by backend",
                                        project_id=self.project_id)

    def dump(self):
        """return this project as a dict"""
        return {'project_id': self.project_id,
                'name': self.name,
                'language': self.language,
                'backend': {'backend_id': self.config.get('backend')},
                'is_trained': self.is_trained,
                'modification_time': self.modification_time
                }

    def remove_model_data(self):
        """remove the data of this project"""
        datadir_path = self._datadir_path
        if os.path.isdir(datadir_path):
            rmtree(datadir_path)
            logger.info('Removed model data for project {}.'
                        .format(self.project_id))
        else:
            logger.warning('No model data to remove for project {}.'
                           .format(self.project_id))


class AnnifRegistry:
    """Class that keeps track of the Annif projects"""

    # Note: The individual projects are stored in a shared static variable,
    # keyed by the "registry ID" which is unique to the registry instance.
    # This is done to make it possible to serialize AnnifRegistry instances
    # without including the potentially huge project objects (which contain
    # backends with large models, vocabularies with lots of concepts etc).
    # Serialized AnnifRegistry instances can then be passed between
    # processes when using the multiprocessing module.
    _projects = {}

    def __init__(self, projects_file, datadir, init_projects):
        self._rid = id(self)
        self._projects[self._rid] = \
            self._create_projects(projects_file, datadir)
        if init_projects:
            for project in self._projects[self._rid].values():
                project.initialize()

    def _create_projects(self, projects_file, datadir):
        if not os.path.exists(projects_file):
            logger.warning(
                'Project configuration file "%s" is missing. ' +
                'Please provide one. You can set the path to the project ' +
                'configuration file using the ANNIF_PROJECTS environment ' +
                'variable or the command-line option "--projects".',
                projects_file)
            return {}

        config = configparser.ConfigParser()
        config.optionxform = annif.util.identity
        with open(projects_file, encoding='utf-8-sig') as projf:
            try:
                config.read_file(projf)
            except (configparser.DuplicateOptionError,
                    configparser.DuplicateSectionError) as err:
                raise ConfigurationException(err)

        # create AnnifProject objects from the configuration file
        projects = collections.OrderedDict()
        for project_id in config.sections():
            projects[project_id] = AnnifProject(project_id,
                                                config[project_id],
                                                datadir,
                                                self)
        return projects

    def get_projects(self, min_access=Access.private):
        """Return the available projects as a dict of project_id ->
        AnnifProject. The min_access parameter may be used to set the minimum
        access level required for the returned projects."""

        return {project_id: project
                for project_id, project in self._projects[self._rid].items()
                if project.access >= min_access}

    def get_project(self, project_id, min_access=Access.private):
        """return the definition of a single Project by project_id"""

        projects = self.get_projects(min_access)
        try:
            return projects[project_id]
        except KeyError:
            raise ValueError("No such project {}".format(project_id))


def initialize_projects(app):
    projects_file = app.config['PROJECTS_FILE']
    datadir = app.config['DATADIR']
    init_projects = app.config['INITIALIZE_PROJECTS']
    app.annif_registry = AnnifRegistry(projects_file, datadir, init_projects)


def get_projects(min_access=Access.private):
    """Return the available projects as a dict of project_id ->
    AnnifProject. The min_access parameter may be used to set the minimum
    access level required for the returned projects."""
    if not hasattr(current_app, 'annif_registry'):
        initialize_projects(current_app)

    return current_app.annif_registry.get_projects(min_access)


def get_project(project_id, min_access=Access.private):
    """return the definition of a single Project by project_id"""

    projects = get_projects(min_access)
    try:
        return projects[project_id]
    except KeyError:
        raise ValueError("No such project {}".format(project_id))


class ProjectSuggestMap:
    """A utility class that can be used to wrap a project and provide a
    mapping method that converts Document objects to suggestions. Intended
    to be used with the multiprocessing module."""

    def __init__(self, project, backend_params, limit, threshold):
        self.project_id = project.project_id
        self.registry = project.registry
        self.backend_params = backend_params
        self.limit = limit
        self.threshold = threshold

    def suggest(self, doc):
        project = self.registry.get_project(self.project_id)
        hits = project.suggest(doc.text, self.backend_params)
        filtered_hits = hits.filter(
            project.subjects, self.limit, self.threshold)
        return (filtered_hits, doc.uris, doc.labels)