diff --git a/.env.example b/.env.example index 4ea3d39..8585401 100644 --- a/.env.example +++ b/.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 diff --git a/backend/fluksio/flow/nodes/delay.py b/backend/fluksio/flow/nodes/delay.py index fc86531..a67dd27 100644 --- a/backend/fluksio/flow/nodes/delay.py +++ b/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 diff --git a/backend/fluksio/flow/nodes/http.py b/backend/fluksio/flow/nodes/http.py index 7d0cca8..80d6d3f 100644 --- a/backend/fluksio/flow/nodes/http.py +++ b/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, ) diff --git a/backend/fluksio/flow/nodes/mqtt.py b/backend/fluksio/flow/nodes/mqtt.py index bcd8dd5..f7d9ef2 100644 --- a/backend/fluksio/flow/nodes/mqtt.py +++ b/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 diff --git a/backend/fluksio/flow/nodes/trigger.py b/backend/fluksio/flow/nodes/trigger.py index 29e5aa8..d85d21e 100644 --- a/backend/fluksio/flow/nodes/trigger.py +++ b/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) diff --git a/backend/tests/flow/test_house_vocabulary.py b/backend/tests/flow/test_house_vocabulary.py new file mode 100644 index 0000000..10db4b7 --- /dev/null +++ b/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 diff --git a/docker/compose.yml b/docker/compose.yml index fb78247..5662981 100644 --- a/docker/compose.yml +++ b/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 diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index 1adb450..d37d8df 100644 --- a/docs/reference/configuration.md +++ b/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 diff --git a/docs/reference/node-types.md b/docs/reference/node-types.md index d0ce49c..ec6e3dc 100644 --- a/docs/reference/node-types.md +++ b/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