diff --git a/backend/app/flow/controller.py b/backend/app/flow/controller.py index b1e5220..5ed05bf 100644 --- a/backend/app/flow/controller.py +++ b/backend/app/flow/controller.py @@ -27,12 +27,21 @@ from app.flow.events import EventBus from app.flow.executor import ExecutionService from app.flow.messages import MessageSpec, qualify from app.flow.nodes import ( + ChangeNode, DelayNode, + ExecNode, + FileNode, HttpNode, InfluxDbNode, + InjectNode, + JoinNode, MLPNode, MqttNode, Node, + NtfyNode, + RbeNode, + SwitchNode, + TriggerNode, ) from app.flow.pipeline import Pipeline, ValidationIssue from app.flow.schemas import ( @@ -140,6 +149,60 @@ NODE_TYPES: dict[str, NodeType] = { cls=MLPNode, params_schema=_schema_of(MLPNode), ), + "inject": NodeType( + title="Inject", + description="Emit a value on request, on a timer, or when the flow starts.", + cls=InjectNode, + params_schema=_schema_of(InjectNode), + ), + "switch": NodeType( + title="Switch", + description="Send a value down one branch or another, by rule.", + cls=SwitchNode, + params_schema=_schema_of(SwitchNode), + ), + "change": NodeType( + title="Change", + description="Scale, offset, round or map a value on its way past.", + cls=ChangeNode, + params_schema=_schema_of(ChangeNode), + ), + "rbe": NodeType( + title="Filter unchanged", + description="Pass a value on only when it has actually changed.", + cls=RbeNode, + params_schema=_schema_of(RbeNode), + ), + "join": NodeType( + title="Join", + description="Gather several inputs into one object or list.", + cls=JoinNode, + params_schema=_schema_of(JoinNode), + ), + "trigger": NodeType( + title="Trigger", + description="Send one value now and another once things go quiet.", + cls=TriggerNode, + params_schema=_schema_of(TriggerNode), + ), + "exec": NodeType( + title="Command", + description="Run a command in the engine's container and read its output.", + cls=ExecNode, + params_schema=_schema_of(ExecNode), + ), + "file": NodeType( + title="File", + description="Read a file into the flow, or write one out of it.", + cls=FileNode, + params_schema=_schema_of(FileNode), + ), + "ntfy": NodeType( + title="Notification", + description="Push an incoming value to a phone through ntfy.", + cls=NtfyNode, + params_schema=_schema_of(NtfyNode), + ), } diff --git a/backend/app/flow/executor.py b/backend/app/flow/executor.py index ed509b6..df9a88a 100644 --- a/backend/app/flow/executor.py +++ b/backend/app/flow/executor.py @@ -239,6 +239,11 @@ class ExecutionService: self.queue.park(node.flow, item) return True + if item.guard_key and str(node.recall(item.guard_key, "")) != item.guard_value: + # The node moved on while this waited — a restarted timer, say. + logger.debug("Guard no longer holds for '%s', dropped", item.node) + return True + pipeline.apply_outputs(node, item.outputs or None) pipeline.run_downstream( node, entry_id=item.entry_id, replay=item.deliveries > 1 diff --git a/backend/app/flow/nodes/__init__.py b/backend/app/flow/nodes/__init__.py index cdcc9f6..0bde49f 100644 --- a/backend/app/flow/nodes/__init__.py +++ b/backend/app/flow/nodes/__init__.py @@ -6,16 +6,31 @@ Split by the outside world each one talks to. Importing from from app.flow.nodes.base import Node from app.flow.nodes.delay import DelayNode +from app.flow.nodes.exec import ExecNode +from app.flow.nodes.file import FileNode from app.flow.nodes.http import HttpNode from app.flow.nodes.influx import InfluxDbNode +from app.flow.nodes.inject import InjectNode +from app.flow.nodes.logic import ChangeNode, JoinNode, RbeNode, SwitchNode from app.flow.nodes.mlp import MLPNode from app.flow.nodes.mqtt import MqttNode +from app.flow.nodes.ntfy import NtfyNode +from app.flow.nodes.trigger import TriggerNode __all__ = [ + "ChangeNode", "DelayNode", + "ExecNode", + "FileNode", "HttpNode", "InfluxDbNode", + "InjectNode", + "JoinNode", "MLPNode", "MqttNode", "Node", + "NtfyNode", + "RbeNode", + "SwitchNode", + "TriggerNode", ] diff --git a/backend/app/flow/nodes/base.py b/backend/app/flow/nodes/base.py index 3be6515..438ee84 100644 --- a/backend/app/flow/nodes/base.py +++ b/backend/app/flow/nodes/base.py @@ -198,6 +198,37 @@ class Node: """ self._pipeline = pipeline + # ------------------------------------------------------------------------- + # Node-private state + # + # Code you write in a `python` node has no state handle: it is a pure + # function of its messages, and remembers things by reading a message it + # also writes. Node *types* the engine ships are a different matter — a + # rate limiter or a filter-on-change is about what happened last time, and + # that belongs to the engine rather than to any flow. It lives under a + # reserved key prefix, beside the timestamps and versions the pipeline + # already keeps there, so it survives a rebuild and never shows up as a + # message. + # ------------------------------------------------------------------------- + + def _state_key(self, key: str) -> str: + return f"__node__:{self.id}:{key}" + + def remember(self, key: str, value: Any) -> None: + """Keep a value for the next run of this node.""" + if self._pipeline is not None: + self._pipeline.state[self._state_key(key)] = value + + def recall(self, key: str, default: Any = None) -> Any: + """What this node last remembered under ``key``.""" + if self._pipeline is None: + return default + return self._pipeline.state.get(self._state_key(key), default) + + def forget(self, key: str) -> None: + if self._pipeline is not None: + self._pipeline.state.delete(self._state_key(key)) + def __repr__(self) -> str: return self.id diff --git a/backend/app/flow/nodes/exec.py b/backend/app/flow/nodes/exec.py new file mode 100644 index 0000000..4129bd1 --- /dev/null +++ b/backend/app/flow/nodes/exec.py @@ -0,0 +1,106 @@ +"""Exec: run a command and hand back what it said. + +The command runs inside the backend container, not on the host. That matters +when porting: a flow that read the host's journal or poked a host script needs +either a mount or a small listener on the host side, not this node. +""" + +from __future__ import annotations + +import logging +import shlex +import subprocess +from collections.abc import Iterable +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field + +from app.flow.messages import MessageSpec +from app.flow.nodes.base import Node + +logger = logging.getLogger(__name__) + + +class ExecNode(Node): + """Run a command, returning its output, error text and exit code.""" + + # Running a command twice is running it twice. + idempotent = False + + class Params(BaseModel): + model_config = ConfigDict(extra="allow") + + command: str = Field(description="The command to run.") + append_payload: bool = Field( + default=False, + description="Add the incoming value to the command as one argument.", + ) + timeout: float = Field( + default=30.0, gt=0, description="Give up after this many seconds." + ) + fail_on_error: bool = Field( + default=False, + description="Treat a non-zero exit as a node failure rather than output.", + ) + + __slots__ = ("cfg",) + + 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 {}) + super().__init__( + f=self._run, + requires=requires, + provides=provides, + params=params, + name=name or "exec", + ) + + def _run(self, params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None: + argv = shlex.split(self.cfg.command) + if not argv: + raise ValueError("exec node has no command") + if self.cfg.append_payload and kwargs: + argv.append(str(next(iter(kwargs.values())))) + + try: + # No shell: the command is a list, so a value carrying a semicolon + # is an argument rather than a second command. + completed = subprocess.run( + argv, + capture_output=True, + text=True, + timeout=self.cfg.timeout, + check=False, + ) + except subprocess.TimeoutExpired as exc: + raise TimeoutError( + f"'{argv[0]}' did not finish within {self.cfg.timeout}s" + ) from exc + except FileNotFoundError as exc: + raise FileNotFoundError( + f"'{argv[0]}' is not available in this container" + ) from exc + + if self.cfg.fail_on_error and completed.returncode != 0: + raise RuntimeError( + f"'{argv[0]}' exited {completed.returncode}: " + f"{completed.stderr.strip()[:200]}" + ) + + available = { + "stdout": completed.stdout, + "stderr": completed.stderr, + "code": completed.returncode, + } + # Ports named after one of those get it; anything else gets stdout, + # which is what a single-output exec node is almost always after. + return { + spec.port: available.get(spec.port, completed.stdout) + for spec in self.output_ports + } or None diff --git a/backend/app/flow/nodes/file.py b/backend/app/flow/nodes/file.py new file mode 100644 index 0000000..0b2dc9a --- /dev/null +++ b/backend/app/flow/nodes/file.py @@ -0,0 +1,98 @@ +"""File: read a file into the graph, or write one out of it. + +Confined to a directory the engine owns. A flow that could name any path +would be a way to read the secrets store or overwrite a node's source. +""" + +from __future__ import annotations + +import json +import logging +from collections.abc import Iterable +from pathlib import Path +from typing import Any, Literal + +from pydantic import BaseModel, ConfigDict, Field + +from app.flow.messages import MessageSpec +from app.flow.nodes.base import Node + +logger = logging.getLogger(__name__) + + +def files_root() -> Path: + """Where flow-readable files live: beside the flow store, not inside it.""" + from app.core.config import settings + + return settings.FLOWS_DIR.parent / "files" + + +def resolve(name: str) -> Path: + """Turn a flow-supplied name into a path inside the sandbox, or refuse.""" + root = files_root().resolve() + root.mkdir(parents=True, exist_ok=True) + target = (root / name).resolve() + if target != root and root not in target.parents: + raise ValueError(f"'{name}' is outside the files directory") + return target + + +class FileNode(Node): + """Read or write a file under the engine's files directory.""" + + class Params(BaseModel): + model_config = ConfigDict(extra="allow") + + path: str = Field(description="Path relative to the engine's files directory.") + mode: Literal["read", "write", "append"] = "read" + format: Literal["text", "json"] = Field( + default="text", description="Parse or serialize the contents as JSON." + ) + newline: bool = Field( + default=True, description="End each written record with a newline." + ) + + __slots__ = ("cfg",) + + 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 {}) + super().__init__( + f=self._act, + requires=requires, + provides=provides, + params=params, + name=name or f"file_{self.cfg.mode}", + ) + + @property + def idempotent(self) -> bool: # type: ignore[override] + # Reading twice is harmless; appending twice writes the line twice. + return self.cfg.mode == "read" + + def _act(self, params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None: + target = resolve(self.cfg.path) + + if self.cfg.mode == "read": + if not target.exists(): + raise FileNotFoundError(f"'{self.cfg.path}' does not exist") + raw = target.read_text() + value = json.loads(raw) if self.cfg.format == "json" else raw + return {spec.port: value for spec in self.output_ports} or None + + if not kwargs: + return None + value = next(iter(kwargs.values())) + text = json.dumps(value) if self.cfg.format == "json" else str(value) + if self.cfg.newline: + text += "\n" + + target.parent.mkdir(parents=True, exist_ok=True) + with target.open("a" if self.cfg.mode == "append" else "w") as handle: + handle.write(text) + return None diff --git a/backend/app/flow/nodes/inject.py b/backend/app/flow/nodes/inject.py new file mode 100644 index 0000000..4f7552b --- /dev/null +++ b/backend/app/flow/nodes/inject.py @@ -0,0 +1,156 @@ +"""Inject: the node that starts something, on a timer or on request. + +Node-RED's *inject* is the most placed trigger in a real installation — mostly +as a button someone presses, sometimes on an interval, occasionally once when +everything comes up. +""" + +from __future__ import annotations + +import asyncio +import logging +import time +from collections.abc import Iterable +from typing import TYPE_CHECKING, Any + +from pydantic import BaseModel, ConfigDict, Field + +from app.flow.messages import MessageSpec +from app.flow.nodes.base import Node + +if TYPE_CHECKING: + from fastapi import FastAPI + +logger = logging.getLogger(__name__) + + +class InjectNode(Node): + """Emit a value: on request, every n seconds, on a schedule, or at startup. + + The value is whatever ``payload`` says, or the current time when it says + nothing — a timestamp is what most schedules actually want. + """ + + class Params(BaseModel): + model_config = ConfigDict(extra="allow") + + payload: Any = Field( + default=None, + description="What to emit. Empty emits the current time.", + ) + interval: float = Field( + default=0, + ge=0, + description="Emit every this many seconds. 0 means never on its own.", + ) + cron: str = Field( + default="", + description="A five-field cron expression, if it should follow a schedule.", + ) + at_start: bool = Field( + default=False, + description="Emit once when the flow starts.", + ) + start_delay: float = Field( + default=1.0, + ge=0, + description="How long to wait before the startup emission.", + ) + + __slots__ = ("cfg", "_stop") + + 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 {}) + self._stop: asyncio.Event | None = None + super().__init__( + f=self._emit, + requires=requires, + provides=provides, + params=params, + name=name or "inject", + ) + + def _payload(self) -> Any: + return time.time() if self.cfg.payload is None else self.cfg.payload + + def _emit(self, params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None: + """Every output carries the same value; that is what injecting means.""" + value = self._payload() + return {spec.port: value for spec in self.output_ports} or None + + # ------------------------------------------------------------------------- + # Its own schedule + # ------------------------------------------------------------------------- + + async def start(self, app: FastAPI | None = None) -> None: + if self._stop is not None: + return + if not (self.cfg.interval or self.cfg.cron or self.cfg.at_start): + # A manual inject waits to be pressed. + return + self._stop = asyncio.Event() + self._run_supervised("inject", self._loop) + + async def stop(self, app: FastAPI | None = None) -> None: + if self._stop is None: + return + self._stop.set() + self._stop = None + + async def _loop(self) -> None: + stop = self._stop + if stop is None: + return + + if self.cfg.at_start: + # A moment's grace, so subscribers are listening before it fires. + await self._sleep(stop, self.cfg.start_delay) + if stop.is_set(): + return + await self._fire() + + if self.cfg.cron: + await self._cron_loop(stop) + elif self.cfg.interval: + while not stop.is_set(): + await self._sleep(stop, self.cfg.interval) + if stop.is_set(): + return + await self._fire() + + async def _cron_loop(self, stop: asyncio.Event) -> None: + from datetime import datetime + + from croniter import croniter # type: ignore[import-untyped] + + if not croniter.is_valid(self.cfg.cron): + raise ValueError(f"'{self.cfg.cron}' is not a cron expression") + + cron = croniter(self.cfg.cron, datetime.now()) + while not stop.is_set(): + wait = max(0.0, (cron.get_next(datetime) - datetime.now()).total_seconds()) + await self._sleep(stop, wait) + if stop.is_set(): + return + await self._fire() + + @staticmethod + async def _sleep(stop: asyncio.Event, seconds: float) -> None: + """Wait, but wake immediately if the node is being stopped.""" + try: + await asyncio.wait_for(stop.wait(), timeout=seconds) + except asyncio.TimeoutError: + pass + + async def _fire(self) -> None: + value = self._payload() + outputs = {spec.port: value for spec in self.output_ports} + if outputs: + # inject runs the graph, which is blocking work. + await asyncio.to_thread(self.inject, outputs) diff --git a/backend/app/flow/nodes/logic.py b/backend/app/flow/nodes/logic.py new file mode 100644 index 0000000..898f0ec --- /dev/null +++ b/backend/app/flow/nodes/logic.py @@ -0,0 +1,294 @@ +"""The flow-logic vocabulary: routing, mapping, filtering and joining. + +Anything here could be written as a `python` node — that is what the function +node is for. These exist because the same handful of shapes account for most of +a real installation, and a rule you fill in is easier to read on a canvas, and +to change, than five lines of code repeated eighty times. +""" + +from __future__ import annotations + +import logging +from collections.abc import Iterable +from typing import Any, Literal + +from pydantic import BaseModel, ConfigDict, Field + +from app.flow.messages import MessageSpec +from app.flow.nodes.base import Node + +logger = logging.getLogger(__name__) + +Comparison = Literal["eq", "ne", "gt", "gte", "lt", "lte", "contains", "between"] + + +def compare(value: Any, op: Comparison, operand: Any, operand2: Any = None) -> bool: + """Evaluate one rule against one value, never raising on a bad pairing.""" + try: + if op == "eq": + return bool(value == operand) + if op == "ne": + return bool(value != operand) + if op == "gt": + return bool(value > operand) + if op == "gte": + return bool(value >= operand) + if op == "lt": + return bool(value < operand) + if op == "lte": + return bool(value <= operand) + if op == "contains": + return operand in value + if op == "between": + return bool(operand <= value <= operand2) + except TypeError: + # Comparing a string to a number is a mistake in the rule, not a + # reason to take the flow down. + return False + return False + + +class SwitchNode(Node): + """Send a value down one branch or another, by rule. + + Each rule names an output port; a value that matches leaves through that + port and nothing else. Node-RED's *switch*. + """ + + class Rule(BaseModel): + port: str + op: Comparison = "eq" + value: Any = None + # Only for ``between``. + value2: Any = None + + class Params(BaseModel): + model_config = ConfigDict(extra="allow") + + rules: list[SwitchNode.Rule] = Field( + default_factory=list, + description="Checked in order. Each names the output it routes to.", + ) + stop_at_first: bool = Field( + default=True, + description="Leave through the first matching rule only.", + ) + otherwise: str = Field( + default="", + description="Output for a value that matched nothing.", + ) + + 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 {}) + super().__init__( + f=self._route, + requires=requires, + provides=provides, + params=params, + name=name or "switch", + ) + + def _route(self, params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None: + if not kwargs: + return None + # One input; routing several at once has no obvious meaning. + value = next(iter(kwargs.values())) + + out: dict[str, Any] = {} + for rule in self.cfg.rules: + if compare(value, rule.op, rule.value, rule.value2): + out[rule.port] = value + if self.cfg.stop_at_first: + return out + if not out and self.cfg.otherwise: + out[self.cfg.otherwise] = value + return out or None + + +class ChangeNode(Node): + """Reshape a value on its way past: scale, offset, map, or replace. + + Node-RED's *change*, which is the second most common node in a real + installation after the function. + """ + + class Params(BaseModel): + model_config = ConfigDict(extra="allow") + + scale: float = Field(default=1.0, description="Multiply numbers by this.") + offset: float = Field(default=0.0, description="Then add this.") + round_to: int | None = Field( + default=None, description="Decimal places to round to, if any." + ) + mapping: dict[str, Any] = Field( + default_factory=dict, + description="Replace a value with another, looked up as text.", + ) + default: Any = Field( + default=None, + description="Value to use when the lookup misses. Empty passes it through.", + ) + + 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 {}) + super().__init__( + f=self._change, + requires=requires, + provides=provides, + params=params, + name=name or "change", + ) + + def _convert(self, value: Any) -> Any: + if self.cfg.mapping: + key = str(value) + if key in self.cfg.mapping: + return self.cfg.mapping[key] + if self.cfg.default is not None: + return self.cfg.default + return value + + if isinstance(value, bool) or not isinstance(value, (int, float)): + return value + converted = value * self.cfg.scale + self.cfg.offset + if self.cfg.round_to is not None: + converted = round(converted, self.cfg.round_to) + return converted + + def _change(self, params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None: + if not kwargs: + return None + pairs = list(zip(self.input_ports, self.output_ports, strict=False)) + if not pairs: + return None + return { + out.port: self._convert(kwargs[inp.port]) + for inp, out in pairs + if inp.port in kwargs + } or None + + +class RbeNode(Node): + """Pass a value on only when it has actually changed. + + Node-RED's *rbe* (report by exception). A sensor that publishes the same + reading every two seconds should not wake everything downstream of it. + """ + + class Params(BaseModel): + model_config = ConfigDict(extra="allow") + + deadband: float = Field( + default=0.0, + ge=0, + description="Ignore numeric changes smaller than this.", + ) + deadband_percent: bool = Field( + default=False, description="Read the deadband as a percentage instead." + ) + + 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 {}) + super().__init__( + f=self._filter, + requires=requires, + provides=provides, + params=params, + name=name or "rbe", + ) + + def _changed(self, port: str, value: Any) -> bool: + previous = self.recall(port, _MISSING) + if previous is _MISSING: + return True + if self.cfg.deadband and isinstance(value, (int, float)): + if isinstance(previous, (int, float)): + span = abs(value - previous) + if self.cfg.deadband_percent: + scale = abs(previous) or 1.0 + return (span / scale) * 100 >= self.cfg.deadband + return span >= self.cfg.deadband + return bool(value != previous) + + def _filter(self, params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None: + pairs = list(zip(self.input_ports, self.output_ports, strict=False)) + out: dict[str, Any] = {} + for inp, outp in pairs: + if inp.port not in kwargs: + continue + value = kwargs[inp.port] + if self._changed(inp.port, value): + self.remember(inp.port, value) + out[outp.port] = value + return out or None + + +class JoinNode(Node): + """Gather several inputs into one message. + + The engine already waits for every input a node declares, so joining is + about the shape of the result: an object keyed by port, or a list. + """ + + class Params(BaseModel): + model_config = ConfigDict(extra="allow") + + mode: Literal["object", "array"] = Field( + default="object", description="Combine inputs into an object or a list." + ) + + 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 {}) + # Every input has to be fresh, or a join would emit the same + # combination each time any one of them arrived. + params = dict(params or {}) + params.setdefault("synchronous", True) + super().__init__( + f=self._join, + requires=requires, + provides=provides, + params=params, + name=name or "join", + ) + + def _join(self, params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None: + if not kwargs or not self.output_ports: + return None + ordered = [spec.port for spec in self.input_ports if spec.port in kwargs] + combined: Any + if self.cfg.mode == "array": + combined = [kwargs[port] for port in ordered] + else: + combined = {port: kwargs[port] for port in ordered} + return {self.output_ports[0].port: combined} + + +class _Missing: + """Distinguishes "never seen" from a value that happens to be falsy.""" + + +_MISSING = _Missing() diff --git a/backend/app/flow/nodes/ntfy.py b/backend/app/flow/nodes/ntfy.py new file mode 100644 index 0000000..dc7c175 --- /dev/null +++ b/backend/app/flow/nodes/ntfy.py @@ -0,0 +1,87 @@ +"""Ntfy: push a notification to a phone. + +The flow-level counterpart to the engine's own alerting — this is a flow +deciding something is worth saying, rather than the engine reporting that it +broke. +""" + +from __future__ import annotations + +import logging +from collections.abc import Iterable +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field + +from app.flow.messages import MessageSpec +from app.flow.nodes.base import Node +from app.flow.nodes.http import shared_client + +logger = logging.getLogger(__name__) + + +class NtfyNode(Node): + """Publish an incoming value as a notification on an ntfy topic.""" + + # A notification sent twice is read twice. + idempotent = False + + class Params(BaseModel): + model_config = ConfigDict(extra="allow") + + server: str = Field( + default="https://ntfy.sh", description="Base URL of the ntfy server." + ) + topic: str = Field(description="Topic to publish to.") + title: str = Field(default="", description="Notification title.") + priority: str = Field( + default="default", + description="min, low, default, high or urgent.", + ) + tags: str = Field(default="", description="Comma-separated ntfy tags.") + token: str = Field( + default="", + description="Access token, for a server that needs one.", + json_schema_extra={"x-secret": True}, + ) + timeout: float = Field(default=10.0, gt=0) + + __slots__ = ("cfg",) + + 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 {}) + super().__init__( + f=self._notify, + requires=requires, + provides=provides, + params=params, + name=name or "ntfy", + ) + + def _notify(self, params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None: + if not kwargs: + return None + body = str(next(iter(kwargs.values()))) + + headers = {"Priority": self.cfg.priority} + if self.cfg.title: + headers["Title"] = self.cfg.title + if self.cfg.tags: + headers["Tags"] = self.cfg.tags + if self.cfg.token: + headers["Authorization"] = f"Bearer {self.cfg.token}" + + response = shared_client().post( + f"{self.cfg.server.rstrip('/')}/{self.cfg.topic}", + content=body.encode(), + headers=headers, + timeout=self.cfg.timeout, + ) + response.raise_for_status() + return None diff --git a/backend/app/flow/nodes/trigger.py b/backend/app/flow/nodes/trigger.py new file mode 100644 index 0000000..2f69ff0 --- /dev/null +++ b/backend/app/flow/nodes/trigger.py @@ -0,0 +1,97 @@ +"""Trigger: send one thing now, another if nothing follows. + +Node-RED's *trigger*. The shape it exists for is "the door opened — turn the +light on, and off again in two minutes unless it opens again", which is +awkward to express any other way. +""" + +from __future__ import annotations + +import logging +from collections.abc import Iterable +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field + +from app.flow.messages import MessageSpec +from app.flow.nodes.base import Node + +logger = logging.getLogger(__name__) + + +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. + """ + + class Params(BaseModel): + model_config = ConfigDict(extra="allow") + + first: Any = Field(default=True, description="Sent as soon as a value arrives.") + then: Any = Field( + default=False, + description="Sent when the wait expires. Empty sends nothing.", + ) + wait: float = Field( + default=60.0, gt=0, description="Seconds of quiet before the second value." + ) + extend: bool = Field( + default=True, + description="A value arriving during the wait starts it over.", + ) + + __slots__ = ("cfg",) + + 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 {}) + super().__init__( + f=self._trigger, + requires=requires, + provides=provides, + params=params, + name=name or "trigger", + ) + + def _outputs(self, value: Any) -> dict[str, Any]: + return {spec.port: value for spec in self.output_ports} + + def _trigger(self, params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None: + if not kwargs or not self.output_ports: + return None + + 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. + generation = int(self.recall("generation", 0)) + 1 + self.remember("generation", generation) + self.remember("armed", 1) + + if self.cfg.then is not None and self._pipeline is not None: + deferred = self._outputs(self.cfg.then) + scheduled = self._pipeline.defer( + self, + self._to_messages(deferred) or {}, + self.cfg.wait, + guard=("generation", generation), + ) + if not scheduled: + logger.warning( + "Node '%s' cannot wait without a work queue; sending only " + "its first value", + self.name, + ) + self.remember("armed", 0) + + return self._outputs(self.cfg.first) diff --git a/backend/app/flow/pipeline.py b/backend/app/flow/pipeline.py index dc8e154..4b14525 100644 --- a/backend/app/flow/pipeline.py +++ b/backend/app/flow/pipeline.py @@ -756,9 +756,19 @@ class Pipeline: self.apply_outputs(node, outputs) return self.run_downstream(node) - def defer(self, node: Node, outputs: dict[str, Any], seconds: float) -> bool: + def defer( + self, + node: Node, + outputs: dict[str, Any], + seconds: float, + guard: tuple[str, Any] | None = None, + ) -> bool: """Publish a node's outputs later, without holding a worker thread. + ``guard`` names something the node must still remember when the wait is + over; if it has moved on, the item is dropped. That is how a wait which + gets restarted cancels the one it replaced. + Returns False when there is no queue to hold the item, in which case the caller has to wait however it waited before. """ @@ -772,6 +782,8 @@ class Pipeline: flow=node.flow, outputs=outputs, cause="delay", + guard_key=guard[0] if guard else "", + guard_value=str(guard[1]) if guard else "", ) try: self._queue.add_delayed(item, time.time() + seconds) diff --git a/backend/app/flow/queue.py b/backend/app/flow/queue.py index fc35cfd..84c6a38 100644 --- a/backend/app/flow/queue.py +++ b/backend/app/flow/queue.py @@ -47,6 +47,9 @@ class WorkItem: :param outputs: What the source node emitted (cascade only). :param cause: Where the work came from, for logs and debugging. :param not_before: Epoch seconds before which the item must not run. + :param guard: ``(key, value)`` the target node must still remember for + this item to be worth running — how a rescheduled wait cancels the + one it replaced. :param entry_id: Set by the queue on claim; stable across redeliveries, which is what makes it usable as an idempotency key. :param deliveries: How many times this item has been handed out. @@ -58,6 +61,8 @@ class WorkItem: outputs: dict[str, Any] = field(default_factory=dict) cause: str = "system" not_before: float = 0.0 + guard_key: str = "" + guard_value: str = "" entry_id: str = "" deliveries: int = 1 @@ -69,6 +74,8 @@ class WorkItem: "outputs": json.dumps(self.outputs), "cause": self.cause, "not_before": str(self.not_before), + "guard_key": self.guard_key, + "guard_value": self.guard_value, } @classmethod @@ -82,6 +89,8 @@ class WorkItem: outputs=json.loads(fields.get("outputs") or "{}"), cause=fields.get("cause", "system"), not_before=float(fields.get("not_before") or 0.0), + guard_key=fields.get("guard_key", ""), + guard_value=fields.get("guard_value", ""), entry_id=entry_id, deliveries=deliveries, ) diff --git a/backend/tests/flow/test_node_types.py b/backend/tests/flow/test_node_types.py index df90e66..1e303d6 100644 --- a/backend/tests/flow/test_node_types.py +++ b/backend/tests/flow/test_node_types.py @@ -42,6 +42,54 @@ FIXTURES: dict[str, dict] = { "requires": [MessageSpec(name="temp", dtype=DType.FLOAT)], "provides": [MessageSpec(name="score", dtype=DType.FLOAT)], }, + "inject": { + "params": {"payload": 1.0, "interval": 60}, + "requires": [], + "provides": [MessageSpec(name="tick", dtype=DType.FLOAT)], + }, + "switch": { + "params": {"rules": [{"port": "hot", "op": "gt", "value": 20}]}, + "requires": [MessageSpec(name="temp", dtype=DType.FLOAT)], + "provides": [MessageSpec(name="hot", dtype=DType.FLOAT)], + }, + "change": { + "params": {"scale": 0.1, "offset": -273.15}, + "requires": [MessageSpec(name="temp", dtype=DType.FLOAT)], + "provides": [MessageSpec(name="celsius", dtype=DType.FLOAT)], + }, + "rbe": { + "params": {"deadband": 0.5}, + "requires": [MessageSpec(name="temp", dtype=DType.FLOAT)], + "provides": [MessageSpec(name="changed", dtype=DType.FLOAT)], + }, + "join": { + "params": {"mode": "object"}, + "requires": [ + MessageSpec(name="temp", dtype=DType.FLOAT), + MessageSpec(name="humidity", dtype=DType.FLOAT), + ], + "provides": [MessageSpec(name="reading", dtype=DType.JSON)], + }, + "trigger": { + "params": {"first": True, "then": False, "wait": 120}, + "requires": [MessageSpec(name="motion", dtype=DType.BOOL)], + "provides": [MessageSpec(name="light", dtype=DType.BOOL)], + }, + "exec": { + "params": {"command": "echo hello"}, + "requires": [MessageSpec(name="go", dtype=DType.BOOL)], + "provides": [MessageSpec(name="stdout", dtype=DType.STR)], + }, + "file": { + "params": {"path": "readings.log", "mode": "append"}, + "requires": [MessageSpec(name="line", dtype=DType.STR)], + "provides": [], + }, + "ntfy": { + "params": {"topic": "house", "title": "Alert"}, + "requires": [MessageSpec(name="text", dtype=DType.STR)], + "provides": [], + }, } diff --git a/backend/tests/flow/test_vocabulary.py b/backend/tests/flow/test_vocabulary.py new file mode 100644 index 0000000..173b829 --- /dev/null +++ b/backend/tests/flow/test_vocabulary.py @@ -0,0 +1,256 @@ +"""What the flow-logic nodes actually do. + +These are the shapes a Node-RED installation is mostly made of, so their +behaviour is worth pinning rather than just their construction. +""" + +from app.flow.messages import DType, MessageSpec +from app.flow.nodes import ChangeNode, ExecNode, FileNode, JoinNode, RbeNode, SwitchNode +from app.flow.pipeline import Pipeline +from app.flow.state import MemoryState + + +def spec(name: str, dtype: DType = DType.FLOAT) -> MessageSpec: + return MessageSpec(name=name, port=name, dtype=dtype) + + +def place(node, flow: str = "f"): + """Give a node a flow and a pipeline, as the controller would.""" + node.assign_flow(flow, node.name) + Pipeline(nodes=[node], state=MemoryState()) + return node + + +# --------------------------------------------------------------------------- +# Switch +# --------------------------------------------------------------------------- + + +def switch(**params) -> SwitchNode: + node = SwitchNode( + requires=[spec("temp")], + provides=[spec("hot"), spec("cold")], + params=params, + name="switch", + ) + return place(node) + + +def test_a_value_leaves_through_the_branch_whose_rule_it_matches(): + node = switch( + rules=[ + {"port": "hot", "op": "gt", "value": 20}, + {"port": "cold", "op": "lte", "value": 20}, + ] + ) + + assert node.execute({"f.temp": 25.0}) == {"f.hot": 25.0} + assert node.execute({"f.temp": 15.0}) == {"f.cold": 15.0} + + +def test_a_value_matching_nothing_goes_nowhere_by_default(): + node = switch(rules=[{"port": "hot", "op": "gt", "value": 100}]) + + assert node.execute({"f.temp": 15.0}) is None + + +def test_an_otherwise_branch_catches_what_the_rules_missed(): + node = switch(rules=[{"port": "hot", "op": "gt", "value": 100}], otherwise="cold") + + assert node.execute({"f.temp": 15.0}) == {"f.cold": 15.0} + + +def test_a_rule_comparing_incompatible_things_does_not_take_the_flow_down(): + node = switch(rules=[{"port": "hot", "op": "gt", "value": "twenty"}]) + + assert node.execute({"f.temp": 25.0}) is None + + +# --------------------------------------------------------------------------- +# Change +# --------------------------------------------------------------------------- + + +def change(**params) -> ChangeNode: + node = ChangeNode( + requires=[spec("raw")], + provides=[spec("scaled")], + params=params, + name="change", + ) + return place(node) + + +def test_scaling_and_offsetting_a_reading(): + node = change(scale=0.1, offset=-273.15, round_to=2) + + assert node.execute({"f.raw": 3000.0}) == {"f.scaled": 26.85} + + +def test_mapping_one_value_onto_another(): + # The mapped output is text, and the port has to say so — the type check + # between nodes applies to a change node like any other. + node = ChangeNode( + requires=[spec("raw")], + provides=[spec("label", DType.STR)], + params={"mapping": {"1": "on", "0": "off"}, "default": "unknown"}, + name="change", + ) + place(node) + + assert node.execute({"f.raw": 1}) == {"f.label": "on"} + assert node.execute({"f.raw": 7}) == {"f.label": "unknown"} + + +# --------------------------------------------------------------------------- +# Filter unchanged +# --------------------------------------------------------------------------- + + +def rbe(**params) -> RbeNode: + node = RbeNode( + requires=[spec("temp")], + provides=[spec("changed")], + params=params, + name="rbe", + ) + return place(node) + + +def test_the_same_reading_twice_only_passes_once(): + node = rbe() + + assert node.execute({"f.temp": 21.0}) == {"f.changed": 21.0} + assert node.execute({"f.temp": 21.0}) is None + assert node.execute({"f.temp": 22.0}) == {"f.changed": 22.0} + + +def test_a_deadband_swallows_the_jitter(): + node = rbe(deadband=1.0) + + assert node.execute({"f.temp": 21.0}) == {"f.changed": 21.0} + assert node.execute({"f.temp": 21.4}) is None + assert node.execute({"f.temp": 22.5}) == {"f.changed": 22.5} + + +def test_a_falsy_first_reading_still_counts_as_new(): + """Zero is a reading, not the absence of one.""" + node = rbe() + + assert node.execute({"f.temp": 0.0}) == {"f.changed": 0.0} + + +# --------------------------------------------------------------------------- +# Join +# --------------------------------------------------------------------------- + + +def test_join_gathers_its_inputs_into_one_message(): + node = JoinNode( + requires=[spec("temp"), spec("humidity")], + provides=[spec("reading", DType.JSON)], + params={"mode": "object"}, + name="join", + ) + place(node) + + result = node.execute({"f.temp": 21.0, "f.humidity": 40.0}) + + assert result == {"f.reading": {"temp": 21.0, "humidity": 40.0}} + + +def test_join_can_produce_a_list_instead(): + node = JoinNode( + requires=[spec("a"), spec("b")], + provides=[spec("both", DType.JSON)], + params={"mode": "array"}, + name="join", + ) + place(node) + + assert node.execute({"f.a": 1.0, "f.b": 2.0}) == {"f.both": [1.0, 2.0]} + + +def test_join_waits_for_every_input_to_be_fresh(): + node = JoinNode( + requires=[spec("a"), spec("b")], + provides=[spec("both", DType.JSON)], + name="join", + ) + + # Otherwise one input arriving would re-emit the previous combination. + assert node.synchronous + + +# --------------------------------------------------------------------------- +# Command and file +# --------------------------------------------------------------------------- + + +def test_a_command_hands_back_what_it_printed(): + node = ExecNode( + requires=[spec("go", DType.BOOL)], + provides=[spec("stdout", DType.STR)], + params={"command": "echo hello"}, + name="exec", + ) + place(node) + + assert node.execute({"f.go": True}) == {"f.stdout": "hello\n"} + + +def test_a_command_that_is_not_installed_says_so(): + node = ExecNode( + requires=[spec("go", DType.BOOL)], + provides=[spec("stdout", DType.STR)], + params={"command": "definitely-not-a-real-command"}, + name="exec", + ) + place(node) + + try: + node.execute({"f.go": True}) + except FileNotFoundError as exc: + assert "not available in this container" in str(exc) + else: # pragma: no cover - the command really should not exist + raise AssertionError("expected a FileNotFoundError") + + +def test_a_file_node_refuses_to_leave_its_directory(): + node = FileNode( + provides=[spec("contents", DType.STR)], + params={"path": "../../secrets.enc", "mode": "read"}, + name="file", + ) + place(node) + + try: + node.execute({}) + except ValueError as exc: + assert "outside the files directory" in str(exc) + else: # pragma: no cover + raise AssertionError("expected the path to be refused") + + +def test_a_file_round_trips_through_the_sandbox(tmp_path, monkeypatch): + from app.core.config import settings + + monkeypatch.setattr(settings, "FLOWS_DIR", tmp_path / "flows") + + writer = FileNode( + requires=[spec("line", DType.STR)], + params={"path": "readings.log", "mode": "append"}, + name="writer", + ) + place(writer) + writer.execute({"f.line": "21.0"}) + writer.execute({"f.line": "22.0"}) + + reader = FileNode( + provides=[spec("contents", DType.STR)], + params={"path": "readings.log", "mode": "read"}, + name="reader", + ) + place(reader) + + assert reader.execute({}) == {"f.contents": "21.0\n22.0\n"} diff --git a/frontend/src/components/Flow/FlowNode.tsx b/frontend/src/components/Flow/FlowNode.tsx index 7efeecd..b2face8 100644 --- a/frontend/src/components/Flow/FlowNode.tsx +++ b/frontend/src/components/Flow/FlowNode.tsx @@ -1,5 +1,21 @@ import { Handle, type NodeProps, Position } from "@xyflow/react" -import { Braces, Clock, Code2, Database, Globe, Radio } from "lucide-react" +import { + Bell, + Braces, + Clock, + Code2, + Database, + FileText, + Filter, + Globe, + Merge, + Play, + Radio, + Shuffle, + Split, + Terminal, + Timer, +} from "lucide-react" import { memo } from "react" import type { MessageSpec, NodeDef_Input } from "@/client" @@ -19,6 +35,15 @@ const NODE_ICONS = { influxdb: Database, delay: Clock, mlp: Braces, + inject: Play, + switch: Split, + change: Shuffle, + rbe: Filter, + join: Merge, + trigger: Timer, + exec: Terminal, + file: FileText, + ntfy: Bell, } as const // One dot says everything about a node's state. Idle nodes carry no dot at all,