From d8edce55f9f57e06b61e55967ae87c0c6f464643 Mon Sep 17 00:00:00 2001 From: stroblme Date: Mon, 5 Oct 2026 20:58:59 +0200 Subject: [PATCH] Add the wren node: a typed command into values for the devices wired to it Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_013GAeg4Jo6xjZzjwy9QDWMQ --- backend/fluksio/flow/controller.py | 7 + backend/fluksio/flow/nodes/__init__.py | 2 + backend/fluksio/flow/nodes/wren.py | 348 +++++++++++++++++++++++++ backend/fluksio/flow/pipeline.py | 2 +- backend/pyproject.toml | 6 + backend/tests/flow/conftest.py | 46 ++++ backend/tests/flow/test_node_types.py | 7 +- backend/tests/flow/test_wren.py | 145 +++++++++++ 8 files changed, 561 insertions(+), 2 deletions(-) create mode 100644 backend/fluksio/flow/nodes/wren.py create mode 100644 backend/tests/flow/test_wren.py diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index ce3f30d..675c26f 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -54,6 +54,7 @@ from fluksio.flow.nodes import ( RbeNode, SwitchNode, TriggerNode, + WrenNode, ) from fluksio.flow.pipeline import ( ADVISORY_ISSUES, @@ -352,6 +353,12 @@ NODE_TYPES: dict[str, NodeType] = { cls=MLPNode, params_schema=_schema_of(MLPNode), ), + "wren": NodeType( + title="Wren", + description="Turn a typed command into values for the devices wired to it.", + cls=WrenNode, + params_schema=_schema_of(WrenNode), + ), "inject": NodeType( title="Inject", description="Emit a value on request, on a timer, or when the flow starts.", diff --git a/backend/fluksio/flow/nodes/__init__.py b/backend/fluksio/flow/nodes/__init__.py index 6237ae9..808cb4f 100644 --- a/backend/fluksio/flow/nodes/__init__.py +++ b/backend/fluksio/flow/nodes/__init__.py @@ -16,6 +16,7 @@ from fluksio.flow.nodes.mlp import MLPNode from fluksio.flow.nodes.mqtt import MqttNode from fluksio.flow.nodes.ntfy import NtfyNode from fluksio.flow.nodes.trigger import TriggerNode +from fluksio.flow.nodes.wren import WrenNode __all__ = [ "ChangeNode", @@ -34,4 +35,5 @@ __all__ = [ "RbeNode", "SwitchNode", "TriggerNode", + "WrenNode", ] diff --git a/backend/fluksio/flow/nodes/wren.py b/backend/fluksio/flow/nodes/wren.py new file mode 100644 index 0000000..1f9e2c7 --- /dev/null +++ b/backend/fluksio/flow/nodes/wren.py @@ -0,0 +1,348 @@ +"""Wren: a typed command in, values for the devices wired to it out. + +`Wren `_ is a small language model trained on +one job: read a house — its areas, the devices in each and their states — and +answer a command with calls on those devices. This node is the bridge. Its +output ports *are* the devices, named ``_``, so the wiring is the +allowlist: the model can reach nothing that is not wired to the node. +""" + +from __future__ import annotations + +import time +from collections.abc import Iterable +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field + +from fluksio.flow.messages import DType, MessageSpec +from fluksio.flow.nodes.base import Node + + +def _wren() -> Any: + """The model package, which is optional. + + Imported when a node is built rather than at boot, so an instance without + it still offers every other type, and a node placed without it says so + on the canvas straight away. + """ + try: + import wren + except ImportError: + raise RuntimeError("the wren node needs the model: pip install wren") from None + return wren + + +#: The last word of a target port, and the kind of device it stands for. +KIND_WORDS = { + "light": "light", + "lamp": "light", + "brightness": "light", + "setpoint": "heat", + "heating": "heat", + "heat": "heat", + "boost": "heat", + "media": "media", + "music": "media", + "tv": "media", + "shutter": "shutter", + "blind": "shutter", + "cover": "shutter", + "water": "water", + "watering": "water", + "sprinkler": "water", +} + +#: What the node reads, and the types each may be. Only the prompt wakes it. +INPUTS = { + "prompt": {DType.STR}, + "presence": {DType.STR}, + "outdoor": {DType.FLOAT, DType.INT}, + "dark": {DType.BOOL}, +} + +#: The output the calls are reported on. Every other output is a device. +ACTIONS = "actions" + +_TARGET_TYPES = {DType.BOOL, DType.INT, DType.FLOAT, DType.STR} + +#: The tools that act on each kind. +_TOOLS = { + "light": "lights", + "media": "media", + "heat": "heating", + "shutter": "shutters", + "water": "watering", +} + +#: What a call sets a target to, by the target's type: (bool, number, text). +#: None is a call that type has no value for, which is then not applied. +#: `heating.set` carries its own number. +_VALUES: dict[str, tuple[bool | None, float | None, str | None]] = { + "lights.on": (True, 100, "on"), + "lights.off": (False, 0, "off"), + "media.play": (True, None, "play"), + "media.stop": (False, None, "stop"), + "heating.boost": (True, None, "boost"), + "heating.off": (False, None, "off"), + "shutters.up": (True, 100, "up"), + "shutters.down": (False, 0, "down"), + "shutters.stop": (None, None, "stop"), + "watering.start": (True, None, "on"), + "watering.stop": (False, None, "off"), +} + +#: What every area held in training: a light, and in the living room, which +#: always existed, media and heating besides. A house missing them reads as +#: unfamiliar and the confidence drops with it, so they are filled in; a call +#: on one that nothing is wired to is reported and not applied. +_TRAINED = {"living room": ("light", "media", "heat")} + + +def _device(port: str, areas: dict[str, str]) -> tuple[str, str] | None: + """The (area, kind) a port name stands for, or None.""" + for slug, area in areas.items(): + word = port.removeprefix(f"{slug}_") + if word != port and word in KIND_WORDS: + return area, KIND_WORDS[word] + return None + + +def _on(value: Any) -> bool: + if isinstance(value, bool): + return value + if isinstance(value, (int, float)): + return value > 0 + return str(value).strip().lower() in {"on", "up", "open", "play", "true"} + + +def _state(kind: str, values: list[Any]) -> str: + """A device's state as the model reads it, from what its targets hold.""" + if kind == "heat": + temps = [ + v for v in values if isinstance(v, (int, float)) and not isinstance(v, bool) + ] + return f"{round(temps[0])}C" if temps else "20C" + on = bool(values) and _on(values[0]) + if kind == "shutter": + return "up" if on else "down" + return "on" if on else "off" + + +def _value(call: Any, dtype: DType) -> Any: + """What ``call`` sets a target of this type to, or None.""" + if call.tool == "heating.set": + temp = call.args.get("temp") + flag, number, text = ( + None, + temp if isinstance(temp, (int, float)) else None, + None, + ) + else: + flag, number, text = _VALUES.get(call.tool, (None, None, None)) + if dtype is DType.BOOL: + return flag + if dtype is DType.STR: + return text + if number is None: + return None + return int(number) if dtype is DType.INT else float(number) + + +def _text(call: Any) -> str: + """The call as the chat shows it, less the device it names.""" + args = ", ".join(f"{k}={v}" for k, v in call.args.items() if k != "id") + return f"{call.tool}({args})" + + +class WrenNode(Node): + """Turn the command on ``prompt`` into values for the devices wired to it. + + Inputs: ``prompt`` (the command, and the only one that wakes the node), + ``presence`` (the area the user is in), ``outdoor`` (°C) and ``dark`` + (otherwise read off the clock: before 07:00 or from 18:00). + + Outputs: ``actions``, a list of records, one per call and target, and + any number of targets named ``_``. Targets of one area and kind + are one device to the model; each takes what the call means for its type. + A delayed call is held by the engine's work queue, like the Trigger + node's second value. + """ + + class Params(BaseModel): + model_config = ConfigDict(extra="allow") + + apply: bool = Field( + default=False, + description="Write to the targets. Off, the node only reports what it would do.", + ) + min_confidence: float = Field( + default=0.93, + ge=0, + le=1, + description="Below this, nothing is written; the calls are still reported.", + ) + + __slots__ = ("cfg", "devices") + + # A call is a command to a device; doing it twice is not doing it once. + idempotent = False + + def __init__( + self, + requires: MessageSpec | Iterable[MessageSpec] = (), + provides: MessageSpec | Iterable[MessageSpec] = (), + params: dict[str, Any] | None = None, + name: str | None = None, + ): + self.cfg = self.Params.model_validate(params or {}) + wren = _wren() + inputs = self._normalize_ports(requires) + outputs = self._normalize_ports(provides) + + for spec in inputs: + allowed = INPUTS.get(spec.port) + if allowed is None: + raise ValueError( + f"'{spec.port}' is not an input of wren. It reads " + f"{', '.join(INPUTS)}." + ) + if spec.dtype not in allowed: + raise ValueError( + f"'{spec.port}' is {spec.dtype.value}; wren reads it as " + f"{' or '.join(sorted(d.value for d in allowed))}." + ) + if not any(spec.port == "prompt" for spec in inputs): + raise ValueError("wren needs a 'prompt' input: the command to act on.") + + areas = {area.replace(" ", "_"): area for area in wren.AREAS} + #: (area, kind) to the ports standing for that device. + self.devices: dict[tuple[str, str], list[str]] = {} + for spec in outputs: + if spec.port == ACTIONS: + if spec.dtype is not DType.LIST or spec.item_dtype is not DType.RECORD: + raise ValueError(f"'{ACTIONS}' is a list of records.") + continue + device = _device(spec.port, areas) + if device is None: + raise ValueError( + f"'{spec.port}' names no device wren knows. A target is " + f"_, the area one of {', '.join(areas)}, the kind " + f"one of {', '.join(KIND_WORDS)}." + ) + if spec.dtype not in _TARGET_TYPES: + raise ValueError( + f"'{spec.port}' is {spec.dtype.value}; a target is bool, int, " + "float or str." + ) + self.devices.setdefault(device, []).append(spec.port) + + super().__init__( + self._act, + # The prompt is the event; the rest is read when it arrives. + requires=[ + spec.model_copy(update={"trigger": spec.port == "prompt"}) + for spec in inputs + ], + provides=outputs, + params=params, + name=name or "wren", + ) + + def _house( + self, + wren: Any, + specs: dict[str, MessageSpec], + presence: str | None, + outdoor: float | None, + dark: bool | None, + ) -> Any: + """The house as the model reads it, each device in its current state.""" + names = [specs[port].name for ports in self.devices.values() for port in ports] + current = ( + self._pipeline.state.get_present(names) + if self._pipeline is not None + else {} + ) + areas: dict[str, list[Any]] = {} + for area in wren.AREAS: + wired = {kind for (a, kind) in self.devices if a == area} + if not wired and area not in _TRAINED: + continue + kinds = wired | set(_TRAINED.get(area, ("light",))) + areas[area] = [] + for kind in (k for k in wren.KINDS if k in kinds): + ports = self.devices.get((area, kind), []) + held = [ + current[specs[p].name] for p in ports if specs[p].name in current + ] + # The id is ours to choose: the model is shown aliases anyway. + areas[area].append( + wren.Device(kind, f"{area}:{kind}", _state(kind, held)) + ) + clock = time.localtime() + where = str(presence or "").strip().lower().replace("_", " ") + return wren.House( + time=time.strftime("%H:%M", clock), + # The range the model was trained on; past it a number is unfamiliar. + outdoor_c=min(max(float(20 if outdoor is None else outdoor), -5), 30), + dark=clock.tm_hour < 7 or clock.tm_hour >= 18 if dark is None else dark, + user_area=where if where in wren.AREAS else None, + areas=areas, + ) + + def _act( + self, + params: dict[str, Any], + prompt: str = "", + presence: str | None = None, + outdoor: float | None = None, + dark: bool | None = None, + ) -> dict[str, Any] | None: + if not prompt.strip(): + # An empty command is nothing to do, which is what a flow input + # starts as. + return None + wren = _wren() + specs = {spec.port: spec for spec in self.output_ports} + house = self._house(wren, specs, presence, outdoor, dark) + calls, confidence = wren.act(prompt, house) + confidence = round(float(confidence), 3) + write = self.cfg.apply and confidence >= self.cfg.min_confidence + writes: dict[str, Any] = {} + actions: list[dict[str, Any]] = [] + for call in calls: + area, _, kind = str(call.args.get("id", "")).partition(":") + ports = self.devices.get((area, kind), []) + if call.tool.split(".")[0] != _TOOLS.get(kind): + ports = [] + hits = [ + (specs[port], value) + for port in ports + if (value := _value(call, specs[port].dtype)) is not None + ] + record = { + "message": "", + "value": None, + "applied": False, + "delay_s": call.delay_s, + "call": _text(call), + "confidence": confidence, + } + if not hits: + actions.append(record) + for spec, value in hits: + applied = write + if applied and call.delay_s: + applied = self._pipeline is not None and self._pipeline.defer( + self, self._to_messages({spec.port: value}) or {}, call.delay_s + ) + elif applied: + writes[spec.port] = value + actions.append( + record | {"message": spec.name, "value": value, "applied": applied} + ) + + if any(spec.port == ACTIONS for spec in self.output_ports): + writes[ACTIONS] = actions + return writes or None diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py index 868599f..9faf9cc 100644 --- a/backend/fluksio/flow/pipeline.py +++ b/backend/fluksio/flow/pipeline.py @@ -85,7 +85,7 @@ class ValueSource(BaseModel): flow, an agent — be shown at all, since none of those is a node here. """ - #: node, dashboard, flow, agent or api. + #: node, dashboard, flow, agent, chat or api. kind: str = "node" #: Node id, dashboard name, or whatever identifies the caller. id: str = "" diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 0cd7afa..f7a19a3 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -157,6 +157,12 @@ ignore_missing_imports = true module = ["http_ece"] ignore_missing_imports = true +# The model behind the `wren` node: optional and untyped, imported inside the +# guard that says how to install it. +[[tool.mypy.overrides]] +module = ["wren"] +ignore_missing_imports = true + [tool.ruff] target-version = "py312" diff --git a/backend/tests/flow/conftest.py b/backend/tests/flow/conftest.py index da2907c..8c6859b 100644 --- a/backend/tests/flow/conftest.py +++ b/backend/tests/flow/conftest.py @@ -1,4 +1,6 @@ from collections.abc import Generator +from dataclasses import dataclass +from typing import Any import pytest @@ -7,3 +9,47 @@ import pytest def db() -> Generator[None, None, None]: """The engine holds no database state, so these tests need no database.""" yield + + +@dataclass +class _Device: + kind: str + id: str + state: str + + +@dataclass +class _House: + time: str + outdoor_c: float + dark: bool + user_area: str | None + areas: dict[str, list[_Device]] + + +class FakeWren: + """The model package's surface, answering whatever a test sets ``answer`` to.""" + + AREAS = [ + *("living room", "kitchen", "bathroom", "bedroom", "office"), + *("garden", "patio", "garage", "driveway", "balcony"), + ] + KINDS = ["light", "media", "heat", "shutter", "water"] + Device = _Device + House = _House + + def __init__(self) -> None: + self.answer: tuple[list[Any], float] = ([], 1.0) + self.houses: list[_House] = [] + + def act(self, command: str, house: _House) -> tuple[list[Any], float]: + self.houses.append(house) + return self.answer + + +@pytest.fixture +def fake_wren(monkeypatch: pytest.MonkeyPatch) -> FakeWren: + """Stands in for the optional model package the wren node imports.""" + fake = FakeWren() + monkeypatch.setattr("fluksio.flow.nodes.wren._wren", lambda: fake) + return fake diff --git a/backend/tests/flow/test_node_types.py b/backend/tests/flow/test_node_types.py index 056593c..27b7421 100644 --- a/backend/tests/flow/test_node_types.py +++ b/backend/tests/flow/test_node_types.py @@ -90,11 +90,16 @@ FIXTURES: dict[str, dict] = { "requires": [MessageSpec(name="text", dtype=DType.STR)], "provides": [], }, + "wren": { + "params": {"apply": True}, + "requires": [MessageSpec(name="prompt", dtype=DType.STR)], + "provides": [MessageSpec(name="kitchen_light", dtype=DType.BOOL)], + }, } @pytest.mark.parametrize("type_name", sorted(FIXTURES)) -def test_node_type_builds_and_qualifies(type_name: str): +def test_node_type_builds_and_qualifies(type_name: str, fake_wren): fixture = FIXTURES[type_name] node = NODE_TYPES[type_name].cls( requires=fixture["requires"], diff --git a/backend/tests/flow/test_wren.py b/backend/tests/flow/test_wren.py new file mode 100644 index 0000000..2f3fff3 --- /dev/null +++ b/backend/tests/flow/test_wren.py @@ -0,0 +1,145 @@ +"""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.wren import WrenNode, _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), + ) + assert node.devices == { + ("living room", "heat"): ["living_room_setpoint", "living_room_boost"], + ("kitchen", "light"): ["kitchen_lamp"], + } + + +@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, + } + + +@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), + ("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), + ] + 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, + } + ] + }