import os import importlib import importlib.util import re from pathlib import Path import random from typing import List import thinc from thinc.api import NumpyOps, get_current_ops, Adam, require_gpu, Config import functools import itertools import numpy.random import srsly import catalogue import sys import warnings try: import cupy.random except ImportError: cupy = None try: # Python 3.8 import importlib.metadata as importlib_metadata except ImportError: import importlib_metadata from .symbols import ORTH from .compat import cupy, CudaStream from .errors import Errors, Warnings from . import about _PRINT_ENV = False class registry(thinc.registry): languages = catalogue.create("spacy", "languages", entry_points=True) architectures = catalogue.create("spacy", "architectures", entry_points=True) lookups = catalogue.create("spacy", "lookups", entry_points=True) factories = catalogue.create("spacy", "factories", entry_points=True) displacy_colors = catalogue.create("spacy", "displacy_colors", entry_points=True) assets = catalogue.create("spacy", "assets", entry_points=True) # This is mostly used to get a list of all installed models in the current # environment. spaCy models packaged with `spacy package` will "advertise" # themselves via entry points. models = catalogue.create("spacy", "models", entry_points=True) def set_env_log(value): global _PRINT_ENV _PRINT_ENV = value def lang_class_is_loaded(lang): """Check whether a Language class is already loaded. Language classes are loaded lazily, to avoid expensive setup code associated with the language data. lang (unicode): Two-letter language code, e.g. 'en'. RETURNS (bool): Whether a Language class has been loaded. """ return lang in registry.languages def get_lang_class(lang): """Import and load a Language class. lang (unicode): Two-letter language code, e.g. 'en'. RETURNS (Language): Language class. """ # Check if language is registered / entry point is available if lang in registry.languages: return registry.languages.get(lang) else: try: module = importlib.import_module(f".lang.{lang}", "spacy") except ImportError as err: raise ImportError(Errors.E048.format(lang=lang, err=err)) set_lang_class(lang, getattr(module, module.__all__[0])) return registry.languages.get(lang) def set_lang_class(name, cls): """Set a custom Language class name that can be loaded via get_lang_class. name (unicode): Name of Language class. cls (Language): Language class. """ registry.languages.register(name, func=cls) def ensure_path(path): """Ensure string is converted to a Path. path: Anything. If string, it's converted to Path. RETURNS: Path or original argument. """ if isinstance(path, str): return Path(path) else: return path def load_language_data(path): """Load JSON language data using the given path as a base. If the provided path isn't present, will attempt to load a gzipped version before giving up. path (unicode / Path): The data to load. RETURNS: The loaded data. """ path = ensure_path(path) if path.exists(): return srsly.read_json(path) path = path.with_suffix(path.suffix + ".gz") if path.exists(): return srsly.read_gzip_json(path) raise ValueError(Errors.E160.format(path=path)) def get_module_path(module): if not hasattr(module, "__module__"): raise ValueError(Errors.E169.format(module=repr(module))) return Path(sys.modules[module.__module__].__file__).parent def load_model(name, **overrides): """Load a model from a package or data path. name (unicode): Package name or model path. **overrides: Specific overrides, like pipeline components to disable. RETURNS (Language): `Language` class with the loaded model. """ if isinstance(name, str): # name or string path if is_package(name): # installed as package return load_model_from_package(name, **overrides) if Path(name).exists(): # path to model data directory return load_model_from_path(Path(name), **overrides) elif hasattr(name, "exists"): # Path or Path-like to model data return load_model_from_path(name, **overrides) raise IOError(Errors.E050.format(name=name)) def load_model_from_package(name, **overrides): """Load a model from an installed package.""" cls = importlib.import_module(name) return cls.load(**overrides) def load_model_from_path(model_path, meta=False, **overrides): """Load a model from a data directory path. Creates Language class with pipeline from meta.json and then calls from_disk() with path.""" if not meta: meta = get_model_meta(model_path) nlp_config = get_model_config(model_path) if nlp_config.get("nlp", None): return load_model_from_config(nlp_config["nlp"]) # Support language factories registered via entry points (e.g. custom # language subclass) while keeping top-level language identifier "lang" lang = meta.get("lang_factory", meta["lang"]) cls = get_lang_class(lang) nlp = cls(meta=meta, **overrides) pipeline = meta.get("pipeline", []) factories = meta.get("factories", {}) disable = overrides.get("disable", []) if pipeline is True: pipeline = nlp.Defaults.pipe_names elif pipeline in (False, None): pipeline = [] for name in pipeline: if name not in disable: config = meta.get("pipeline_args", {}).get(name, {}) config.update(overrides) factory = factories.get(name, name) if nlp_config.get(name, None): model_config = nlp_config[name]["model"] config["model"] = model_config component = nlp.create_pipe(factory, config=config) nlp.add_pipe(component, name=name) return nlp.from_disk(model_path, exclude=disable) def load_model_from_config(nlp_config): if "name" in nlp_config: nlp = load_model(**nlp_config) elif "lang" in nlp_config: lang_class = get_lang_class(nlp_config["lang"]) nlp = lang_class() else: raise ValueError(Errors.E993) if "pipeline" in nlp_config: for name, component_cfg in nlp_config["pipeline"].items(): factory = component_cfg.pop("factory") component = nlp.create_pipe(factory, config=component_cfg) nlp.add_pipe(component, name=name) return nlp def load_model_from_init_py(init_file, **overrides): """Helper function to use in the `load()` method of a model package's __init__.py. init_file (unicode): Path to model's __init__.py, i.e. `__file__`. **overrides: Specific overrides, like pipeline components to disable. RETURNS (Language): `Language` class with loaded model. """ model_path = Path(init_file).parent meta = get_model_meta(model_path) data_dir = f"{meta['lang']}_{meta['name']}-{meta['version']}" data_path = model_path / data_dir if not model_path.exists(): raise IOError(Errors.E052.format(path=data_path)) return load_model_from_path(data_path, meta, **overrides) def get_installed_models(): """List all model packages currently installed in the environment. RETURNS (list): The string names of the models. """ return list(registry.models.get_all().keys()) def get_package_version(name): """Get the version of an installed package. Typically used to get model package versions. name (unicode): The name of the installed Python package. RETURNS (unicode / None): The version or None if package not installed. """ try: return importlib_metadata.version(name) except importlib_metadata.PackageNotFoundError: return None def split_version(version): """RETURNS (tuple): Two integers, the major and minor spaCy version.""" pieces = version.split(".", 3) return int(pieces[0]), int(pieces[1]) def is_compatible_model(meta): """Check if a model is compatible with the current version of spaCy, based on its meta.json. We compare the version of spaCy the model was created with with the current version. If the minor version is different, it's considered incompatible. meta (dict): The model's meta. RETURNS (bool / None): Whether the model is compatible with the current spaCy or None if we don't have enough info. """ cur_v = about.__version__ pkg_v = meta.get("spacy_version") if not pkg_v or not isinstance(pkg_v, str): return None # Handle spacy_version values like >=x,>> sizes = compounding(1., 10., 1.5) >>> assert next(sizes) == 1. >>> assert next(sizes) == 1 * 1.5 >>> assert next(sizes) == 1.5 * 1.5 """ def clip(value): return max(value, stop) if (start > stop) else min(value, stop) curr = float(start) while True: yield clip(curr) curr *= compound def stepping(start, stop, steps): """Yield an infinite series of values that step from a start value to a final value over some number of steps. Each step is (stop-start)/steps. After the final value is reached, the generator continues yielding that value. EXAMPLE: >>> sizes = stepping(1., 200., 100) >>> assert next(sizes) == 1. >>> assert next(sizes) == 1 * (200.-1.) / 100 >>> assert next(sizes) == 1 + (200.-1.) / 100 + (200.-1.) / 100 """ def clip(value): return max(value, stop) if (start > stop) else min(value, stop) curr = float(start) while True: yield clip(curr) curr += (stop - start) / steps def decaying(start, stop, decay): """Yield an infinite series of linearly decaying values.""" curr = float(start) while True: yield max(curr, stop) curr -= decay def minibatch_by_words(examples, size, tuples=True, count_words=len, tolerance=0.2): """Create minibatches of roughly a given number of words. If any examples are longer than the specified batch length, they will appear in a batch by themselves.""" if isinstance(size, int): size_ = itertools.repeat(size) elif isinstance(size, List): size_ = iter(size) else: size_ = size examples = iter(examples) oversize = [] while True: batch_size = next(size_) tol_size = batch_size * 0.2 batch = [] if oversize: example = oversize.pop(0) n_words = count_words(example.doc) batch.append(example) batch_size -= n_words while batch_size >= 1: try: example = next(examples) except StopIteration: if batch: yield batch return n_words = count_words(example.doc) if n_words < (batch_size + tol_size): batch_size -= n_words batch.append(example) else: oversize.append(example) if batch: yield batch def itershuffle(iterable, bufsize=1000): """Shuffle an iterator. This works by holding `bufsize` items back and yielding them sometime later. Obviously, this is not unbiased – but should be good enough for batching. Larger bufsize means less bias. From https://gist.github.com/andres-erbsen/1307752 iterable (iterable): Iterator to shuffle. bufsize (int): Items to hold back. YIELDS (iterable): The shuffled iterator. """ iterable = iter(iterable) buf = [] try: while True: for i in range(random.randint(1, bufsize - len(buf))): buf.append(next(iterable)) random.shuffle(buf) for i in range(random.randint(1, bufsize)): if buf: yield buf.pop() else: break except StopIteration: random.shuffle(buf) while buf: yield buf.pop() raise StopIteration def filter_spans(spans): """Filter a sequence of spans and remove duplicates or overlaps. Useful for creating named entities (where one token can only be part of one entity) or when merging spans with `Retokenizer.merge`. When spans overlap, the (first) longest span is preferred over shorter spans. spans (iterable): The spans to filter. RETURNS (list): The filtered spans. """ get_sort_key = lambda span: (span.end - span.start, -span.start) sorted_spans = sorted(spans, key=get_sort_key, reverse=True) result = [] seen_tokens = set() for span in sorted_spans: # Check for end - 1 here because boundaries are inclusive if span.start not in seen_tokens and span.end - 1 not in seen_tokens: result.append(span) seen_tokens.update(range(span.start, span.end)) result = sorted(result, key=lambda span: span.start) return result def to_bytes(getters, exclude): serialized = {} for key, getter in getters.items(): # Split to support file names like meta.json if key.split(".")[0] not in exclude: serialized[key] = getter() return srsly.msgpack_dumps(serialized) def from_bytes(bytes_data, setters, exclude): msg = srsly.msgpack_loads(bytes_data) for key, setter in setters.items(): # Split to support file names like meta.json if key.split(".")[0] not in exclude and key in msg: setter(msg[key]) return msg def to_disk(path, writers, exclude): path = ensure_path(path) if not path.exists(): path.mkdir() for key, writer in writers.items(): # Split to support file names like meta.json if key.split(".")[0] not in exclude: writer(path / key) return path def from_disk(path, readers, exclude): path = ensure_path(path) for key, reader in readers.items(): # Split to support file names like meta.json if key.split(".")[0] not in exclude: reader(path / key) return path def import_file(name, loc): """Import module from a file. Used to load models from a directory. name (unicode): Name of module to load. loc (unicode / Path): Path to the file. RETURNS: The loaded module. """ loc = str(loc) spec = importlib.util.spec_from_file_location(name, str(loc)) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def minify_html(html): """Perform a template-specific, rudimentary HTML minification for displaCy. Disclaimer: NOT a general-purpose solution, only removes indentation and newlines. html (unicode): Markup to minify. RETURNS (unicode): "Minified" HTML. """ return html.strip().replace(" ", "").replace("\n", "") def escape_html(text): """Replace <, >, &, " with their HTML encoded representation. Intended to prevent HTML errors in rendered displaCy markup. text (unicode): The original text. RETURNS (unicode): Equivalent text to be safely used within HTML. """ text = text.replace("&", "&") text = text.replace("<", "<") text = text.replace(">", ">") text = text.replace('"', """) return text def use_gpu(gpu_id): return require_gpu(gpu_id) def fix_random_seed(seed=0): random.seed(seed) numpy.random.seed(seed) if cupy is not None: cupy.random.seed(seed) def get_serialization_exclude(serializers, exclude, kwargs): """Helper function to validate serialization args and manage transition from keyword arguments (pre v2.1) to exclude argument. """ exclude = list(exclude) # Split to support file names like meta.json options = [name.split(".")[0] for name in serializers] for key, value in kwargs.items(): if key in ("vocab",) and value is False: warnings.warn(Warnings.W015.format(arg=key), DeprecationWarning) exclude.append(key) elif key.split(".")[0] in options: raise ValueError(Errors.E128.format(arg=key)) # TODO: user warning? return exclude class SimpleFrozenDict(dict): """Simplified implementation of a frozen dict, mainly used as default function or method argument (for arguments that should default to empty dictionary). Will raise an error if user or spaCy attempts to add to dict. """ def __setitem__(self, key, value): raise NotImplementedError(Errors.E095) def pop(self, key, default=None): raise NotImplementedError(Errors.E095) def update(self, other): raise NotImplementedError(Errors.E095) class DummyTokenizer(object): # add dummy methods for to_bytes, from_bytes, to_disk and from_disk to # allow serialization (see #1557) def to_bytes(self, **kwargs): return b"" def from_bytes(self, _bytes_data, **kwargs): return self def to_disk(self, _path, **kwargs): return None def from_disk(self, _path, **kwargs): return self def link_vectors_to_models(vocab): vectors = vocab.vectors if vectors.name is None: vectors.name = VECTORS_KEY if vectors.data.size != 0: warnings.warn(Warnings.W020.format(shape=vectors.data.shape)) for word in vocab: if word.orth in vectors.key2row: word.rank = vectors.key2row[word.orth] else: word.rank = 0 VECTORS_KEY = "spacy_pretrained_vectors" def create_default_optimizer(): learn_rate = env_opt("learn_rate", 0.001) beta1 = env_opt("optimizer_B1", 0.9) beta2 = env_opt("optimizer_B2", 0.999) eps = env_opt("optimizer_eps", 1e-8) L2 = env_opt("L2_penalty", 1e-6) grad_clip = env_opt("grad_norm_clip", 10.0) L2_is_weight_decay = env_opt("L2_is_weight_decay", False) optimizer = Adam( learn_rate, L2=L2, beta1=beta1, beta2=beta2, eps=eps, grad_clip=grad_clip, L2_is_weight_decay=L2_is_weight_decay, ) return optimizer