1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
|
"""
Pipeline that runs tokenize,mwt,pos,lemma,depparse
"""
import io
import itertools
import sys
import torch
import logging
import json
import os
from distutils.util import strtobool
from stanza.pipeline._constants import *
from stanza.models.common.doc import Document
from stanza.pipeline.processor import Processor, ProcessorRequirementsException
from stanza.pipeline.registry import NAME_TO_PROCESSOR_CLASS, PIPELINE_NAMES
from stanza.pipeline.langid_processor import LangIDProcessor
from stanza.pipeline.tokenize_processor import TokenizeProcessor
from stanza.pipeline.mwt_processor import MWTProcessor
from stanza.pipeline.pos_processor import POSProcessor
from stanza.pipeline.lemma_processor import LemmaProcessor
from stanza.pipeline.depparse_processor import DepparseProcessor
from stanza.pipeline.sentiment_processor import SentimentProcessor
from stanza.pipeline.constituency_processor import ConstituencyProcessor
from stanza.pipeline.ner_processor import NERProcessor
from stanza.resources.common import DEFAULT_MODEL_DIR, \
maintain_processor_list, add_dependencies, add_mwt, build_default_config, set_logging_level, process_pipeline_parameters, sort_processors
from stanza.utils.helper_func import make_table
logger = logging.getLogger('stanza')
class ResourcesFileNotFoundError(FileNotFoundError):
def __init__(self, resources_filepath):
super().__init__(f"Resources file not found at: {resources_filepath} Try to download the model again.")
self.resources_filepath = resources_filepath
class LanguageNotDownloadedError(FileNotFoundError):
def __init__(self, lang, lang_dir, model_path):
super().__init__(f'Could not find the model file {model_path}. The expected model directory {lang_dir} is missing. Perhaps you need to run stanza.download("{lang}")')
self.lang = lang
self.lang_dir = lang_dir
self.model_path = model_path
class UnsupportedProcessorError(FileNotFoundError):
def __init__(self, processor, lang):
super().__init__(f'Processor {processor} is not known for language {lang}. If you have created your own model, please specify the {processor}_model_path parameter when creating the pipeline.')
self.processor = processor
self.lang = lang
class PipelineRequirementsException(Exception):
"""
Exception indicating one or more requirements failures while attempting to build a pipeline.
Contains a ProcessorRequirementsException list.
"""
def __init__(self, processor_req_fails):
self._processor_req_fails = processor_req_fails
self.build_message()
@property
def processor_req_fails(self):
return self._processor_req_fails
def build_message(self):
err_msg = io.StringIO()
print(*[req_fail.message for req_fail in self.processor_req_fails], sep='\n', file=err_msg)
self.message = '\n\n' + err_msg.getvalue()
def __str__(self):
return self.message
class Pipeline:
def __init__(self, lang='en', dir=DEFAULT_MODEL_DIR, package='default', processors={}, logging_level=None, verbose=None, use_gpu=True, model_dir=None, **kwargs):
self.lang, self.dir, self.kwargs = lang, dir, kwargs
if model_dir is not None and dir == DEFAULT_MODEL_DIR:
self.dir = model_dir
# set global logging level
set_logging_level(logging_level, verbose)
# process different pipeline parameters
lang, self.dir, package, processors = process_pipeline_parameters(lang, self.dir, package, processors)
# Load resources.json to obtain latest packages.
logger.debug('Loading resource file...')
resources_filepath = os.path.join(self.dir, 'resources.json')
if not os.path.exists(resources_filepath):
raise ResourcesFileNotFoundError(resources_filepath)
with open(resources_filepath) as infile:
resources = json.load(infile)
if lang in resources:
if 'alias' in resources[lang]:
logger.info(f'"{lang}" is an alias for "{resources[lang]["alias"]}"')
lang = resources[lang]['alias']
lang_name = resources[lang]['lang_name'] if 'lang_name' in resources[lang] else ''
else:
logger.warning(f'Unsupported language: {lang}.')
# Maintain load list
if (not kwargs.get("tokenize_pretokenized")
and TOKENIZE in processors
and MWT not in processors):
add_mwt(processors, resources, lang)
self.load_list = maintain_processor_list(resources, lang, package, processors) if lang in resources else []
self.load_list = add_dependencies(resources, lang, self.load_list) if lang in resources else []
self.load_list = self.update_kwargs(kwargs, self.load_list)
if len(self.load_list) == 0:
raise ValueError('No processors to load for language {}. Please check if your language or package is correctly set.'.format(lang))
load_table = make_table(['Processor', 'Package'], [row[:2] for row in self.load_list])
logger.info(f'Loading these models for language: {lang} ({lang_name}):\n{load_table}')
self.config = build_default_config(resources, lang, self.dir, self.load_list)
self.config.update(kwargs)
# Load processors
self.processors = {}
# configs that are the same for all processors
pipeline_level_configs = {'lang': lang, 'mode': 'predict'}
self.use_gpu = torch.cuda.is_available() and use_gpu
logger.info("Use device: {}".format("gpu" if self.use_gpu else "cpu"))
# set up processors
pipeline_reqs_exceptions = []
for item in self.load_list:
processor_name, _, _ = item
logger.info('Loading: ' + processor_name)
curr_processor_config = self.filter_config(processor_name, self.config)
curr_processor_config.update(pipeline_level_configs)
# TODO: this is obviously a hack
# a better solution overall would be to make a pretagged version of the pos annotator
# and then subsequent modules can use those tags without knowing where those tags came from
if "pretagged" in self.config and "pretagged" not in curr_processor_config:
curr_processor_config["pretagged"] = self.config["pretagged"]
logger.debug('With settings: ')
logger.debug(curr_processor_config)
try:
# try to build processor, throw an exception if there is a requirements issue
self.processors[processor_name] = NAME_TO_PROCESSOR_CLASS[processor_name](config=curr_processor_config,
pipeline=self,
use_gpu=self.use_gpu)
except ProcessorRequirementsException as e:
# if there was a requirements issue, add it to list which will be printed at end
pipeline_reqs_exceptions.append(e)
# add the broken processor to the loaded processors for the sake of analyzing the validity of the
# entire proposed pipeline, but at this point the pipeline will not be built successfully
self.processors[processor_name] = e.err_processor
except FileNotFoundError as e:
# For a FileNotFoundError, we try to guess if there's
# a missing model directory or file. If so, we
# suggest the user try to download the models
if 'model_path' in curr_processor_config:
model_path = curr_processor_config['model_path']
model_dir, model_name = os.path.split(model_path)
lang_dir = os.path.dirname(model_dir)
if not os.path.exists(lang_dir):
# model files for this language can't be found in the expected directory
raise LanguageNotDownloadedError(lang, lang_dir, model_path) from e
if processor_name not in resources[lang]:
# user asked for a model which doesn't exist for this language?
raise UnsupportedProcessorError(processor_name, lang)
if not os.path.exists(model_path):
model_name, _ = os.path.splitext(model_name)
# TODO: before recommending this, check that such a thing exists in resources.json.
# currently that case is handled by ignoring the model, anyway
raise FileNotFoundError('Could not find model file %s, although there are other models downloaded for language %s. Perhaps you need to download a specific model. Try: stanza.download(lang="%s",package=None,processors={"%s":"%s"})' % (model_path, lang, lang, processor_name, model_name)) from e
# if we couldn't find a more suitable description of the
# FileNotFoundError, just raise the old error
raise
# if there are any processor exceptions, throw an exception to indicate pipeline build failure
if pipeline_reqs_exceptions:
logger.info('\n')
raise PipelineRequirementsException(pipeline_reqs_exceptions)
logger.info("Done loading processors!")
@staticmethod
def update_kwargs(kwargs, processor_list):
processor_dict = {processor: {'package': package, 'dependencies': dependencies} for (processor, package, dependencies) in processor_list}
for key, value in kwargs.items():
pieces = key.split('_', 1)
if len(pieces) == 1:
continue
k, v = pieces
if v == 'model_path':
package = value if len(value) < 25 else value[:10]+ '...' + value[-10:]
dependencies = processor_dict.get(k, {}).get('dependencies')
processor_dict[k] = {'package': package, 'dependencies': dependencies}
processor_list = [[processor, processor_dict[processor]['package'], processor_dict[processor]['dependencies']] for processor in processor_dict]
processor_list = sort_processors(processor_list)
return processor_list
@staticmethod
def filter_config(prefix, config_dict):
filtered_dict = {}
for key in config_dict.keys():
pieces = key.split('_', 1) # split tokenize_pretokenize to tokenize+pretokenize
if len(pieces) == 1:
continue
k, v = pieces
if k == prefix:
filtered_dict[v] = config_dict[key]
return filtered_dict
@property
def loaded_processors(self):
"""
Return all currently loaded processors in execution order.
:return: list of Processor instances
"""
return [self.processors[processor_name] for processor_name in PIPELINE_NAMES if self.processors.get(processor_name)]
def process(self, doc):
# run the pipeline
# determine whether we are in bulk processing mode for multiple documents
bulk=(isinstance(doc, list) and len(doc) > 0 and isinstance(doc[0], Document))
for processor_name in PIPELINE_NAMES:
if self.processors.get(processor_name):
process = self.processors[processor_name].bulk_process if bulk else self.processors[processor_name].process
doc = process(doc)
return doc
def __call__(self, doc):
assert any([isinstance(doc, str), isinstance(doc, list),
isinstance(doc, Document)]), 'input should be either str, list or Document'
doc = self.process(doc)
return doc
|
