332 lines
12 KiB
Python
332 lines
12 KiB
Python
"""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.
|
|
"""
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from fluksio.flow.messages import DType, MessageSpec
|
|
from fluksio.flow.nodes.base import Node
|
|
from fluksio.flow.nodes.wren import CANT, WrenNode, _plain, _value
|
|
from fluksio.flow.pipeline import Pipeline
|
|
|
|
|
|
@dataclass
|
|
class FakeCall:
|
|
tool: str
|
|
args: dict[str, Any] = field(default_factory=dict)
|
|
delay_s: float | None = None
|
|
|
|
|
|
def _node(*targets: tuple[str, DType], **params: object) -> WrenNode:
|
|
node = WrenNode(
|
|
requires=[MessageSpec(name="command", port="prompt", dtype=DType.STR)],
|
|
provides=[
|
|
MessageSpec(name="actions", dtype=DType.LIST),
|
|
*(MessageSpec(name=port, dtype=dtype) for port, dtype in targets),
|
|
],
|
|
params=params,
|
|
)
|
|
node.assign_flow("house", "control")
|
|
return node
|
|
|
|
|
|
def test_targets_of_one_area_and_kind_are_one_device(fake_wren):
|
|
node = _node(
|
|
("living_room_setpoint", DType.FLOAT),
|
|
("living_room_boost", DType.BOOL),
|
|
("kitchen_lamp", DType.BOOL),
|
|
("office_plug", DType.BOOL),
|
|
)
|
|
assert node.devices == {
|
|
("living room", "heat"): ["living_room_setpoint", "living_room_boost"],
|
|
("kitchen", "light"): ["kitchen_lamp"],
|
|
("office", "switch"): ["office_plug"],
|
|
}
|
|
|
|
|
|
def test_a_kind_the_model_does_not_know_is_refused(fake_wren, monkeypatch):
|
|
monkeypatch.setattr(fake_wren, "KINDS", ["light"])
|
|
with pytest.raises(ValueError, match="switch"):
|
|
_node(("office_plug", DType.BOOL))
|
|
|
|
|
|
@pytest.mark.parametrize("port", ["hallway_light", "kitchen_fridge", "light"])
|
|
def test_a_port_naming_no_trained_device_is_refused(fake_wren, port):
|
|
with pytest.raises(ValueError, match="living_room") as refused:
|
|
_node((port, DType.BOOL))
|
|
assert port in str(refused.value)
|
|
|
|
|
|
def test_a_prompt_is_required(fake_wren):
|
|
with pytest.raises(ValueError, match="prompt"):
|
|
WrenNode(provides=[MessageSpec(name="kitchen_light", dtype=DType.BOOL)])
|
|
|
|
|
|
def test_only_the_prompt_wakes_it(fake_wren):
|
|
node = WrenNode(
|
|
requires=[
|
|
MessageSpec(name="command", port="prompt", dtype=DType.STR),
|
|
MessageSpec(name="outside", port="outdoor", dtype=DType.FLOAT),
|
|
],
|
|
)
|
|
assert {spec.port: spec.trigger for spec in node.input_ports} == {
|
|
"prompt": True,
|
|
"outdoor": 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"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("tool", "args", "dtype", "value"),
|
|
[
|
|
("lights.on", {}, DType.BOOL, True),
|
|
("lights.on", {}, DType.FLOAT, 100.0),
|
|
("lights.off", {}, DType.STR, "off"),
|
|
("heating.set", {"temp": 22}, DType.FLOAT, 22.0),
|
|
("heating.set", {"temp": 22}, DType.BOOL, None),
|
|
("heating.boost", {}, DType.BOOL, True),
|
|
("heating.boost", {}, DType.INT, None),
|
|
("switches.off", {}, DType.INT, 0),
|
|
("sensors.temperature", {"area": "kitchen"}, DType.FLOAT, None),
|
|
],
|
|
)
|
|
def test_a_call_means_what_its_target_type_can_hold(tool, args, dtype, value):
|
|
assert _value(FakeCall(tool, args), dtype) == value
|
|
|
|
|
|
def test_applied_calls_reach_their_targets(fake_wren):
|
|
node = _node(
|
|
("living_room_setpoint", DType.FLOAT),
|
|
("living_room_boost", DType.BOOL),
|
|
apply=True,
|
|
)
|
|
Pipeline([node]).state["house.living_room_setpoint"] = 21.0
|
|
fake_wren.answer = (
|
|
[
|
|
FakeCall("heating.set", {"id": "living room:heat", "temp": 23}),
|
|
# The living room's light is filled in for the model, not wired.
|
|
FakeCall("lights.on", {"id": "living room:light"}),
|
|
],
|
|
0.98,
|
|
)
|
|
|
|
out = node.execute({"house.command": "set the heating to 23 degrees"})
|
|
|
|
assert out is not None
|
|
assert out["house.living_room_setpoint"] == 23.0
|
|
assert "house.living_room_boost" not in out
|
|
assert [(a["message"], a["applied"]) for a in out["house.actions"]] == [
|
|
("house.living_room_setpoint", True),
|
|
("", False),
|
|
]
|
|
# Areas nothing is wired to are shown too, so a call can name them.
|
|
assert set(fake_wren.houses[0].areas) == set(fake_wren.AREAS)
|
|
living = fake_wren.houses[0].areas["living room"]
|
|
assert [(d.kind, d.state) for d in living] == [
|
|
("light", "off"),
|
|
("media", "off"),
|
|
("heat", "21C"),
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("params", "confidence"),
|
|
[({}, 0.99), ({"apply": True}, 0.5), ({"apply": True, "min_confidence": 1}, 0.99)],
|
|
)
|
|
def test_nothing_is_written_unless_applied_and_sure(fake_wren, params, confidence):
|
|
node = _node(("kitchen_light", DType.BOOL), **params)
|
|
fake_wren.answer = ([FakeCall("lights.on", {"id": "kitchen:light"})], confidence)
|
|
|
|
out = node.execute({"house.command": "turn on the lights in the kitchen"})
|
|
|
|
assert out == {
|
|
"house.actions": [
|
|
{
|
|
"message": "house.kitchen_light",
|
|
"value": True,
|
|
"applied": False,
|
|
"delay_s": None,
|
|
"call": "lights.on()",
|
|
"confidence": confidence,
|
|
}
|
|
]
|
|
}
|
|
|
|
|
|
def test_the_model_runs_on_one_blas_thread(fake_wren, monkeypatch):
|
|
"""About twice as fast for a model this small; numpy is loaded by `mlp`."""
|
|
from threadpoolctl import threadpool_info
|
|
|
|
threads: list[int] = []
|
|
act = fake_wren.act
|
|
|
|
def counted(command: str, house: Any) -> tuple[list[Any], float]:
|
|
threads.extend(
|
|
pool["num_threads"]
|
|
for pool in threadpool_info()
|
|
if pool["user_api"] == "blas"
|
|
)
|
|
return act(command, house)
|
|
|
|
monkeypatch.setattr(fake_wren, "act", counted)
|
|
_node(("kitchen_lamp", DType.BOOL)).execute({"house.command": "lights on"})
|
|
|
|
assert threads
|
|
assert set(threads) == {1}
|
|
|
|
|
|
def test_a_reply_says_what_was_applied_and_the_rest_is_unhandled(fake_wren):
|
|
node = WrenNode(
|
|
requires=[MessageSpec(name="command", port="prompt", dtype=DType.STR)],
|
|
provides=[
|
|
MessageSpec(name="kitchen_light", dtype=DType.BOOL),
|
|
MessageSpec(name="reply", dtype=DType.STR),
|
|
MessageSpec(name="unhandled", dtype=DType.STR),
|
|
],
|
|
params={"apply": True},
|
|
)
|
|
node.assign_flow("house", "control")
|
|
fake_wren.answer = ([FakeCall("lights.on", {"id": "kitchen:light"})], 0.99)
|
|
|
|
out = node.execute({"house.command": "turn on the kitchen light"})
|
|
|
|
assert out == {"house.kitchen_light": True, "house.reply": "Kitchen light on."}
|
|
|
|
# Nothing wired for the call, or nothing to call at all. Nothing reads
|
|
# what is left, so the node says so itself.
|
|
asked = "what is the weather tomorrow"
|
|
for calls in ([FakeCall("lights.on", {"id": "garage:light"})], []):
|
|
fake_wren.answer = (calls, 0.99)
|
|
out = node.execute({"house.command": asked})
|
|
assert out == {"house.unhandled": asked, "house.reply": CANT}
|
|
|
|
# An assistant reading it answers instead.
|
|
unhandled = MessageSpec(name="house.unhandled", port="unhandled", dtype=DType.STR)
|
|
Pipeline([node, Node(lambda params, unhandled: None, requires=[unhandled])])
|
|
assert node.execute({"house.command": asked}) == {"house.unhandled": asked}
|
|
|
|
|
|
def test_a_call_on_an_area_the_command_does_not_name_is_not_applied(fake_wren):
|
|
"""A word the model never learned lands on a neighbour it was shown."""
|
|
node = _node(("bathroom_light", DType.BOOL), apply=True)
|
|
fake_wren.answer = ([FakeCall("lights.on", {"id": "bathroom:light"})], 0.97)
|
|
|
|
out = node.execute({"house.command": "switch on the appliances"})
|
|
|
|
assert out is not None
|
|
assert "house.bathroom_light" not in out
|
|
assert [a["applied"] for a in out["house.actions"]] == [False]
|
|
|
|
|
|
def _speaking(*targets: tuple[str, DType], inputs=(), **params: object) -> WrenNode:
|
|
node = WrenNode(
|
|
requires=[MessageSpec(name="command", port="prompt", dtype=DType.STR), *inputs],
|
|
provides=[
|
|
*(MessageSpec(name=port, dtype=dtype) for port, dtype in targets),
|
|
MessageSpec(name="reply", dtype=DType.STR),
|
|
MessageSpec(name="unhandled", dtype=DType.STR),
|
|
],
|
|
params={"apply": True, **params},
|
|
)
|
|
node.assign_flow("house", "control")
|
|
return node
|
|
|
|
|
|
def test_the_house_words_and_number_words_reach_the_model_as_its_own(fake_wren):
|
|
node = _speaking(("patio_light", DType.BOOL), aliases={"Page Too": "patio"})
|
|
fake_wren.answer = ([], 0.99)
|
|
|
|
out = node.execute({"house.command": "Turn on the page too light in five minutes."})
|
|
|
|
assert fake_wren.commands == ["turn on the patio light in 5 minutes"]
|
|
# The assistant reads the house's words too, as they were said otherwise.
|
|
assert out == {
|
|
"house.unhandled": "Turn on the patio light in five minutes.",
|
|
"house.reply": CANT,
|
|
}
|
|
|
|
|
|
def test_a_command_for_all_of_a_kind_reaches_every_wired_one(fake_wren):
|
|
lights = [(f"{area}_light", DType.BOOL) for area in ("kitchen", "office")]
|
|
node = _speaking(*lights, ("office_shutter", DType.STR))
|
|
|
|
out = node.execute({"house.command": "Turn off all lights."})
|
|
assert out == {
|
|
"house.kitchen_light": False,
|
|
"house.office_light": False,
|
|
"house.reply": "All lights off.",
|
|
}
|
|
out = node.execute({"house.command": "turn everything off in the office"})
|
|
assert out == {
|
|
"house.office_light": False,
|
|
"house.reply": "Lights off in the office.",
|
|
}
|
|
assert fake_wren.commands == [] # the node's, not the model's
|
|
|
|
# A question or an exception is the model's again.
|
|
fake_wren.answer = ([], 0.99)
|
|
node.execute({"house.command": "are all lights off"})
|
|
node.execute({"house.command": "turn off all lights but the kitchen"})
|
|
assert len(fake_wren.commands) == 2
|
|
|
|
|
|
def test_a_question_for_a_temperature_is_answered_from_its_reading(fake_wren):
|
|
node = _speaking(
|
|
inputs=[
|
|
MessageSpec(name="outside", port="outdoor", dtype=DType.FLOAT),
|
|
MessageSpec(
|
|
name="kitchen_temp", port="kitchen_temperature", dtype=DType.FLOAT
|
|
),
|
|
]
|
|
)
|
|
assert node.readings == {"kitchen": "kitchen_temperature"}
|
|
readings = {"house.outside": 7.6, "house.kitchen_temp": 21.2}
|
|
|
|
for area, said in [("garden", "outside"), ("kitchen", "in the kitchen")]:
|
|
fake_wren.answer = ([FakeCall("sensors.temperature", {"area": area})], 0.99)
|
|
command = f"what is the temperature in the {area}"
|
|
out = node.execute({"house.command": command, **readings})
|
|
degrees = 8 if area == "garden" else 21
|
|
assert out == {"house.reply": f"It's {degrees} degrees {said}."}
|
|
|
|
# No reading for the bedroom, or a model unsure of it: the assistant's.
|
|
for area, confidence in [("bedroom", 0.99), ("kitchen", 0.5)]:
|
|
fake_wren.answer = (
|
|
[FakeCall("sensors.temperature", {"area": area})],
|
|
confidence,
|
|
)
|
|
command = f"how warm is it in the {area}"
|
|
out = node.execute({"house.command": command, **readings})
|
|
assert out == {"house.unhandled": command, "house.reply": CANT}
|
|
|
|
|
|
def test_a_brightness_call_sets_the_house_brightness(fake_wren):
|
|
node = _speaking(("brightness", DType.INT))
|
|
# Whatever area the call names: the brightness is the house's.
|
|
for call, level in [
|
|
(FakeCall("lights.set", {"id": "kitchen:light", "level": 30}), 30),
|
|
(FakeCall("lights.dim", {"id": "kitchen:light"}), 40),
|
|
(FakeCall("lights.brighten", {"id": "bedroom:light"}), 100),
|
|
]:
|
|
fake_wren.answer = ([call], 0.95)
|
|
out = node.execute({"house.command": "dim the lights in the kitchen"})
|
|
assert out == {
|
|
"house.brightness": level,
|
|
"house.reply": f"Brightness at {level} percent.",
|
|
}
|
|
|
|
# Nothing wired for it: the assistant's.
|
|
node = _speaking(("kitchen_light", DType.BOOL))
|
|
fake_wren.answer = ([FakeCall("lights.dim", {"id": "kitchen:light"})], 0.95)
|
|
out = node.execute({"house.command": "dim the lights"})
|
|
assert out == {"house.unhandled": "dim the lights", "house.reply": CANT}
|