nodes: what porting the house needed from the vocabulary

Four small things, each with a device behind it.

An MQTT filter now routes what it subscribed to. `+` and `#` reached the
broker and were then looked up in an exact-match dict, so every message a
wildcard subscription received was dropped in silence.

`json_key` lifts a value out of the object a device wraps it in — Victron
publishes `{"value": 47}` on every path, which was otherwise a Python node
per port.

The trigger node learned `passthrough` and `wait_port`, because how long to
wait can be a value rather than a constant: a rollershutter takes 26 seconds
up and 28 down. A wait of zero sends nothing afterwards and still cancels
what the last message scheduled, which is how a stop is commanded once
instead of forever.

The HTTP sender takes fixed `query` parameters, so an API key is a secret
reference rather than a message on the canvas, and `send_inputs` off for a
request whose inputs are only a trigger.

Also: `delay` accepts fractional seconds, and `TZ` reaches the container, so
a cron expression means local time. Left unset it is UTC, as before.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

.env.example

@@ -57,3 +57,6 @@ DOCKER_IMAGE_FRONTEND=fluksio-frontend
 
 # The MCP endpoint agents connect to, and the OAuth server behind it.
 MCP_ENABLED=true
+
+# Local time for schedules; set from the root .env by scripts/setup.sh.
+TZ=UTC

backend/fluksio/flow/nodes/delay.py

@@ -100,8 +100,9 @@ class DelayNode(Node):
     class Params(BaseModel):
         model_config = ConfigDict(extra="allow")
 
-        delay: int = 0
-        interval: int = 0
+        # Seconds, fractional: a shutter's run-time is not a whole number.
+        delay: float = 0
+        interval: float = 0
         # Input port → output port; ports are paired in order when empty.
         mapping: dict[str, str] = {}
         cron: str | None = None

backend/fluksio/flow/nodes/http.py

@@ -130,6 +130,8 @@ class HttpNode(Node):
         "mode",
         "timeout",
         "headers",
+        "query",
+        "send_inputs",
         "secret",
         "_route_registered",
     )
@@ -141,6 +143,21 @@ class HttpNode(Node):
         method: Literal["GET", "POST"] = "POST"
         timeout: float = 30.0
         headers: dict[str, str] = {}
+        query: dict[str, Any] = Field(
+            default_factory=dict,
+            description=(
+                "Fixed query parameters, sender mode. A value may be a secret "
+                "reference, which is how an API key stays out of the flow file."
+            ),
+        )
+        send_inputs: bool = Field(
+            default=True,
+            description=(
+                "Send the node's inputs as the request body or query. Off when "
+                "the inputs are only a trigger and the request is fully "
+                "described by 'url' and 'query'."
+            ),
+        )
         secret: str = Field(
             default="",
             description=(
@@ -197,6 +214,8 @@ class HttpNode(Node):
         self.method = cfg.method.upper()
         self.timeout = cfg.timeout
         self.headers = cfg.headers
+        self.query = cfg.query
+        self.send_inputs = cfg.send_inputs
         self.secret = cfg.secret
         self._route_registered = False
 
@@ -251,18 +270,20 @@ class HttpNode(Node):
         :rtype: dict | None
         """
         client = shared_client()
+        payload = dict(kwargs) if self.send_inputs else {}
         try:
             if self.method == "GET":
                 response = client.get(
                     self.url,
-                    params=kwargs,
+                    params={**self.query, **payload},
                     headers=self.headers,
                     timeout=self.timeout,
                 )
             else:  # POST
                 response = client.post(
                     self.url,
-                    json=kwargs,
+                    params=self.query or None,
+                    json=payload,
                     headers=self.headers,
                     timeout=self.timeout,
                 )

backend/fluksio/flow/nodes/mqtt.py

@@ -24,6 +24,30 @@ logger = logging.getLogger(__name__)
 PUBLISH_QUEUE_SIZE = 256
 
 
+def topic_matches(filter_: str, topic: str) -> bool:
+    """Does an MQTT topic filter cover this topic?
+
+    The broker already decided to send it, so this is only about which port it
+    belongs to. ``+`` stands for one level, ``#`` for the rest of them; a
+    filter with neither is an exact string, which is the common case and the
+    one the caller checks first.
+    """
+    if filter_ == topic:
+        return True
+    parts = filter_.split("/")
+    levels = topic.split("/")
+    for i, part in enumerate(parts):
+        if part == "#":
+            # Everything before it already matched, and '#' takes the rest —
+            # including nothing at all, so 'a/#' covers 'a' as the spec says.
+            return True
+        if i >= len(levels):
+            return False
+        if part != "+" and part != levels[i]:
+            return False
+    return len(parts) == len(levels)
+
+
 class MqttNode(Node):
     """
     MQTT node that can act as a subscriber (trigger) or publisher (sender).
@@ -127,7 +151,9 @@ class MqttNode(Node):
         "qos",
         "retain",
         "keepalive",
+        "json_keys",
         "_topic_to_ports",
+        "_wildcards",
         "_subscription_task",
         "_mqtt_client",
         "_stop_event",
@@ -149,6 +175,10 @@ class MqttNode(Node):
         qos: int = 0
         retain: bool = False
         keepalive: int = 60
+        # Which key to lift out of a JSON object payload. A device that wraps
+        # its reading — Victron's ``{"value": 5}`` — is otherwise a Python node
+        # per port. One key for every port, or a per-port mapping.
+        json_key: str | dict[str, str] = ""
 
     @classmethod
     def instance_key(cls, params: dict[str, Any]) -> str | None:
@@ -194,10 +224,20 @@ class MqttNode(Node):
         else:
             self.topics = {spec.port: cfg.topic for spec in ports}
 
-        # Reverse lookup for routing incoming payloads back to ports.
+        # Reverse lookup for routing incoming payloads back to ports. A filter
+        # holding a wildcard cannot be found by the incoming topic, so those
+        # are kept aside and walked when the exact lookup misses.
         self._topic_to_ports: dict[str, list[str]] = {}
         for port, topic in self.topics.items():
             self._topic_to_ports.setdefault(topic, []).append(port)
+        self._wildcards = [t for t in self._topic_to_ports if "+" in t or "#" in t]
+
+        if isinstance(cfg.json_key, dict):
+            self.json_keys: dict[str, str] = dict(cfg.json_key)
+        else:
+            self.json_keys = (
+                {spec.port: cfg.json_key for spec in ports} if cfg.json_key else {}
+            )
 
         self.broker_host = cfg.broker_host
         self.broker_port = cfg.broker_port
@@ -488,6 +528,17 @@ class MqttNode(Node):
             self.name,
         )
 
+    def _ports_for(self, topic: str) -> list[str]:
+        """Which ports an arriving topic feeds — exact mapping, then filters."""
+        ports = self._topic_to_ports.get(topic)
+        if ports is not None:
+            return ports
+        matched: list[str] = []
+        for pattern in self._wildcards:
+            if topic_matches(pattern, topic):
+                matched.extend(self._topic_to_ports[pattern])
+        return matched
+
     async def _subscription_loop(self) -> None:
         """
         Listen for MQTT messages and trigger the pipeline.
@@ -536,7 +587,7 @@ class MqttNode(Node):
                             )
 
                             # Find which port(s) this topic feeds
-                            ports = self._topic_to_ports.get(incoming_topic, [])
+                            ports = self._ports_for(incoming_topic)
                             if not ports:
                                 logger.debug(
                                     "[%s] No mapping for topic '%s', ignoring",
@@ -558,10 +609,12 @@ class MqttNode(Node):
                                 if spec is None:
                                     continue
 
-                                # A JSON object may carry the port as a key;
-                                # anything else is the value itself.
-                                if isinstance(parsed, dict) and port in parsed:
-                                    value = parsed[port]
+                                # A JSON object may carry the value under a
+                                # named key — the port's own name, or whatever
+                                # ``json_key`` says the device wraps it in.
+                                key = self.json_keys.get(port, port)
+                                if isinstance(parsed, dict) and key in parsed:
+                                    value = parsed[key]
                                 else:
                                     value = parsed
 

backend/fluksio/flow/nodes/trigger.py

@@ -23,7 +23,10 @@ class TriggerNode(Node):
     """Emit ``first`` on arrival, then ``then`` once the wait runs out.
 
     A value arriving during the wait extends it, so the second emission only
-    happens when things have actually gone quiet.
+    happens when things have actually gone quiet. ``passthrough`` sends the
+    incoming value instead of ``first``, and ``wait_port`` names an input
+    carrying the wait, for the case where how long to wait is itself a value —
+    a shutter that takes 26 s up and 28 s down.
     """
 
     class Params(BaseModel):
@@ -41,6 +44,17 @@ class TriggerNode(Node):
             default=True,
             description="A value arriving during the wait starts it over.",
         )
+        passthrough: bool = Field(
+            default=False,
+            description="Send the incoming value instead of 'first'.",
+        )
+        wait_port: str = Field(
+            default="",
+            description=(
+                "An input port carrying the wait in seconds, for a wait that "
+                "differs per message. Zero or less sends nothing afterwards."
+            ),
+        )
 
     __slots__ = ("cfg",)
 
@@ -67,23 +81,30 @@ class TriggerNode(Node):
         if not kwargs or not self.output_ports:
             return None
 
+        wait = self.cfg.wait
+        if self.cfg.wait_port:
+            # A value port rather than a setting: a shutter's run-time differs
+            # by direction, and that is a message the graph carries.
+            wait = float(kwargs.get(self.cfg.wait_port, wait) or 0)
+
         armed = self.recall("armed", 0)
         if armed and not self.cfg.extend:
             # Already counting down and not extending: ignore the new value.
             return None
 
         # A generation counter, so a value arriving mid-wait invalidates the
-        # deferred emission that was already scheduled.
+        # deferred emission that was already scheduled. Bumped even when this
+        # message asks for no second value, which is how a wait is cancelled.
         generation = int(self.recall("generation", 0)) + 1
         self.remember("generation", generation)
-        self.remember("armed", 1)
+        self.remember("armed", 1 if wait > 0 else 0)
 
-        if self.cfg.then is not None and self._pipeline is not None:
+        if self.cfg.then is not None and self._pipeline is not None and wait > 0:
             deferred = self._outputs(self.cfg.then)
             scheduled = self._pipeline.defer(
                 self,
                 self._to_messages(deferred) or {},
-                self.cfg.wait,
+                wait,
                 guard=("generation", generation),
             )
             if not scheduled:
@@ -94,4 +115,14 @@ class TriggerNode(Node):
                 )
                 self.remember("armed", 0)
 
+        if self.cfg.passthrough:
+            # Whatever woke the node, minus the port that only said how long
+            # to wait. Several inputs are ambiguous, so the first one wins.
+            values = {
+                port: value
+                for port, value in kwargs.items()
+                if port != self.cfg.wait_port
+            }
+            if values:
+                return self._outputs(next(iter(values.values())))
         return self._outputs(self.cfg.first)

backend/tests/flow/test_house_vocabulary.py

@@ -0,0 +1,132 @@
+"""What the house port needed from the built-in nodes.
+
+Three additions, each with a device behind it: an MQTT filter that actually
+routes what it subscribed to, a value lifted out of the object a device wraps
+it in, and a wait that differs per message because a shutter takes longer to
+come up than to go down.
+"""
+
+from fluksio.flow.messages import DType, MessageSpec
+from fluksio.flow.nodes import MqttNode
+from fluksio.flow.nodes.mqtt import topic_matches
+from fluksio.flow.nodes.trigger import TriggerNode
+
+# ── MQTT topic filters ────────────────────────────────────────────────────
+
+
+def test_a_filter_covers_what_the_broker_would_send_it():
+    assert topic_matches("sensors/#", "sensors/living/temp")
+    assert topic_matches("sensors/#", "sensors")
+    assert topic_matches("sensors/+/temp", "sensors/living/temp")
+    assert topic_matches("#", "anything/at/all")
+    assert topic_matches("shelly/status", "shelly/status")
+
+
+def test_a_filter_does_not_cover_a_neighbouring_topic():
+    assert not topic_matches("sensors/+/temp", "sensors/living/kitchen/temp")
+    assert not topic_matches("sensors/+/temp", "sensors/living/hum")
+    assert not topic_matches("sensors/#", "actors/living/temp")
+    assert not topic_matches("sensors/living/temp", "sensors/living")
+
+
+def _subscriber(topic, **params) -> MqttNode:
+    node = MqttNode(
+        provides=[MessageSpec(name="reading", port="reading", dtype=DType.FLOAT)],
+        params={"topic": {"reading": topic}, **params},
+    )
+    node.assign_flow("house", "in")
+    return node
+
+
+def test_a_wildcard_subscription_routes_to_its_port():
+    """It used to subscribe and then drop every message it was sent."""
+    node = _subscriber("shellypm/status/#")
+    assert node._ports_for("shellypm/status/pm1:0") == ["reading"]
+    assert node._ports_for("shellypmc/status/pm1:0") == []
+
+
+def test_an_exact_topic_still_wins_over_a_filter():
+    node = MqttNode(
+        provides=[
+            MessageSpec(name="soc", port="soc", dtype=DType.FLOAT),
+            MessageSpec(name="rest", port="rest", dtype=DType.JSON),
+        ],
+        params={"topic": {"soc": "N/x/battery/Soc", "rest": "N/x/#"}},
+    )
+    node.assign_flow("power", "in")
+    assert node._ports_for("N/x/battery/Soc") == ["soc"]
+    assert node._ports_for("N/x/vebus/P") == ["rest"]
+
+
+# ── json_key ──────────────────────────────────────────────────────────────
+
+
+def test_json_key_lifts_the_value_a_device_wraps():
+    """Victron publishes {"value": 47} on every one of its topics."""
+    node = _subscriber("N/x/battery/Soc", json_key="value")
+    assert node.json_keys == {"reading": "value"}
+
+
+def test_without_json_key_the_port_name_is_the_key_as_before():
+    node = _subscriber("sensors/temp")
+    assert node.json_keys == {}
+
+
+# ── trigger: passthrough and a per-message wait ───────────────────────────
+
+
+class _Pipeline:
+    """Just enough pipeline to record what was deferred and for how long."""
+
+    def __init__(self):
+        self.deferred = []
+        self.state = {}
+
+    def defer(self, node, outputs, seconds, guard=None, kind="cascade"):
+        self.deferred.append((outputs, seconds, guard))
+        return True
+
+
+def _trigger(**params) -> tuple[TriggerNode, _Pipeline]:
+    node = TriggerNode(
+        requires=[
+            MessageSpec(name="run", port="run", dtype=DType.STR),
+            MessageSpec(name="run_for", port="run_for", dtype=DType.FLOAT),
+        ],
+        provides=[MessageSpec(name="cmd", port="cmd", dtype=DType.STR)],
+        params={"then": "STOP", "passthrough": True, "wait_port": "run_for", **params},
+    )
+    node.assign_flow("shutters", "door")
+    pipeline = _Pipeline()
+    node._pipeline = pipeline
+    return node, pipeline
+
+
+def test_the_incoming_value_passes_through_and_the_wait_comes_from_a_port():
+    node, pipeline = _trigger()
+
+    assert node._trigger({}, run="DOWN", run_for=28.0) == {"cmd": "DOWN"}
+    (outputs, seconds, _guard) = pipeline.deferred[-1]
+    assert outputs == {"shutters.cmd": "STOP"} and seconds == 28.0
+
+    node._trigger({}, run="UP", run_for=26.0)
+    assert pipeline.deferred[-1][1] == 26.0
+
+
+def test_a_second_command_invalidates_the_stop_the_first_one_scheduled():
+    node, pipeline = _trigger()
+    node._trigger({}, run="DOWN", run_for=28.0)
+    first = pipeline.deferred[-1][2]
+    node._trigger({}, run="UP", run_for=26.0)
+    assert pipeline.deferred[-1][2] != first
+
+
+def test_a_wait_of_zero_sends_nothing_afterwards():
+    """STOP is commanded once. Nothing follows it, forever."""
+    node, pipeline = _trigger()
+    node._trigger({}, run="DOWN", run_for=28.0)
+    scheduled = len(pipeline.deferred)
+
+    assert node._trigger({}, run="STOP", run_for=0.0) == {"cmd": "STOP"}
+    assert len(pipeline.deferred) == scheduled
+    assert node.recall("armed", 0) == 0

docker/compose.yml

@@ -122,6 +122,10 @@ services:
       # be lost on the next rebuild — un-pairing every screen.
       - ALERTS_FILE=/data/alerts.json
       - PANELS_FILE=/data/panels.json
+      # Schedules are written in local time: "off at 02:00" means the house's
+      # two in the morning, not the container's. Unset, an image is UTC, and a
+      # cron would be right twice a year.
+      - TZ=${TZ:-UTC}
 
     volumes:
       - app-flow-data:/data

docs/reference/configuration.md

@@ -88,6 +88,13 @@ remember it is fixed at build time: changing it means rebuilding that image.
 |---|---|---|
 | `ENVIRONMENT` | `local` | `local`, `staging` or `production` |
 | `PRIVATE_API_ENABLED` | `false` | unauthenticated test-only endpoints; needs `ENVIRONMENT=local` too |
+| `TZ` | `UTC` | the timezone every schedule is written in |
+
+`TZ` is the container's own, not a setting the code reads: an `inject` or a
+`delay` with a cron expression fires on local time. Left at `UTC`, "off at
+02:00" means two in the morning UTC, which in most of the world is neither two
+o'clock nor the same hour in summer as in winter. Set it to where the
+installation is.
 
 `production` closes `/docs`, `/redoc` and the OpenAPI document, because the
 schema enumerates every endpoint the installation serves — including the paths

docs/reference/node-types.md

@@ -40,8 +40,15 @@ A node with *outputs only* subscribes; a node with *inputs* publishes.
 | `qos` | `0` | 0, 1 or 2 |
 | `retain` | `false` | on published messages |
 | `keepalive` | `60` | seconds |
+| `json_key` | — | key to lift out of an object payload; one for every port, or `{"port": "key"}` |
 
-Nodes sharing a broker share one connection.
+A topic may be a filter: `+` matches one level, `#` the rest. Everything a
+filter matches lands on the same port, so use one port per topic where the
+difference matters.
+
+`json_key` is for a device that wraps its reading — Victron publishes
+`{"value": 47}` on every path. Without it, a payload object is unwrapped only
+when it happens to carry the port's own name as a key.
 
 ### HTTP
 
@@ -56,6 +63,8 @@ Outputs only makes it a **webhook**: the engine mounts a route at
 | `method` | `POST` | `GET` or `POST` |
 | `timeout` | `30` | seconds, sender mode |
 | `headers` | `{}` | |
+| `query` | `{}` | fixed query parameters; a value may be a secret reference |
+| `send_inputs` | `true` | off when the inputs only trigger the request |
 | `secret` | — | shared secret appended to the webhook URL; takes a secret reference |
 
 !!! warning "A webhook with no secret is open to anyone who can reach the host."
@@ -137,7 +146,7 @@ presses.
 
 | Setting | Default | Notes |
 |---|---|---|
-| `delay` | `0` | seconds to hold each message |
+| `delay` | `0` | seconds to hold each message, fractional |
 | `interval` | `0` | minimum seconds between forwards |
 | `mapping` | `{}` | input port → output port; paired in order when empty |
 | `cron` | — | five-field expression |
@@ -162,10 +171,18 @@ received.
 | `then` | `false` | sent when the wait expires; empty sends nothing |
 | `wait` | `60` | seconds of quiet before the second value |
 | `extend` | `true` | a value arriving during the wait starts it over |
+| `passthrough` | `false` | send the incoming value instead of `first` |
+| `wait_port` | — | an input carrying the wait, when it differs per message |
 
 The shape this exists for: *the door opened — turn the light on, and off again
 in two minutes unless it opens again.*
 
+`wait_port` covers the case where how long to wait is itself a value: a
+rollershutter takes 26 seconds up and 28 down, so the node that decides the
+direction says how long to run for as well. A wait of zero or less sends
+nothing afterwards — and still cancels whatever the last message scheduled,
+which is how a *stop* is commanded exactly once.
+
 ## Logic
 
 ### Switch