annif.backend.backend

Complexity

Total Complexity

20

Size/Duplication

Total Lines	131
Duplicated Lines	0 %

Importance

Changes

0

18 Methods

Rating	Name	Duplication	Size	Complexity
A	AnnifBackend.__init__()	0	8	1
A	AnnifBackend.default_params()	0	2	1
A	AnnifBackend.is_trained()	0	3	1
A	AnnifBackend.params()	0	6	1
A	AnnifBackend._get_backend_params()	0	5	2
A	AnnifBackend._suggest()	0	5	1
A	AnnifBackend.train()	0	4	1
A	AnnifBackend.initialize()	0	6	1
A	AnnifBackend._train()	0	4	1
A	AnnifBackend.modification_time()	0	10	2
A	AnnifBackend.suggest()	0	6	1
A	AnnifBackend.suggest_batch()	0	6	1
A	AnnifBackend._suggest_batch()	0	7	1
A	AnnifLearningBackend.learn()	0	4	1
A	AnnifBackend.debug()	0	3	1
A	AnnifBackend.info()	0	3	1
A	AnnifBackend.warning()	0	3	1
A	AnnifLearningBackend._learn()	0	5	1

"""Common functionality for backends."""

import abc
import os.path
from datetime import datetime, timezone
from glob import glob

from annif import logger


class AnnifBackend(metaclass=abc.ABCMeta):
    """Base class for Annif backends that perform analysis. The
    non-implemented methods should be overridden in subclasses."""

    name = None

    DEFAULT_PARAMETERS = {"limit": 100}

    def __init__(self, backend_id, config_params, project):
        """Initialize backend with specific parameters. The
        parameters are a dict. Keys and values depend on the specific
        backend type."""
        self.backend_id = backend_id
        self.config_params = config_params
        self.project = project
        self.datadir = project.datadir

    def default_params(self):
        return self.DEFAULT_PARAMETERS

    @property
    def params(self):
        params = {}
        params.update(self.default_params())
        params.update(self.config_params)
        return params

    @property
    def is_trained(self):
        return bool(glob(os.path.join(self.datadir, "*")))

    @property
    def modification_time(self):
        mtimes = [
            datetime.utcfromtimestamp(os.path.getmtime(p))
            for p in glob(os.path.join(self.datadir, "*"))
        ]
        most_recent = max(mtimes, default=None)
        if most_recent is None:
            return None
        return most_recent.replace(tzinfo=timezone.utc)

    def _get_backend_params(self, params):
        backend_params = dict(self.params)
        if params is not None:
            backend_params.update(params)
        return backend_params

    def _train(self, corpus, params, jobs=0):
        """This method can be overridden by backends. It implements
        the train functionality, with pre-processed parameters."""
        pass  # default is to do nothing, subclasses may override

    def train(self, corpus, params=None, jobs=0):
        """Train the model on the given document or subject corpus."""
        beparams = self._get_backend_params(params)
        return self._train(corpus, params=beparams, jobs=jobs)

    def initialize(self, parallel=False):
        """This method can be overridden by backends. It should cause the
        backend to pre-load all data it needs during operation.
        If parallel is True, the backend should expect to be used for
        parallel operation."""
        pass

    @abc.abstractmethod
    def _suggest(self, text, params):
        """This method should implemented by backends. It implements
        the suggest functionality, with pre-processed parameters."""
        pass  # pragma: no cover

    def _suggest_batch(self, corpus, transform, params):
        """This method can be implemented by backends to use batching of documents in
        their operations. This default implementation uses the regular suggest
        functionality."""
        return [
            self._suggest(transform.transform_text(doc.text), params)
            for doc in corpus.documents
        ]

    def suggest(self, text, params=None):
        """Suggest subjects for the input text and return a list of subjects
        represented as a list of SubjectSuggestion objects."""
        beparams = self._get_backend_params(params)
        self.initialize()
        return self._suggest(text, params=beparams)

    def suggest_batch(self, corpus, transform, params=None):
        """Suggest subjects for the input documents and return a list of subject sets
        represented as a list of SubjectSuggestion objects."""
        beparams = self._get_backend_params(params)
        self.initialize()
        return self._suggest_batch(corpus, transform, params=beparams)

    def debug(self, message):
        """Log a debug message from this backend"""
        logger.debug("Backend {}: {}".format(self.backend_id, message))

    def info(self, message):
        """Log an info message from this backend"""
        logger.info("Backend {}: {}".format(self.backend_id, message))

    def warning(self, message):
        """Log a warning message from this backend"""
        logger.warning("Backend {}: {}".format(self.backend_id, message))


class AnnifLearningBackend(AnnifBackend):
    """Base class for Annif backends that can perform online learning"""

    @abc.abstractmethod
    def _learn(self, corpus, params):
        """This method should implemented by backends. It implements the learn
        functionality, with pre-processed parameters."""
        pass  # pragma: no cover

    def learn(self, corpus, params=None):
        """Further train the model on the given document or subject corpus."""
        beparams = self._get_backend_params(params)
        return self._learn(corpus, params=beparams)