import functools

import importlib

import importlib.util

import inspect

import itertools

import logging

import os

import re

import shlex

import shutil

import socket

import stat

import subprocess

import sys

import tempfile

import warnings

from collections import defaultdict

from contextlib import contextmanager

from pathlib import Path

from types import ModuleType

from typing import (

TYPE_CHECKING,

Any,

Callable,

Dict,

Generator,

Iterable,

Iterator,

List,

Mapping,

NoReturn,

Optional,

Pattern,

Set,

Tuple,

Type,

Union,

cast,

)

import catalogue

import numpy

import srsly

import thinc

from catalogue import Registry, RegistryError

from packaging.requirements import Requirement

from packaging.specifiers import InvalidSpecifier, SpecifierSet

from packaging.version import InvalidVersion, Version

from thinc.api import (

Adam,

Config,

ConfigValidationError,

Model,

NumpyOps,

Optimizer,

get_current_ops,

)

try:

import cupy.random

except ImportError:

cupy = None

# These are functions that were previously (v2.x) available from spacy.util

# and have since moved to Thinc. We're importing them here so people's code

# doesn't break, but they should always be imported from Thinc from now on,

# not from spacy.util.

from thinc.api import compounding, decaying, fix_random_seed # noqa: F401

from . import about

from .compat import CudaStream, cupy, importlib_metadata, is_windows

from .errors import OLD_MODEL_SHORTCUTS, Errors, Warnings

from .symbols import ORTH

if TYPE_CHECKING:

# This lets us add type hints for mypy etc. without causing circular imports

from .language import Language, PipeCallable # noqa: F401

from .tokens import Doc, Span # noqa: F401

from .vocab import Vocab # noqa: F401

# fmt: off

OOV_RANK = numpy.iinfo(numpy.uint64).max

DEFAULT_OOV_PROB = -20

LEXEME_NORM_LANGS = ["cs", "da", "de", "el", "en", "grc", "id", "lb", "mk", "pt", "ru", "sr", "ta", "th"]

# Default order of sections in the config file. Not all sections needs to exist,

# and additional sections are added at the end, in alphabetical order.

CONFIG_SECTION_ORDER = ["paths", "variables", "system", "nlp", "components", "corpora", "training", "pretraining", "initialize"]

LANG_ALIASES = {

"af": ["afr"],

"am": ["amh"],

"ar": ["ara"],

"az": ["aze"],

"bg": ["bul"],

"bn": ["ben"],

"bo": ["bod", "tib"],

"ca": ["cat"],

"cs": ["ces", "cze"],

"da": ["dan"],

"de": ["deu", "ger"],

"el": ["ell", "gre"],

"en": ["eng"],

"es": ["spa"],

"et": ["est"],

"eu": ["eus", "baq"],

"fa": ["fas", "per"],

"fi": ["fin"],

"fo": ["fao"],

"fr": ["fra", "fre"],

"ga": ["gle"],

"gd": ["gla"],

"gu": ["guj"],

"he": ["heb", "iw"], # "iw" is the obsolete ISO 639-1 code for Hebrew

"hi": ["hin"],

"hr": ["hrv", "scr"], # "scr" is the deprecated ISO 639-2/B for Croatian

"hu": ["hun"],

"hy": ["hye"],

"id": ["ind", "in"], # "in" is the obsolete ISO 639-1 code for Hebrew

"is": ["isl", "ice"],

"it": ["ita"],

"ja": ["jpn"],

"kn": ["kan"],

"ko": ["kor"],

"ky": ["kir"],

"la": ["lat"],

"lb": ["ltz"],

"lg": ["lug"],

"lt": ["lit"],

"lv": ["lav"],

"mk": ["mkd", "mac"],

"ml": ["mal"],

"mr": ["mar"],

"ms": ["msa", "may"],

"nb": ["nob"],

"ne": ["nep"],

"nl": ["nld", "dut"],

"nn": ["nno"],

"pl": ["pol"],

"pt": ["por"],

"ro": ["ron", "rom", "mo", "mol"], # "mo" and "mol" are deprecated codes for Moldavian

"ru": ["rus"],

"sa": ["san"],

"si": ["sin"],

"sk": ["slk", "slo"],

"sl": ["slv"],

"sq": ["sqi", "alb"],

"sr": ["srp", "scc"], # "scc" is the deprecated ISO 639-2/B code for Serbian

"sv": ["swe"],

"ta": ["tam"],

"te": ["tel"],

"th": ["tha"],

"ti": ["tir"],

"tl": ["tgl"],

"tn": ["tsn"],

"tr": ["tur"],

"tt": ["tat"],

"uk": ["ukr"],

"ur": ["urd"],

"vi": ["viw"],

"yo": ["yor"],

"zh": ["zho", "chi"],

"xx": ["mul"],

}

# fmt: on

logger = logging.getLogger("spacy")

logger_stream_handler = logging.StreamHandler()

logger_stream_handler.setFormatter(

logging.Formatter("[%(asctime)s] [%(levelname)s] %(message)s")

)

logger.addHandler(logger_stream_handler)

class ENV_VARS:

CONFIG_OVERRIDES = "SPACY_CONFIG_OVERRIDES"

class registry(thinc.registry):

languages = catalogue.create("spacy", "languages", entry_points=True)

architectures = catalogue.create("spacy", "architectures", entry_points=True)

tokenizers = catalogue.create("spacy", "tokenizers", entry_points=True)

lemmatizers = catalogue.create("spacy", "lemmatizers", entry_points=True)

lookups = catalogue.create("spacy", "lookups", entry_points=True)

displacy_colors = catalogue.create("spacy", "displacy_colors", entry_points=True)

misc = catalogue.create("spacy", "misc", entry_points=True)

# Callback functions used to manipulate nlp object etc.

callbacks = catalogue.create("spacy", "callbacks", entry_points=True)

batchers = catalogue.create("spacy", "batchers", entry_points=True)

readers = catalogue.create("spacy", "readers", entry_points=True)

augmenters = catalogue.create("spacy", "augmenters", entry_points=True)

loggers = catalogue.create("spacy", "loggers", entry_points=True)

scorers = catalogue.create("spacy", "scorers", entry_points=True)

vectors = catalogue.create("spacy", "vectors", entry_points=True)

# These are factories registered via third-party packages and the

# spacy_factories entry point. This registry only exists so we can easily

# load them via the entry points. The "true" factories are added via the

# Language.factory decorator (in the spaCy code base and user code) and those

# are the factories used to initialize components via registry.resolve.

_entry_point_factories = catalogue.create("spacy", "factories", entry_points=True)

factories = catalogue.create("spacy", "internal_factories")

# 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)

cli = catalogue.create("spacy", "cli", entry_points=True)

@classmethod

def ensure_populated(cls) -> None:

"""Ensure the registry is populated with all necessary components."""

from .registrations import REGISTRY_POPULATED, populate_registry

if not REGISTRY_POPULATED:

populate_registry()

@classmethod

def get_registry_names(cls) -> List[str]:

"""List all available registries."""

cls.ensure_populated()

names = []

for name, value in inspect.getmembers(cls):

if not name.startswith("_") and isinstance(value, Registry):

names.append(name)

return sorted(names)

@classmethod

def get(cls, registry_name: str, func_name: str) -> Callable:

"""Get a registered function from the registry."""

cls.ensure_populated()

# We're overwriting this classmethod so we're able to provide more

# specific error messages and implement a fallback to spacy-legacy.

if not hasattr(cls, registry_name):

names = ", ".join(cls.get_registry_names()) or "none"

raise RegistryError(Errors.E892.format(name=registry_name, available=names))

reg = getattr(cls, registry_name)

try:

func = reg.get(func_name)

except RegistryError:

if func_name.startswith("spacy."):

legacy_name = func_name.replace("spacy.", "spacy-legacy.")

try:

return reg.get(legacy_name)

except catalogue.RegistryError:

pass

available = ", ".join(sorted(reg.get_all().keys())) or "none"

raise RegistryError(

Errors.E893.format(

name=func_name, reg_name=registry_name, available=available

)

) from None

return func

@classmethod

def find(

cls, registry_name: str, func_name: str

) -> Dict[str, Optional[Union[str, int]]]:

"""Find information about a registered function, including the

module and path to the file it's defined in, the line number and the

docstring, if available.

registry_name (str): Name of the catalogue registry.

func_name (str): Name of the registered function.

RETURNS (Dict[str, Optional[Union[str, int]]]): The function info.

"""

cls.ensure_populated()

# We're overwriting this classmethod so we're able to provide more

# specific error messages and implement a fallback to spacy-legacy.

if not hasattr(cls, registry_name):

names = ", ".join(cls.get_registry_names()) or "none"

raise RegistryError(Errors.E892.format(name=registry_name, available=names))

reg = getattr(cls, registry_name)

try:

func_info = reg.find(func_name)

except RegistryError:

if func_name.startswith("spacy."):

legacy_name = func_name.replace("spacy.", "spacy-legacy.")

try:

return reg.find(legacy_name)

except catalogue.RegistryError:

pass

available = ", ".join(sorted(reg.get_all().keys())) or "none"

raise RegistryError(

Errors.E893.format(

name=func_name, reg_name=registry_name, available=available

)

) from None

return func_info

@classmethod

def has(cls, registry_name: str, func_name: str) -> bool:

"""Check whether a function is available in a registry."""

cls.ensure_populated()

if not hasattr(cls, registry_name):

return False

reg = getattr(cls, registry_name)

if func_name.startswith("spacy."):

legacy_name = func_name.replace("spacy.", "spacy-legacy.")

return func_name in reg or legacy_name in reg

return func_name in reg

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 __init__(self, *args, error: str = Errors.E095, **kwargs) -> None:

"""Initialize the frozen dict. Can be initialized with pre-defined

values.

error (str): The error message when user tries to assign to dict.

"""

super().__init__(*args, **kwargs)

self.error = error

def __setitem__(self, key, value):

raise NotImplementedError(self.error)

def pop(self, key, default=None):

raise NotImplementedError(self.error)

def update(self, other):

raise NotImplementedError(self.error)

class SimpleFrozenList(list):

"""Wrapper class around a list that lets us raise custom errors if certain

attributes/methods are accessed. Mostly used for properties like

Language.pipeline that return an immutable list (and that we don't want to

convert to a tuple to not break too much backwards compatibility). If a user

accidentally calls nlp.pipeline.append(), we can raise a more helpful error.

"""

def __init__(self, *args, error: str = Errors.E927) -> None:

"""Initialize the frozen list.

error (str): The error message when user tries to mutate the list.

"""

self.error = error

super().__init__(*args)

def append(self, *args, **kwargs):

raise NotImplementedError(self.error)

def clear(self, *args, **kwargs):

raise NotImplementedError(self.error)

def extend(self, *args, **kwargs):

raise NotImplementedError(self.error)

def insert(self, *args, **kwargs):

raise NotImplementedError(self.error)

def pop(self, *args, **kwargs):

raise NotImplementedError(self.error)

def remove(self, *args, **kwargs):

raise NotImplementedError(self.error)

def reverse(self, *args, **kwargs):

raise NotImplementedError(self.error)

def sort(self, *args, **kwargs):

raise NotImplementedError(self.error)

def lang_class_is_loaded(lang: str) -> bool:

"""Check whether a Language class is already loaded. Language classes are

loaded lazily, to avoid expensive setup code associated with the language

data.

lang (str): Two-letter language code, e.g. 'en'.

RETURNS (bool): Whether a Language class has been loaded.

"""

return lang in registry.languages

def find_matching_language(lang: str) -> Optional[str]:

"""

Given a two-letter ISO 639-1 or three-letter ISO 639-3 language code,

find a supported spaCy language.

Returns the language code if a matching language is available, or None

if there is no matching language.

>>> find_matching_language('fra') # ISO 639-3 code for French

'fr'

>>> find_matching_language('fre') # ISO 639-2/B code for French

'fr'

>>> find_matching_language('iw') # Obsolete ISO 639-1 code for Hebrew

'he'

>>> find_matching_language('mo') # Deprecated code for Moldavian

'ro'

>>> find_matching_language('scc') # Deprecated ISO 639-2/B code for Serbian

'sr'

>>> find_matching_language('zxx')

None

"""

import spacy.lang # noqa: F401

# Check aliases

for lang_code, aliases in LANG_ALIASES.items():

if lang in aliases:

return lang_code

return None

def get_lang_class(lang: str) -> Type["Language"]:

"""Import and load a Language class.

lang (str): Two-letter ISO 639-1 or three-letter ISO 639-3 language code, such as 'en' and 'eng'.

RETURNS (Language): Language class.

"""

# Check if language is registered / entry point is available

if lang in registry.languages:

return registry.languages.get(lang)

else:

# Find the language in the spacy.lang subpackage

try:

module = importlib.import_module(f".lang.{lang}", "spacy")

except ImportError as err:

# Find a matching language. For example, if the language 'eng' is

# requested, we can use language-matching to load `spacy.lang.en`.

match = find_matching_language(lang)

if match:

lang = match

module = importlib.import_module(f".lang.{lang}", "spacy")

else:

raise ImportError(Errors.E048.format(lang=lang, err=err)) from err

set_lang_class(lang, getattr(module, module.__all__[0])) # type: ignore[attr-defined]

return registry.languages.get(lang)

def set_lang_class(name: str, cls: Type["Language"]) -> None:

"""Set a custom Language class name that can be loaded via get_lang_class.

name (str): Name of Language class.

cls (Language): Language class.

"""

registry.languages.register(name, func=cls)

def ensure_path(path: Any) -> Any:

"""Ensure string is converted to a Path.

path (Any): 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: Union[str, Path]) -> Union[dict, list]:

"""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 (str / 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: ModuleType) -> Path:

"""Get the path of a Python module.

module (ModuleType): The Python module.

RETURNS (Path): The path.

"""

if not hasattr(module, "__module__"):

raise ValueError(Errors.E169.format(module=repr(module)))

file_path = Path(cast(os.PathLike, sys.modules[module.__module__].__file__))

return file_path.parent

# Default value for passed enable/disable values.

_DEFAULT_EMPTY_PIPES = SimpleFrozenList()

def load_model(

name: Union[str, Path],

*,

vocab: Union["Vocab", bool] = True,

disable: Union[str, Iterable[str]] = _DEFAULT_EMPTY_PIPES,

enable: Union[str, Iterable[str]] = _DEFAULT_EMPTY_PIPES,

exclude: Union[str, Iterable[str]] = _DEFAULT_EMPTY_PIPES,

config: Union[Dict[str, Any], Config] = SimpleFrozenDict(),

) -> "Language":

"""Load a model from a package or data path.

name (str): Package name or model path.

vocab (Vocab / True): Optional vocab to pass in on initialization. If True,

a new Vocab object will be created.

disable (Union[str, Iterable[str]]): Name(s) of pipeline component(s) to disable.

enable (Union[str, Iterable[str]]): Name(s) of pipeline component(s) to enable. All others will be disabled.

exclude (Union[str, Iterable[str]]): Name(s) of pipeline component(s) to exclude.

config (Dict[str, Any] / Config): Config overrides as nested dict or dict

keyed by section values in dot notation.

RETURNS (Language): The loaded nlp object.

"""

kwargs = {

"vocab": vocab,

"disable": disable,

"enable": enable,

"exclude": exclude,

"config": config,

}

if isinstance(name, str): # name or string path

if name.startswith("blank:"): # shortcut for blank model

return get_lang_class(name.replace("blank:", ""))()

if is_package(name): # installed as package

return load_model_from_package(name, **kwargs) # type: ignore[arg-type]

if Path(name).exists(): # path to model data directory

return load_model_from_path(Path(name), **kwargs) # type: ignore[arg-type]

elif hasattr(name, "exists"): # Path or Path-like to model data

return load_model_from_path(name, **kwargs) # type: ignore[arg-type]

if name in OLD_MODEL_SHORTCUTS:

raise IOError(Errors.E941.format(name=name, full=OLD_MODEL_SHORTCUTS[name])) # type: ignore[index]

raise IOError(Errors.E050.format(name=name))

def load_model_from_package(

name: str,

*,

vocab: Union["Vocab", bool] = True,

disable: Union[str, Iterable[str]] = _DEFAULT_EMPTY_PIPES,

enable: Union[str, Iterable[str]] = _DEFAULT_EMPTY_PIPES,

exclude: Union[str, Iterable[str]] = _DEFAULT_EMPTY_PIPES,

config: Union[Dict[str, Any], Config] = SimpleFrozenDict(),

) -> "Language":

"""Load a model from an installed package.

name (str): The package name.

vocab (Vocab / True): Optional vocab to pass in on initialization. If True,

a new Vocab object will be created.

disable (Union[str, Iterable[str]]): Name(s) of pipeline component(s) to disable. Disabled

pipes will be loaded but they won't be run unless you explicitly

enable them by calling nlp.enable_pipe.

enable (Union[str, Iterable[str]]): Name(s) of pipeline component(s) to enable. All other

pipes will be disabled (and can be enabled using `nlp.enable_pipe`).

exclude (Union[str, Iterable[str]]): Name(s) of pipeline component(s) to exclude. Excluded

components won't be loaded.

config (Dict[str, Any] / Config): Config overrides as nested dict or dict

keyed by section values in dot notation.

RETURNS (Language): The loaded nlp object.

"""

cls = importlib.import_module(name)

return cls.load(

vocab=vocab, disable=disable, enable=enable, exclude=exclude, config=config

) # type: ignore[attr-defined]

def load_model_from_path(

model_path: Path,

*,

meta: Optional[Dict[str, Any]] = None,

vocab: Union["Vocab", bool] = True,

disable: Union[str, Iterable[str]] = _DEFAULT_EMPTY_PIPES,

enable: Union[str, Iterable[str]] = _DEFAULT_EMPTY_PIPES,

exclude: Union[str, Iterable[str]] = _DEFAULT_EMPTY_PIPES,

config: Union[Dict[str, Any], Config] = SimpleFrozenDict(),

) -> "Language":

"""Load a model from a data directory path. Creates Language class with

pipeline from config.cfg and then calls from_disk() with path.

model_path (Path): Model path.

meta (Dict[str, Any]): Optional model meta.

vocab (Vocab / True): Optional vocab to pass in on initialization. If True,

a new Vocab object will be created.

disable (Union[str, Iterable[str]]): Name(s) of pipeline component(s) to disable. Disabled

pipes will be loaded but they won't be run unless you explicitly

enable them by calling nlp.enable_pipe.

enable (Union[str, Iterable[str]]): Name(s) of pipeline component(s) to enable. All other

pipes will be disabled (and can be enabled using `nlp.enable_pipe`).

exclude (Union[str, Iterable[str]]): Name(s) of pipeline component(s) to exclude. Excluded

components won't be loaded.

config (Dict[str, Any] / Config): Config overrides as nested dict or dict

keyed by section values in dot notation.

RETURNS (Language): The loaded nlp object.

"""

if not model_path.exists():

raise IOError(Errors.E052.format(path=model_path))

if not meta:

meta = get_model_meta(model_path)

config_path = model_path / "config.cfg"

overrides = dict_to_dot(config, for_overrides=True)

config = load_config(config_path, overrides=overrides)

nlp = load_model_from_config(

config,

vocab=vocab,

disable=disable,

enable=enable,

exclude=exclude,

meta=meta,

)

return nlp.from_disk(model_path, exclude=exclude, overrides=overrides)

def load_model_from_config(

config: Union[Dict[str, Any], Config],

*,

meta: Dict[str, Any] = SimpleFrozenDict(),

vocab: Union["Vocab", bool] = True,

disable: Union[str, Iterable[str]] = _DEFAULT_EMPTY_PIPES,

enable: Union[str, Iterable[str]] = _DEFAULT_EMPTY_PIPES,

exclude: Union[str, Iterable[str]] = _DEFAULT_EMPTY_PIPES,

auto_fill: bool = False,

validate: bool = True,

) -> "Language":

"""Create an nlp object from a config. Expects the full config file including

a section "nlp" containing the settings for the nlp object.

name (str): Package name or model path.

meta (Dict[str, Any]): Optional model meta.

vocab (Vocab / True): Optional vocab to pass in on initialization. If True,

a new Vocab object will be created.

disable (Union[str, Iterable[str]]): Name(s) of pipeline component(s) to disable. Disabled

pipes will be loaded but they won't be run unless you explicitly

enable them by calling nlp.enable_pipe.

enable (Union[str, Iterable[str]]): Name(s) of pipeline component(s) to enable. All other

pipes will be disabled (and can be enabled using `nlp.enable_pipe`).

exclude (Union[str, Iterable[str]]): Name(s) of pipeline component(s) to exclude. Excluded

components won't be loaded.

auto_fill (bool): Whether to auto-fill config with missing defaults.

validate (bool): Whether to show config validation errors.

RETURNS (Language): The loaded nlp object.

"""

if "nlp" not in config:

raise ValueError(Errors.E985.format(config=config))

nlp_config = config["nlp"]

if "lang" not in nlp_config or nlp_config["lang"] is None:

raise ValueError(Errors.E993.format(config=nlp_config))

# This will automatically handle all codes registered via the languages

# registry, including custom subclasses provided via entry points

lang_cls = get_lang_class(nlp_config["lang"])

nlp = lang_cls.from_config(

config,

vocab=vocab,

disable=disable,

enable=enable,

exclude=exclude,

auto_fill=auto_fill,

validate=validate,

meta=meta,

)

return nlp

def get_sourced_components(

config: Union[Dict[str, Any], Config],

) -> Dict[str, Dict[str, Any]]:

"""RETURNS (List[str]): All sourced components in the original config,

e.g. {"source": "en_core_web_sm"}. If the config contains a key

"factory", we assume it refers to a component factory.

"""

return {

name: cfg

for name, cfg in config.get("components", {}).items()

if "factory" not in cfg and "source" in cfg

}

def resolve_dot_names(

config: Config, dot_names: List[Optional[str]]

) -> Tuple[Any, ...]:

"""Resolve one or more "dot notation" names, e.g. corpora.train.

The paths could point anywhere into the config, so we don't know which

top-level section we'll be looking within.

We resolve the whole top-level section, although we could resolve less --

we could find the lowest part of the tree.

"""

# TODO: include schema?

resolved = {}

output: List[Any] = []

errors = []

for name in dot_names:

if name is None:

output.append(name)

else:

section = name.split(".")[0]

# We want to avoid resolving the same thing twice

if section not in resolved:

if registry.is_promise(config[section]):

# Otherwise we can't resolve [corpus] if it's a promise

result = registry.resolve({"config": config[section]})["config"]

else:

result = registry.resolve(config[section])

resolved[section] = result

try:

output.append(dot_to_object(resolved, name)) # type: ignore[arg-type]

except KeyError:

msg = f"not a valid section reference: {name}"

errors.append({"loc": name.split("."), "msg": msg})

if errors:

raise ConfigValidationError(config=config, errors=errors)

return tuple(output)

def load_model_from_init_py(

init_file: Union[Path, str],

*,

vocab: Union["Vocab", bool] = True,

disable: Union[str, Iterable[str]] = _DEFAULT_EMPTY_PIPES,

enable: Union[str, Iterable[str]] = _DEFAULT_EMPTY_PIPES,

exclude: Union[str, Iterable[str]] = _DEFAULT_EMPTY_PIPES,

config: Union[Dict[str, Any], Config] = SimpleFrozenDict(),

) -> "Language":

"""Helper function to use in the `load()` method of a model package's

__init__.py.

vocab (Vocab / True): Optional vocab to pass in on initialization. If True,

a new Vocab object will be created.

disable (Union[str, Iterable[str]]): Name(s) of pipeline component(s) to disable. Disabled

pipes will be loaded but they won't be run unless you explicitly

enable them by calling nlp.enable_pipe.

enable (Union[str, Iterable[str]]): Name(s) of pipeline component(s) to enable. All other

pipes will be disabled (and can be enabled using `nlp.enable_pipe`).

exclude (Union[str, Iterable[str]]): Name(s) of pipeline component(s) to exclude. Excluded

components won't be loaded.

config (Dict[str, Any] / Config): Config overrides as nested dict or dict

keyed by section values in dot notation.

RETURNS (Language): The loaded nlp object.

"""

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,

vocab=vocab,

meta=meta,

disable=disable,

enable=enable,

exclude=exclude,

config=config,

)

def load_config(

path: Union[str, Path],

overrides: Dict[str, Any] = SimpleFrozenDict(),

interpolate: bool = False,

) -> Config:

"""Load a config file. Takes care of path validation and section order.

path (Union[str, Path]): Path to the config file or "-" to read from stdin.

overrides: (Dict[str, Any]): Config overrides as nested dict or

dict keyed by section values in dot notation.

interpolate (bool): Whether to interpolate and resolve variables.

RETURNS (Config): The loaded config.

"""

config_path = ensure_path(path)

config = Config(section_order=CONFIG_SECTION_ORDER)

if str(config_path) == "-": # read from standard input

return config.from_str(

sys.stdin.read(), overrides=overrides, interpolate=interpolate

)

else:

if not config_path or not config_path.is_file():

raise IOError(Errors.E053.format(path=config_path, name="config file"))

return config.from_disk(

config_path, overrides=overrides, interpolate=interpolate

)

def load_config_from_str(

text: str, overrides: Dict[str, Any] = SimpleFrozenDict(), interpolate: bool = False

):

"""Load a full config from a string. Wrapper around Thinc's Config.from_str.

text (str): The string config to load.

interpolate (bool): Whether to interpolate and resolve variables.

RETURNS (Config): The loaded config.

"""

return Config(section_order=CONFIG_SECTION_ORDER).from_str(

text, overrides=overrides, interpolate=interpolate

)

def get_installed_models() -> List[str]:

"""List all model packages currently installed in the environment.

RETURNS (List[str]): The string names of the models.

"""

return list(registry.models.get_all().keys())

def get_package_version(name: str) -> Optional[str]:

"""Get the version of an installed package. Typically used to get model

package versions.

name (str): The name of the installed Python package.

RETURNS (str / None): The version or None if package not installed.

"""

try:

return importlib_metadata.version(name) # type: ignore[attr-defined]

except importlib_metadata.PackageNotFoundError: # type: ignore[attr-defined]

return None

def is_compatible_version(

version: str, constraint: str, prereleases: bool = True

) -> Optional[bool]:

"""Check if a version (e.g. "2.0.0") is compatible given a version

constraint (e.g. ">=1.9.0,<2.2.1"). If the constraint is a specific version,

it's interpreted as =={version}.

version (str): The version to check.

constraint (str): The constraint string.

prereleases (bool): Whether to allow prereleases. If set to False,

prerelease versions will be considered incompatible.

RETURNS (bool / None): Whether the version is compatible, or None if the

version or constraint are invalid.

"""

# Handle cases where exact version is provided as constraint

if constraint[0].isdigit():

constraint = f"=={constraint}"

try:

spec = SpecifierSet(constraint)

version = Version(version) # type: ignore[assignment]

except (InvalidSpecifier, InvalidVersion):

return None

spec.prereleases = prereleases

return version in spec

def is_unconstrained_version(

constraint: str, prereleases: bool = True

) -> Optional[bool]:

# We have an exact version, this is the ultimate constrained version

if constraint[0].isdigit():

return False

try:

spec = SpecifierSet(constraint)

except InvalidSpecifier:

return None

spec.prereleases = prereleases

specs = [sp for sp in spec]

# We only have one version spec and it defines > or >=

if len(specs) == 1 and specs[0].operator in (">", ">="):

return True

# One specifier is exact version

if any(sp.operator in ("==") for sp in specs):

return False

has_upper = any(sp.operator in ("<", "<=") for sp in specs)

has_lower = any(sp.operator in (">", ">=") for sp in specs)

# We have a version spec that defines an upper and lower bound

if has_upper and has_lower:

return False

# Everything else, like only an upper version, only a lower version etc.

return True

def split_requirement(requirement: str) -> Tuple[str, str]:

"""Split a requirement like spacy>=1.2.3 into ("spacy", ">=1.2.3")."""

req = Requirement(requirement)

return (req.name, str(req.specifier))

def get_minor_version_range(version: str) -> str:

"""Generate a version range like >=1.2.3,<1.3.0 based on a given version

(e.g. of spaCy).

"""

release = Version(version).release

return f">={version},<{release[0]}.{release[1] + 1}.0"

def get_model_lower_version(constraint: str) -> Optional[str]:

"""From a version range like >=1.2.3,<1.3.0 return the lower pin."""

try:

specset = SpecifierSet(constraint)

for spec in specset:

if spec.operator in (">=", "==", "~="):

return spec.version

except Exception:

pass

return None

def is_prerelease_version(version: str) -> bool:

"""Check whether a version is a prerelease version.

version (str): The version, e.g. "3.0.0.dev1".

RETURNS (bool): Whether the version is a prerelease version.

"""

return Version(version).is_prerelease

def get_base_version(version: str) -> str:

"""Generate the base version without any prerelease identifiers.

version (str): The version, e.g. "3.0.0.dev1".

RETURNS (str): The base version, e.g. "3.0.0".

"""

return Version(version).base_version

def get_minor_version(version: str) -> Optional[str]:

"""Get the major + minor version (without patch or prerelease identifiers).

version (str): The version.

RETURNS (str): The major + minor version or None if version is invalid.

"""

try:

v = Version(version)

except (TypeError, InvalidVersion):

return None

return f"{v.major}.{v.minor}"

def is_minor_version_match(version_a: str, version_b: str) -> bool:

"""Compare two versions and check if they match in major and minor, without

patch or prerelease identifiers. Used internally for compatibility checks

that should be insensitive to patch releases.

version_a (str): The first version

version_b (str): The second version.

RETURNS (bool): Whether the versions match.

"""

a = get_minor_version(version_a)

b = get_minor_version(version_b)

return a is not None and b is not None and a == b

def load_meta(path: Union[str, Path]) -> Dict[str, Any]:

"""Load a model meta.json from a path and validate its contents.

path (Union[str, Path]): Path to meta.json.

RETURNS (Dict[str, Any]): The loaded meta.

"""

path = ensure_path(path)

if not path.parent.exists():

raise IOError(Errors.E052.format(path=path.parent))

if not path.exists() or not path.is_file():

raise IOError(Errors.E053.format(path=path.parent, name="meta.json"))

meta = srsly.read_json(path)

for setting in ["lang", "name", "version"]:

if setting not in meta or not meta[setting]:

raise ValueError(Errors.E054.format(setting=setting))

if "spacy_version" in meta:

if not is_compatible_version(about.__version__, meta["spacy_version"]):

lower_version = get_model_lower_version(meta["spacy_version"])

lower_version = get_base_version(lower_version) # type: ignore[arg-type]

if lower_version is not None:

lower_version = "v" + lower_version

elif "spacy_git_version" in meta:

lower_version = "git commit " + meta["spacy_git_version"]

else:

lower_version = "version unknown"

warn_msg = Warnings.W095.format(

model=f"{meta['lang']}_{meta['name']}",

model_version=meta["version"],

version=lower_version,

current=about.__version__,

)

warnings.warn(warn_msg)

if is_unconstrained_version(meta["spacy_version"]):

warn_msg = Warnings.W094.format(

model=f"{meta['lang']}_{meta['name']}",

model_version=meta["version"],

version=meta["spacy_version"],

example=get_minor_version_range(about.__version__),

)

warnings.warn(warn_msg)

return meta

def get_model_meta(path: Union[str, Path]) -> Dict[str, Any]:

"""Get model meta.json from a directory path and validate its contents.

path (str / Path): Path to model directory.

RETURNS (Dict[str, Any]): The model's meta data.

"""

model_path = ensure_path(path)

return load_meta(model_path / "meta.json")

def is_package(name: str) -> bool:

"""Check if string maps to a package installed via pip.

name (str): Name of package.

RETURNS (bool): True if installed package, False if not.

"""

try:

importlib_metadata.distribution(name) # type: ignore[attr-defined]