import pytest
import os
import ctypes
from pathlib import Path
from spacy.about import __version__ as spacy_version
from spacy import util
from spacy import prefer_gpu, require_gpu, require_cpu
from spacy.ml._precomputable_affine import PrecomputableAffine
from spacy.ml._precomputable_affine import _backprop_precomputable_affine_padding
from spacy.util import dot_to_object, SimpleFrozenList, import_file
from thinc.api import Config, Optimizer, ConfigValidationError, get_current_ops
from thinc.api import set_current_ops
from spacy.training.batchers import minibatch_by_words
from spacy.lang.en import English
from spacy.lang.nl import Dutch
from spacy.language import DEFAULT_CONFIG_PATH
from spacy.schemas import ConfigSchemaTraining

from thinc.api import get_current_ops, NumpyOps, CupyOps

from .util import get_random_doc, make_tempdir


@pytest.fixture
def is_admin():
    """Determine if the tests are run as admin or not."""
    try:
        admin = os.getuid() == 0
    except AttributeError:
        admin = ctypes.windll.shell32.IsUserAnAdmin() != 0

    return admin


@pytest.mark.parametrize("text", ["hello/world", "hello world"])
def test_util_ensure_path_succeeds(text):
    path = util.ensure_path(text)
    assert isinstance(path, Path)


@pytest.mark.parametrize(
    "package,result", [("numpy", True), ("sfkodskfosdkfpsdpofkspdof", False)]
)
def test_util_is_package(package, result):
    """Test that an installed package via pip is recognised by util.is_package."""
    assert util.is_package(package) is result


@pytest.mark.parametrize("package", ["thinc"])
def test_util_get_package_path(package):
    """Test that a Path object is returned for a package name."""
    path = util.get_package_path(package)
    assert isinstance(path, Path)


def test_PrecomputableAffine(nO=4, nI=5, nF=3, nP=2):
    model = PrecomputableAffine(nO=nO, nI=nI, nF=nF, nP=nP).initialize()
    assert model.get_param("W").shape == (nF, nO, nP, nI)
    tensor = model.ops.alloc((10, nI))
    Y, get_dX = model.begin_update(tensor)
    assert Y.shape == (tensor.shape[0] + 1, nF, nO, nP)
    dY = model.ops.alloc((15, nO, nP))
    ids = model.ops.alloc((15, nF))
    ids[1, 2] = -1
    dY[1] = 1
    assert not model.has_grad("pad")
    d_pad = _backprop_precomputable_affine_padding(model, dY, ids)
    assert d_pad[0, 2, 0, 0] == 1.0
    ids.fill(0.0)
    dY.fill(0.0)
    dY[0] = 0
    ids[1, 2] = 0
    ids[1, 1] = -1
    ids[1, 0] = -1
    dY[1] = 1
    ids[2, 0] = -1
    dY[2] = 5
    d_pad = _backprop_precomputable_affine_padding(model, dY, ids)
    assert d_pad[0, 0, 0, 0] == 6
    assert d_pad[0, 1, 0, 0] == 1
    assert d_pad[0, 2, 0, 0] == 0


def test_prefer_gpu():
    current_ops = get_current_ops()
    try:
        import cupy  # noqa: F401

        prefer_gpu()
        assert isinstance(get_current_ops(), CupyOps)
    except ImportError:
        assert not prefer_gpu()
    set_current_ops(current_ops)


def test_require_gpu():
    current_ops = get_current_ops()
    try:
        import cupy  # noqa: F401

        require_gpu()
        assert isinstance(get_current_ops(), CupyOps)
    except ImportError:
        with pytest.raises(ValueError):
            require_gpu()
    set_current_ops(current_ops)


def test_require_cpu():
    current_ops = get_current_ops()
    require_cpu()
    assert isinstance(get_current_ops(), NumpyOps)
    try:
        import cupy  # noqa: F401

        require_gpu()
        assert isinstance(get_current_ops(), CupyOps)
    except ImportError:
        pass
    require_cpu()
    assert isinstance(get_current_ops(), NumpyOps)
    set_current_ops(current_ops)


def test_ascii_filenames():
    """Test that all filenames in the project are ASCII.
    See: https://twitter.com/_inesmontani/status/1177941471632211968
    """
    root = Path(__file__).parent.parent
    for path in root.glob("**/*"):
        assert all(ord(c) < 128 for c in path.name), path.name


def test_load_model_blank_shortcut():
    """Test that using a model name like "blank:en" works as a shortcut for
    spacy.blank("en").
    """
    nlp = util.load_model("blank:en")
    assert nlp.lang == "en"
    assert nlp.pipeline == []
    with pytest.raises(ImportError):
        util.load_model("blank:fjsfijsdof")


@pytest.mark.parametrize(
    "version,constraint,compatible",
    [
        (spacy_version, spacy_version, True),
        (spacy_version, f">={spacy_version}", True),
        ("3.0.0", "2.0.0", False),
        ("3.2.1", ">=2.0.0", True),
        ("2.2.10a1", ">=1.0.0,<2.1.1", False),
        ("3.0.0.dev3", ">=1.2.3,<4.5.6", True),
        ("n/a", ">=1.2.3,<4.5.6", None),
        ("1.2.3", "n/a", None),
        ("n/a", "n/a", None),
    ],
)
def test_is_compatible_version(version, constraint, compatible):
    assert util.is_compatible_version(version, constraint) is compatible


@pytest.mark.parametrize(
    "constraint,expected",
    [
        ("3.0.0", False),
        ("==3.0.0", False),
        (">=2.3.0", True),
        (">2.0.0", True),
        ("<=2.0.0", True),
        (">2.0.0,<3.0.0", False),
        (">=2.0.0,<3.0.0", False),
        ("!=1.1,>=1.0,~=1.0", True),
        ("n/a", None),
    ],
)
def test_is_unconstrained_version(constraint, expected):
    assert util.is_unconstrained_version(constraint) is expected


@pytest.mark.parametrize(
    "a1,a2,b1,b2,is_match",
    [
        ("3.0.0", "3.0", "3.0.1", "3.0", True),
        ("3.1.0", "3.1", "3.2.1", "3.2", False),
        ("xxx", None, "1.2.3.dev0", "1.2", False),
    ],
)
def test_minor_version(a1, a2, b1, b2, is_match):
    assert util.get_minor_version(a1) == a2
    assert util.get_minor_version(b1) == b2
    assert util.is_minor_version_match(a1, b1) is is_match
    assert util.is_minor_version_match(a2, b2) is is_match


@pytest.mark.parametrize(
    "dot_notation,expected",
    [
        (
            {"token.pos": True, "token._.xyz": True},
            {"token": {"pos": True, "_": {"xyz": True}}},
        ),
        (
            {"training.batch_size": 128, "training.optimizer.learn_rate": 0.01},
            {"training": {"batch_size": 128, "optimizer": {"learn_rate": 0.01}}},
        ),
    ],
)
def test_dot_to_dict(dot_notation, expected):
    result = util.dot_to_dict(dot_notation)
    assert result == expected
    assert util.dict_to_dot(result) == dot_notation


def test_set_dot_to_object():
    config = {"foo": {"bar": 1, "baz": {"x": "y"}}, "test": {"a": {"b": "c"}}}
    with pytest.raises(KeyError):
        util.set_dot_to_object(config, "foo.bar.baz", 100)
    with pytest.raises(KeyError):
        util.set_dot_to_object(config, "hello.world", 100)
    with pytest.raises(KeyError):
        util.set_dot_to_object(config, "test.a.b.c", 100)
    util.set_dot_to_object(config, "foo.bar", 100)
    assert config["foo"]["bar"] == 100
    util.set_dot_to_object(config, "foo.baz.x", {"hello": "world"})
    assert config["foo"]["baz"]["x"]["hello"] == "world"
    assert config["test"]["a"]["b"] == "c"
    util.set_dot_to_object(config, "foo", 123)
    assert config["foo"] == 123
    util.set_dot_to_object(config, "test", "hello")
    assert dict(config) == {"foo": 123, "test": "hello"}


@pytest.mark.parametrize(
    "doc_sizes, expected_batches",
    [
        ([400, 400, 199], [3]),
        ([400, 400, 199, 3], [4]),
        ([400, 400, 199, 3, 200], [3, 2]),
        ([400, 400, 199, 3, 1], [5]),
        ([400, 400, 199, 3, 1, 1500], [5]),  # 1500 will be discarded
        ([400, 400, 199, 3, 1, 200], [3, 3]),
        ([400, 400, 199, 3, 1, 999], [3, 3]),
        ([400, 400, 199, 3, 1, 999, 999], [3, 2, 1, 1]),
        ([1, 2, 999], [3]),
        ([1, 2, 999, 1], [4]),
        ([1, 200, 999, 1], [2, 2]),
        ([1, 999, 200, 1], [2, 2]),
    ],
)
def test_util_minibatch(doc_sizes, expected_batches):
    docs = [get_random_doc(doc_size) for doc_size in doc_sizes]
    tol = 0.2
    batch_size = 1000
    batches = list(
        minibatch_by_words(docs, size=batch_size, tolerance=tol, discard_oversize=True)
    )
    assert [len(batch) for batch in batches] == expected_batches

    max_size = batch_size + batch_size * tol
    for batch in batches:
        assert sum([len(doc) for doc in batch]) < max_size


@pytest.mark.parametrize(
    "doc_sizes, expected_batches",
    [
        ([400, 4000, 199], [1, 2]),
        ([400, 400, 199, 3000, 200], [1, 4]),
        ([400, 400, 199, 3, 1, 1500], [1, 5]),
        ([400, 400, 199, 3000, 2000, 200, 200], [1, 1, 3, 2]),
        ([1, 2, 9999], [1, 2]),
        ([2000, 1, 2000, 1, 1, 1, 2000], [1, 1, 1, 4]),
    ],
)
def test_util_minibatch_oversize(doc_sizes, expected_batches):
    """ Test that oversized documents are returned in their own batch"""
    docs = [get_random_doc(doc_size) for doc_size in doc_sizes]
    tol = 0.2
    batch_size = 1000
    batches = list(
        minibatch_by_words(docs, size=batch_size, tolerance=tol, discard_oversize=False)
    )
    assert [len(batch) for batch in batches] == expected_batches


def test_util_dot_section():
    cfg_string = """
    [nlp]
    lang = "en"
    pipeline = ["textcat"]

    [components]

    [components.textcat]
    factory = "textcat"

    [components.textcat.model]
    @architectures = "spacy.TextCatBOW.v1"
    exclusive_classes = true
    ngram_size = 1
    no_output_layer = false
    """
    nlp_config = Config().from_str(cfg_string)
    en_nlp = util.load_model_from_config(nlp_config, auto_fill=True)
    default_config = Config().from_disk(DEFAULT_CONFIG_PATH)
    default_config["nlp"]["lang"] = "nl"
    nl_nlp = util.load_model_from_config(default_config, auto_fill=True)
    # Test that creation went OK
    assert isinstance(en_nlp, English)
    assert isinstance(nl_nlp, Dutch)
    assert nl_nlp.pipe_names == []
    assert en_nlp.pipe_names == ["textcat"]
    # not exclusive_classes
    assert en_nlp.get_pipe("textcat").model.attrs["multi_label"] is False
    # Test that default values got overwritten
    assert en_nlp.config["nlp"]["pipeline"] == ["textcat"]
    assert nl_nlp.config["nlp"]["pipeline"] == []  # default value []
    # Test proper functioning of 'dot_to_object'
    with pytest.raises(KeyError):
        dot_to_object(en_nlp.config, "nlp.pipeline.tagger")
    with pytest.raises(KeyError):
        dot_to_object(en_nlp.config, "nlp.unknownattribute")
    T = util.registry.resolve(nl_nlp.config["training"], schema=ConfigSchemaTraining)
    assert isinstance(dot_to_object({"training": T}, "training.optimizer"), Optimizer)


def test_simple_frozen_list():
    t = SimpleFrozenList(["foo", "bar"])
    assert t == ["foo", "bar"]
    assert t.index("bar") == 1  # okay method
    with pytest.raises(NotImplementedError):
        t.append("baz")
    with pytest.raises(NotImplementedError):
        t.sort()
    with pytest.raises(NotImplementedError):
        t.extend(["baz"])
    with pytest.raises(NotImplementedError):
        t.pop()
    t = SimpleFrozenList(["foo", "bar"], error="Error!")
    with pytest.raises(NotImplementedError):
        t.append("baz")


def test_resolve_dot_names():
    config = {
        "training": {"optimizer": {"@optimizers": "Adam.v1"}},
        "foo": {"bar": "training.optimizer", "baz": "training.xyz"},
    }
    result = util.resolve_dot_names(config, ["training.optimizer"])
    assert isinstance(result[0], Optimizer)
    with pytest.raises(ConfigValidationError) as e:
        util.resolve_dot_names(config, ["training.xyz", "training.optimizer"])
    errors = e.value.errors
    assert len(errors) == 1
    assert errors[0]["loc"] == ["training", "xyz"]


def test_import_code():
    code_str = """
from spacy import Language

class DummyComponent:
    def __init__(self, vocab, name):
        pass

    def initialize(self, get_examples, *, nlp, dummy_param: int):
        pass

@Language.factory(
    "dummy_component",
)
def make_dummy_component(
    nlp: Language, name: str
):
    return DummyComponent(nlp.vocab, name)
"""

    with make_tempdir() as temp_dir:
        code_path = os.path.join(temp_dir, "code.py")
        with open(code_path, "w") as fileh:
            fileh.write(code_str)

        import_file("python_code", code_path)
        config = {"initialize": {"components": {"dummy_component": {"dummy_param": 1}}}}
        nlp = English.from_config(config)
        nlp.add_pipe("dummy_component")
        nlp.initialize()