"""The wren node: its wiring is the allowlist, and nothing is written unasked.

The model is an optional package, so ``fake_wren`` (conftest) answers in its
place: what the node does with an answer is what is under test here.
"""

import asyncio
from types import SimpleNamespace
from typing import Any

import pytest

from fluksio.flow.artifacts import ArtifactStore
from fluksio.flow.messages import DType, MessageSpec
from fluksio.flow.nodes.base import Node
from fluksio.flow.nodes.wren import CANT, WrenNode, _plain
from fluksio.flow.pipeline import Pipeline

PROMPT = MessageSpec(name="command", port="prompt", dtype=DType.STR)
OUTDOOR = MessageSpec(name="outdoor", port="outdoor_temperature", dtype=DType.FLOAT)
LIGHT = MessageSpec(name="kitchen_light", dtype=DType.BOOL)


def _node(
    fake_wren,
    *targets: tuple[str, DType],
    inputs=(),
    saying=False,
    trained=None,
    health=None,
    app=None,
    **params: Any,
) -> WrenNode:
    """A started node, its model trained on its own targets unless ``trained``
    lists others."""
    node = WrenNode(
        requires=[PROMPT, *inputs],
        provides=[
            MessageSpec(name="actions", dtype=DType.LIST),
            *(MessageSpec(name=port, dtype=dtype) for port, dtype in targets),
            *(
                MessageSpec(name=port, dtype=DType.STR)
                for port in ("reply", "unhandled")
                if saying
            ),
        ],
        params=params,
    )
    node.assign_flow("house", "control")
    if health is not None:
        node._on_health = lambda _node, status, detail: health.append((status, detail))
    fake_wren.pack["meta"]["targets"] = (
        list(node.targets.values()) if trained is None else trained
    )
    asyncio.run(node.start(app))
    return node


def _record(port: str, value: Any, call: str, **more: Any) -> dict[str, Any]:
    return {
        "message": f"house.{port}",
        "value": value,
        "applied": True,
        "delay_s": None,
        "call": call,
        "confidence": 1.0,
    } | more


@pytest.mark.parametrize(
    ("requires", "provides", "params", "refusal"),
    [
        ([], [LIGHT], {}, "prompt"),
        ([PROMPT], [MessageSpec(name="plan", dtype=DType.JSON)], {}, "plan"),
        ([PROMPT], [LIGHT], {"targets": {"hall_light": {}}}, "hall_light"),
        (
            [PROMPT],
            [LIGHT],
            {"targets": {"kitchen_light": {"values": ["ON"]}}},
            "values",
        ),
        ([PROMPT], [LIGHT], {"targets": {"kitchen_light": {"range": [0, 1]}}}, "range"),
    ],
)
def test_a_wiring_wren_cannot_act_on_is_refused(
    fake_wren, requires, provides, params, refusal
):
    with pytest.raises(ValueError, match=refusal):
        WrenNode(requires=requires, provides=provides, params=params)


def test_only_the_prompt_wakes_it(fake_wren):
    node = WrenNode(requires=[PROMPT, OUTDOOR])
    assert {spec.port: spec.trigger for spec in node.input_ports} == {
        "prompt": True,
        "outdoor_temperature": False,
    }


def test_the_model_reads_the_command_plain():
    assert _plain("Turn on the kitchen light.") == "turn on the kitchen light"
    assert _plain("Heat it to 22.5, please!") == "heat it to 22.5 please"


def test_the_targets_reach_the_model_with_their_values_and_settings(fake_wren):
    node = _node(
        fake_wren,
        ("kitchen_light", DType.BOOL),
        ("living_room_heating", DType.FLOAT),
        ("kitchen_shutter", DType.STR),
        inputs=[OUTDOOR],
        targets={
            "living_room_heating": {"range": [5, 30], "unit": "°C"},
            "kitchen_shutter": {"values": ["UP", "DOWN"]},
            "outdoor_temperature": {"description": "temperature outside"},
        },
    )
    Pipeline([node]).state["house.kitchen_light"] = True

    node.execute({"house.command": "warm it up", "house.outdoor": 6.2})

    assert fake_wren.loaded == [fake_wren.runtime.MODEL]
    assert fake_wren.models == [fake_wren.pack]
    # Targets nothing has set yet are off, 0 within their range, or their
    # first value.
    target = fake_wren.Target
    assert fake_wren.targets[0] == [
        target("kitchen_light", "bool", True),
        target("living_room_heating", "float", 5.0, range=[5.0, 30.0], unit="°C"),
        target("kitchen_shutter", "str", "UP", values=["UP", "DOWN"]),
        target(
            "outdoor_temperature",
            "float",
            6.2,
            description="temperature outside",
            readonly=True,
        ),
    ]


def test_applied_calls_reach_their_targets_and_the_reply_says_so(fake_wren):
    node = _node(
        fake_wren,
        ("kitchen_light", DType.BOOL),
        ("living_room_heating", DType.FLOAT),
        saying=True,
        apply=True,
        targets={"living_room_heating": {"range": [5, 30], "unit": "°C"}},
    )
    # 0.95 clears the model's own threshold, 0.9 here.
    fake_wren.answer = [
        fake_wren.Call("kitchen_light", True, confidence=0.95),
        fake_wren.Call("living_room_heating", 21),
    ]

    out = node.execute({"house.command": "kitchen light on and the heating to 21"})

    assert out == {
        "house.kitchen_light": True,
        "house.living_room_heating": 21.0,
        "house.reply": "Kitchen light on, living room heating to 21 °C.",
        "house.actions": [
            _record("kitchen_light", True, "set(kitchen_light, true)", confidence=0.95),
            _record("living_room_heating", 21.0, "set(living_room_heating, 21)"),
        ],
    }
    assert isinstance(out["house.living_room_heating"], float)


@pytest.mark.parametrize(
    ("params", "confidence"),
    [
        ({}, 0.99),
        # Below the model's own threshold, 0.9 here.
        ({"apply": True}, 0.85),
        ({"apply": True, "min_confidence": 1}, 0.99),
    ],
)
def test_nothing_is_written_unless_applied_and_sure(fake_wren, params, confidence):
    node = _node(fake_wren, ("kitchen_light", DType.BOOL), **params)
    fake_wren.answer = [fake_wren.Call("kitchen_light", True, confidence=confidence)]

    out = node.execute({"house.command": "turn on the kitchen light"})

    assert out == {
        "house.actions": [
            _record(
                "kitchen_light",
                True,
                "set(kitchen_light, true)",
                applied=False,
                confidence=confidence,
            )
        ]
    }


@pytest.mark.parametrize(
    ("params", "applied"),
    [
        ({}, True),
        # One call short of its target's bar stops the whole command.
        ({"targets": {"kitchen_light": {"min_confidence": 0.99}}}, False),
        ({"targets": {"kitchen_light": {"pass_on": True}}}, False),
        ({"min_confidence": 0.99}, False),
        # A target's own bar is taken over the node's.
        (
            {
                "min_confidence": 0.99,
                "targets": {
                    "kitchen_light": {"min_confidence": 0.9},
                    "garden_light": {"min_confidence": 0.9},
                },
            },
            True,
        ),
    ],
)
def test_a_command_is_applied_whole_or_not_at_all(fake_wren, params, applied):
    node = _node(
        fake_wren,
        ("kitchen_light", DType.BOOL),
        ("garden_light", DType.BOOL),
        saying=True,
        apply=True,
        **params,
    )
    fake_wren.answer = [
        fake_wren.Call("garden_light", True, confidence=0.95),
        fake_wren.Call("kitchen_light", True, confidence=0.95),
    ]
    command = "garden and kitchen lights on"

    out = node.execute({"house.command": command})

    assert out is not None
    assert [a["applied"] for a in out.pop("house.actions")] == [applied, applied]
    assert out == (
        {
            "house.garden_light": True,
            "house.kitchen_light": True,
            "house.reply": "Garden light on, kitchen light on.",
        }
        if applied
        else {"house.unhandled": command, "house.reply": CANT}
    )


def test_a_target_the_model_was_not_trained_on_is_never_applied(fake_wren):
    trained = [
        # As the pack writes it: [5, 30] is the node's (5.0, 30.0), and the
        # unit is not the model's to know.
        {
            "name": "living_room_heating",
            "type": "float",
            "values": None,
            "range": [5, 30],
            "description": "",
            "unit": None,
            "readonly": False,
        },
        # Described otherwise since.
        {
            "name": "canopy",
            "type": "str",
            "values": ["UP", "DOWN", "STOP"],
            "range": None,
            "description": "awning",
            "unit": None,
            "readonly": False,
        },
    ]
    health: list[tuple[str, str | None]] = []
    node = _node(
        fake_wren,
        ("living_room_heating", DType.FLOAT),
        ("canopy", DType.STR),
        ("garden_light", DType.BOOL),
        saying=True,
        trained=trained,
        health=health,
        apply=True,
        targets={
            "living_room_heating": {"range": [5, 30], "unit": "°C"},
            "canopy": {
                "values": ["UP", "DOWN", "STOP"],
                "description": "awning over the patio",
            },
        },
    )

    ((status, detail),) = health
    assert status == "degraded"
    assert detail is not None and "retraining for canopy, garden_light" in detail

    fake_wren.answer = [fake_wren.Call("canopy", "DOWN")]
    out = node.execute({"house.command": "close the awning"})
    # The model is not shown the targets it never learnt.
    assert [t.name for t in fake_wren.targets[0]] == ["living_room_heating"]
    assert out is not None
    assert [a["applied"] for a in out["house.actions"]] == [False]
    assert out["house.unhandled"] == "close the awning"


def test_a_read_is_answered_from_the_target_value(fake_wren):
    node = _node(
        fake_wren,
        inputs=[OUTDOOR],
        saying=True,
        # Not applied: a reading is answered anyway, as it writes nothing.
        targets={"outdoor_temperature": {"unit": "°C"}},
    )
    fake_wren.answer = [fake_wren.Call("outdoor_temperature")]
    command = "how warm is it outside"

    out = node.execute({"house.command": command, "house.outdoor": 6.2})
    assert out == {
        "house.reply": "Outdoor temperature is 6.2 °C.",
        "house.actions": [
            _record("outdoor", 6.2, "read(outdoor_temperature)"),
        ],
    }

    # Nothing has written the reading yet: there is no answer to give.
    out = node.execute({"house.command": command})
    assert out is not None
    assert out["house.unhandled"] == command


def test_a_read_beside_a_set_not_written_goes_on_whole(fake_wren):
    node = _node(
        fake_wren, ("kitchen_light", DType.BOOL), inputs=[OUTDOOR], saying=True
    )
    fake_wren.answer = [
        fake_wren.Call("outdoor_temperature"),
        fake_wren.Call("kitchen_light", True),
    ]
    command = "how warm is it outside and turn on the kitchen light"

    out = node.execute({"house.command": command, "house.outdoor": 6.2})

    assert out is not None
    assert [a["applied"] for a in out["house.actions"]] == [False, False]
    assert out["house.unhandled"] == command
    assert "house.reply" not in out or out["house.reply"] == CANT


def test_a_delayed_call_is_held_by_the_work_queue(fake_wren, monkeypatch):
    node = _node(fake_wren, ("kitchen_light", DType.BOOL), saying=True, apply=True)
    deferred: list[tuple[dict[str, Any], float]] = []

    def defer(self, node, outputs, seconds):
        deferred.append((outputs, seconds))
        return True

    monkeypatch.setattr(Pipeline, "defer", defer)
    Pipeline([node])
    fake_wren.answer = [
        fake_wren.Call("kitchen_light", False),
        fake_wren.Call("kitchen_light", True, delay_s=420),
    ]

    out = node.execute({"house.command": "kitchen light off for 7 minutes"})

    assert deferred == [({"house.kitchen_light": True}, 420)]
    assert out is not None
    assert out["house.kitchen_light"] is False
    assert out["house.reply"] == "Kitchen light off, kitchen light on in 7 minutes."


def test_a_trained_model_is_read_from_the_artifact_store(fake_wren, tmp_path):
    store = ArtifactStore(tmp_path)
    digest = store.put([b"WREN"])["digest"]
    app = SimpleNamespace(state=SimpleNamespace(artifact_store=store))
    health: list[tuple[str, str | None]] = []

    _node(
        fake_wren, ("kitchen_light", DType.BOOL), app=app, health=health, model=digest
    )
    assert fake_wren.loaded == [store.path(digest)]
    assert health == [("ok", None)]

    # One the store does not have: every command goes on.
    node = _node(
        fake_wren,
        ("kitchen_light", DType.BOOL),
        saying=True,
        app=app,
        health=health,
        apply=True,
        model="sha256:" + "0" * 64,
    )
    assert health[-1][0] == "degraded"
    out = node.execute({"house.command": "kitchen light on"})
    assert out == {
        "house.actions": [],
        "house.unhandled": "kitchen light on",
        "house.reply": CANT,
    }
    assert fake_wren.commands == []


def test_the_model_runs_on_one_blas_thread(fake_wren, monkeypatch):
    """About twice as fast for a model this small."""
    from threadpoolctl import threadpool_info

    # BLAS is only listed once something has loaded it, as wren's numpy does.
    pytest.importorskip("numpy")

    threads: list[int] = []
    act = fake_wren.act

    def counted(command: str, targets: Any, *, model: Any = None) -> Any:
        threads.extend(
            pool["num_threads"]
            for pool in threadpool_info()
            if pool["user_api"] == "blas"
        )
        return act(command, targets, model=model)

    monkeypatch.setattr(fake_wren, "act", counted)
    node = _node(fake_wren, ("kitchen_light", DType.BOOL))
    node.execute({"house.command": "lights on"})

    assert threads
    assert set(threads) == {1}


def test_own_words_and_number_words_reach_the_model_as_its_own(fake_wren):
    node = _node(
        fake_wren,
        ("garden_light", DType.BOOL),
        saying=True,
        apply=True,
        aliases={"Page Too": "garden"},
    )

    out = node.execute({"house.command": "Turn on the page too light in five minutes."})

    assert fake_wren.commands == ["turn on the garden light in 5 minutes"]
    # The assistant reads the own words too, as they were said otherwise.
    passed = "Turn on the garden light in five minutes."
    assert out == {
        "house.actions": [],
        "house.unhandled": passed,
        "house.reply": CANT,
    }

    # An assistant reading `unhandled` answers instead.
    unhandled = MessageSpec(name="house.unhandled", port="unhandled", dtype=DType.STR)
    Pipeline([node, Node(lambda params, unhandled: None, requires=[unhandled])])
    out = node.execute({"house.command": "Turn on the page too light in five minutes."})
    assert out == {"house.actions": [], "house.unhandled": passed}