Add the vocabulary a real installation is mostly made of
Every one of these could be written as a function node, and until now had to be. But nine node types account for most of what an actual Node-RED installation contains — inject alone is placed 58 times, change 62 — and a rule you fill in reads better on a canvas, and survives editing better, than the same five lines repeated eighty times. Inject, switch, change, filter-unchanged, join, trigger, command, file and ntfy. They configure themselves through the editor's generated form, so none of them needed frontend work beyond an icon. The stateful ones (filter-unchanged, trigger) keep what they remember in the engine's own state under a reserved prefix, never as a message. That is the line: code you write is a pure function of its inputs, node types the engine ships may remember things. Verified against a live instance — a hook feeding change into filter-unchanged into switch scales a raw reading, swallows a change inside the deadband, and routes the rest to the right branch. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011LF61rxW1FG5YCD2J9YqjY
This commit is contained in:
root
@@ -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),
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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",
|
||||
]
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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)
|
||||
@@ -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()
|
||||
@@ -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
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
@@ -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": [],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -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"}
|
||||
Reference in New Issue
Block a user