annif.project.AnnifProject.transform() - Code Metrics - Inspection of "Support for adding input-transformation operations" - NatLibFi/Annif - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#496)

by Juho

created 2021-08-09 14:32 UTC

annif.project.AnnifProject.transform() A

↳ Parent: annif.project

Complexity

Conditions

Size

Total Lines	6
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

"""Project management functionality for Annif"""

import enum
import os.path
from shutil import rmtree
import annif
import annif.transform
import annif.analyzer
import annif.corpus
import annif.suggestion
import annif.backend
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
    _transform = None
    _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.transform_spec = config.get('transform', 'pass')
        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 transform(self):
        if self._transform is None:
            self._transform = annif.transform.get_transform(
                self.transform_spec, project=self)
        return self._transform

    @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,
                                                      self.language)
        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.warning('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))
        text = self.transform.transform_text(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)
            corpus = self.transform.transform_corpus(corpus)
        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, {})
        corpus = self.transform.transform_corpus(corpus)
        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 hyperopt(self, corpus, trials, jobs, metric, results_file):
        """optimize the hyperparameters of the project using a validation
        corpus against a given metric"""
        if isinstance(
                self.backend,
                annif.backend.hyperopt.AnnifHyperoptBackend):
            optimizer = self.backend.get_hp_optimizer(corpus, metric)
            return optimizer.optimize(trials, jobs, results_file)

        raise NotSupportedException(
            "Hyperparameter optimization 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))


1			"""Project management functionality for Annif"""
2
3			import enum
4			import os.path
5			from shutil import rmtree
6			import annif
7			import annif.transform
8			import annif.analyzer
9			import annif.corpus
10			import annif.suggestion
11			import annif.backend
12			import annif.vocab
13			from annif.datadir import DatadirMixin
14			from annif.exception import AnnifException, ConfigurationException, \
15			NotSupportedException, NotInitializedException
16
17			logger = annif.logger
18
19
20			class Access(enum.IntEnum):
21			"""Enumeration of access levels for projects"""
22			private = 1
23			hidden = 2
24			public = 3
25
26
27			class AnnifProject(DatadirMixin):
28			"""Class representing the configuration of a single Annif project."""
29
30			# defaults for uninitialized instances
31			_transform = None
32			_analyzer = None
33			_backend = None
34			_vocab = None
35			initialized = False
36
37			# default values for configuration settings
38			DEFAULT_ACCESS = 'public'
39
40			def __init__(self, project_id, config, datadir, registry):
41			DatadirMixin.__init__(self, datadir, 'projects', project_id)
42			self.project_id = project_id
43			self.name = config.get('name', project_id)
44			self.language = config['language']
45			self.analyzer_spec = config.get('analyzer', None)
46			self.transform_spec = config.get('transform', 'pass')
47			self.vocab_id = config.get('vocab', None)
48			self.config = config
49			self._base_datadir = datadir
50			self.registry = registry
51			self._init_access()
52
53			def _init_access(self):
54			access = self.config.get('access', self.DEFAULT_ACCESS)
55			try:
56			self.access = getattr(Access, access)
57			except AttributeError:
58			raise ConfigurationException(
59			"'{}' is not a valid access setting".format(access),
60			project_id=self.project_id)
61
62			def _initialize_analyzer(self):
63			if not self.analyzer_spec:
64			return # not configured, so assume it's not needed
65			analyzer = self.analyzer
66			logger.debug("Project '%s': initialized analyzer: %s",
67			self.project_id,
68			str(analyzer))
69
70			def _initialize_subjects(self):
71			try:
72			subjects = self.subjects
73			logger.debug("Project '%s': initialized subjects: %s",
74			self.project_id,
75			str(subjects))
76			except AnnifException as err:
77			logger.warning(err.format_message())
78
79			def _initialize_backend(self):
80			logger.debug("Project '%s': initializing backend", self.project_id)
81			try:
82			if not self.backend:
83			logger.debug("Cannot initialize backend: does not exist")
84			return
85			self.backend.initialize()
86			except AnnifException as err:
87			logger.warning(err.format_message())
88
89			def initialize(self):
90			"""initialize this project and its backend so that they are ready to
91			be used"""
92
93			logger.debug("Initializing project '%s'", self.project_id)
94
95			self._initialize_analyzer()
96			self._initialize_subjects()
97			self._initialize_backend()
98
99			self.initialized = True
100
101			def _suggest_with_backend(self, text, backend_params):
102			if backend_params is None:
103			backend_params = {}
104			beparams = backend_params.get(self.backend.backend_id, {})
105			hits = self.backend.suggest(text, beparams)
106			logger.debug(
107			'Got %d hits from backend %s',
108			len(hits), self.backend.backend_id)
109			return hits
110
111			@property
112			def analyzer(self):
113			if self._analyzer is None:
114			if self.analyzer_spec:
115			self._analyzer = annif.analyzer.get_analyzer(
116			self.analyzer_spec)
117			else:
118			raise ConfigurationException(
119			"analyzer setting is missing (and needed by the backend)",
120			project_id=self.project_id)
121			return self._analyzer
122
123			@property
124			def transform(self):
125			if self._transform is None:
126			self._transform = annif.transform.get_transform(
127			self.transform_spec, project=self)
128			return self._transform
129
130			@property
131			def backend(self):
132			if self._backend is None:
133			if 'backend' not in self.config:
134			raise ConfigurationException(
135			"backend setting is missing", project_id=self.project_id)
136			backend_id = self.config['backend']
137			try:
138			backend_class = annif.backend.get_backend(backend_id)
139			self._backend = backend_class(
140			backend_id, config_params=self.config,
141			project=self)
142			except ValueError:
143			logger.warning(
144			"Could not create backend %s, "
145			"make sure you've installed optional dependencies",
146			backend_id)
147			return self._backend
148
149			@property
150			def vocab(self):
151			if self._vocab is None:
152			if self.vocab_id is None:
153			raise ConfigurationException("vocab setting is missing",
154			project_id=self.project_id)
155			self._vocab = annif.vocab.AnnifVocabulary(self.vocab_id,
156			self._base_datadir,
157			self.language)
158			return self._vocab
159
160			@property
161			def subjects(self):
162			return self.vocab.subjects
163
164			def _get_info(self, key):
165			try:
166			be = self.backend
167			if be is not None:
168			return getattr(be, key)
169			except AnnifException as err:
170			logger.warning(err.format_message())
171			return None
172
173			@property
174			def is_trained(self):
175			return self._get_info('is_trained')
176
177			@property
178			def modification_time(self):
179			return self._get_info('modification_time')
180
181			def suggest(self, text, backend_params=None):
182			"""Suggest subjects the given text by passing it to the backend. Returns a
183			list of SubjectSuggestion objects ordered by decreasing score."""
184			if not self.is_trained:
185			if self.is_trained is None:
186			logger.warning('Could not get train state information.')
187			else:
188			raise NotInitializedException('Project is not trained.')
189			logger.debug('Suggesting subjects for text "%s..." (len=%d)',
190			text[:20], len(text))
191			text = self.transform.transform_text(text)
192			hits = self._suggest_with_backend(text, backend_params)
193			logger.debug('%d hits from backend', len(hits))
194			return hits
195
196			def train(self, corpus, backend_params=None):
197			"""train the project using documents from a metadata source"""
198			if corpus != 'cached':
199			corpus.set_subject_index(self.subjects)
200			corpus = self.transform.transform_corpus(corpus)
201			if backend_params is None:
202			backend_params = {}
203			beparams = backend_params.get(self.backend.backend_id, {})
204			self.backend.train(corpus, beparams)
205
206			def learn(self, corpus, backend_params=None):
207			"""further train the project using documents from a metadata source"""
208			corpus.set_subject_index(self.subjects)
209			if backend_params is None:
210			backend_params = {}
211			beparams = backend_params.get(self.backend.backend_id, {})
212			corpus = self.transform.transform_corpus(corpus)
213			if isinstance(
214			self.backend,
215			annif.backend.backend.AnnifLearningBackend):
216			self.backend.learn(corpus, beparams)
217			else:
218			raise NotSupportedException("Learning not supported by backend",
219			project_id=self.project_id)
220
221			def hyperopt(self, corpus, trials, jobs, metric, results_file):
222			"""optimize the hyperparameters of the project using a validation
223			corpus against a given metric"""
224			if isinstance(
225			self.backend,
226			annif.backend.hyperopt.AnnifHyperoptBackend):
227			optimizer = self.backend.get_hp_optimizer(corpus, metric)
228			return optimizer.optimize(trials, jobs, results_file)
229
230			raise NotSupportedException(
231			"Hyperparameter optimization not supported "
232			"by backend", project_id=self.project_id)
233
234			def dump(self):
235			"""return this project as a dict"""
236			return {'project_id': self.project_id,
237			'name': self.name,
238			'language': self.language,
239			'backend': {'backend_id': self.config.get('backend')},
240			'is_trained': self.is_trained,
241			'modification_time': self.modification_time
242			}
243
244			def remove_model_data(self):
245			"""remove the data of this project"""
246			datadir_path = self._datadir_path
247			if os.path.isdir(datadir_path):
248			rmtree(datadir_path)
249			logger.info('Removed model data for project {}.'
250			.format(self.project_id))
251			else:
252			logger.warning('No model data to remove for project {}.'
253			.format(self.project_id))
254

NatLibFi / Annif

Pull Request — master (#496)

annif.project.AnnifProject.transform() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like