wren node writes the house brightness from lights.set/dim/brighten (wren s11)

backend/fluksio/flow/nodes/wren.py

@@ -79,6 +79,9 @@ REPLY = "reply"
 UNHANDLED = "unhandled"
 #: The reply when nothing was done and nothing reads ``unhandled``.
 CANT = "Sorry, I can't do that yet."
+#: The output for the house's brightness, 0 to 100, which the model's
+#: brightness calls write whatever area they name.
+BRIGHTNESS = "brightness"
 
 _TARGET_TYPES = {DType.BOOL, DType.INT, DType.FLOAT, DType.STR}
 
@@ -111,6 +114,12 @@ _VALUES: dict[str, tuple[bool | None, float | None, str | None]] = {
     "switches.off": (False, 0, "off"),
 }
 
+#: The brightness calls, and what "dim" and "brighter" do: a step from the
+#: brightness now, or from a usual one when none is known. Dimmed lights
+#: stay on.
+_LEVELS = ("lights.set", "lights.dim", "lights.brighten")
+_STEP, _DIMMEST, _BRIGHTEST, _USUAL = 30, 10, 100, 70
+
 #: An applied call, said: "kitchen light on". `heating.set` says its number.
 _SAID = {
     "lights.on": "light on",
@@ -264,6 +273,13 @@ def _value(call: Any, dtype: DType) -> Any:
             temp if isinstance(temp, (int, float)) else None,
             None,
         )
+    elif call.tool == "lights.set":
+        level = call.args.get("level")
+        flag, number, text = (
+            None,
+            level if isinstance(level, (int, float)) else None,
+            None,
+        )
     else:
         flag, number, text = _VALUES.get(call.tool, (None, None, None))
     if dtype is DType.BOOL:
@@ -279,12 +295,15 @@ def _said(call: Any, area: str) -> str:
     """An applied call as the reply says it: "garage light on in 5 minutes"."""
     if call.tool == "heating.set":
         said = f"heating to {_value(call, DType.FLOAT):g} degrees"
+    elif call.tool == "lights.set":
+        said = f"brightness at {_value(call, DType.FLOAT):g} percent"
     else:
         said = _SAID.get(call.tool, call.tool)
     if call.delay_s:
         minutes = round(call.delay_s / 60)
         said += f" in {minutes} minute{'s' * (minutes != 1)}"
-    return f"{area} {said}"
+    # The brightness is the house's, not the area's the call names.
+    return said if call.tool == "lights.set" else f"{area} {said}"
 
 
 def _plain(command: str) -> str:
@@ -364,7 +383,8 @@ class WrenNode(Node):
     Outputs: ``actions``, a list of records, one per call and target,
     ``reply``, a sentence saying what was applied ("Kitchen light on."), or
     that nothing was when no node reads ``unhandled``, ``unhandled``, the
-    prompt when nothing was, and any number of targets
+    prompt when nothing was, ``brightness``, the house's, for "dim the
+    lights" or "set the brightness to 30%", and any number of targets
     named ``<area>_<kind>``. 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
@@ -439,6 +459,10 @@ class WrenNode(Node):
                 if spec.dtype is not DType.STR:
                     raise ValueError(f"'{spec.port}' is a str.")
                 continue
+            if spec.port == BRIGHTNESS:
+                if spec.dtype not in (DType.FLOAT, DType.INT):
+                    raise ValueError(f"'{BRIGHTNESS}' is a float or an int.")
+                continue
             device = _device(spec.port, areas)
             if device is None:
                 raise ValueError(
@@ -556,6 +580,11 @@ class WrenNode(Node):
             ports = self.devices.get((area, kind), [])
             if call.tool.split(".")[0] != _TOOLS.get(kind) or area not in asked:
                 ports = []
+            if call.tool in _LEVELS:
+                ports = []
+                if BRIGHTNESS in specs:
+                    ports = [BRIGHTNESS]
+                    call = self._leveled(call, specs[BRIGHTNESS].name)
             hits = [
                 (specs[port], value)
                 for port in ports
@@ -593,7 +622,7 @@ class WrenNode(Node):
         if ACTIONS in wired:
             writes[ACTIONS] = actions
         if said:
-            reply = ", ".join(said)
+            reply = ", ".join(dict.fromkeys(said))
             answers.insert(0, reply[0].upper() + reply[1:] + ".")
         if answers and REPLY in wired:
             writes[REPLY] = " ".join(answers)
@@ -603,6 +632,25 @@ class WrenNode(Node):
             writes[REPLY] = CANT
         return writes or None
 
+    def _leveled(self, call: Any, name: str) -> _Call:
+        """A brightness call as the level it sets: "dim" is 30 below now."""
+        level = call.args.get("level")
+        if call.tool != "lights.set":
+            now = (
+                self._pipeline.state.get_present([name]).get(name)
+                if self._pipeline is not None
+                else None
+            )
+            if not isinstance(now, (int, float)) or isinstance(now, bool):
+                now = _USUAL
+            if call.tool == "lights.dim":
+                level = max(now - _STEP, min(now, _DIMMEST))
+            else:
+                level = min(now + _STEP, _BRIGHTEST)
+        elif isinstance(level, (int, float)):
+            level = min(max(level, 0), _BRIGHTEST)
+        return _Call("lights.set", {"level": level}, call.delay_s)
+
     def _group(self, command: str, areas: Iterable[str]) -> list[_Call] | None:
         """The calls a command for a whole group means, one per wired device,
         or None when it is not one: "turn off all lights" is every light.

backend/tests/flow/test_wren.py

@@ -307,3 +307,25 @@ def test_a_question_for_a_temperature_is_answered_from_its_reading(fake_wren):
         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}