Rename the trigger node type to debounce, migrating stored flows
The store rewrites `"type": "trigger"` to `debounce` in every published and draft flow when it opens, bumps the version and commits once. No alias stays behind, so `trigger` is now an unknown type.
This commit is contained in:
@@ -40,6 +40,7 @@ from fluksio.flow.messages import MessageSpec, flow_of, qualify
|
||||
from fluksio.flow.nodes import (
|
||||
RESERVED_SETTINGS,
|
||||
ChangeNode,
|
||||
DebounceNode,
|
||||
DelayNode,
|
||||
ExecNode,
|
||||
FileNode,
|
||||
@@ -53,7 +54,6 @@ from fluksio.flow.nodes import (
|
||||
NtfyNode,
|
||||
RbeNode,
|
||||
SwitchNode,
|
||||
TriggerNode,
|
||||
WebSocketNode,
|
||||
WrenNode,
|
||||
)
|
||||
@@ -408,11 +408,11 @@ NODE_TYPES: dict[str, NodeType] = {
|
||||
cls=JoinNode,
|
||||
params_schema=_schema_of(JoinNode),
|
||||
),
|
||||
"trigger": NodeType(
|
||||
title="Trigger",
|
||||
"debounce": NodeType(
|
||||
title="Debounce",
|
||||
description="Send one value now and another once things go quiet.",
|
||||
cls=TriggerNode,
|
||||
params_schema=_schema_of(TriggerNode),
|
||||
cls=DebounceNode,
|
||||
params_schema=_schema_of(DebounceNode),
|
||||
),
|
||||
"exec": NodeType(
|
||||
title="Command",
|
||||
|
||||
@@ -5,6 +5,7 @@ Split by the outside world each one talks to. Importing from
|
||||
"""
|
||||
|
||||
from fluksio.flow.nodes.base import RESERVED_SETTINGS, Node
|
||||
from fluksio.flow.nodes.debounce import DebounceNode
|
||||
from fluksio.flow.nodes.delay import DelayNode
|
||||
from fluksio.flow.nodes.exec import ExecNode
|
||||
from fluksio.flow.nodes.file import FileNode
|
||||
@@ -15,12 +16,12 @@ from fluksio.flow.nodes.logic import ChangeNode, JoinNode, RbeNode, SwitchNode
|
||||
from fluksio.flow.nodes.mlp import MLPNode
|
||||
from fluksio.flow.nodes.mqtt import MqttNode
|
||||
from fluksio.flow.nodes.ntfy import NtfyNode
|
||||
from fluksio.flow.nodes.trigger import TriggerNode
|
||||
from fluksio.flow.nodes.websocket import WebSocketNode
|
||||
from fluksio.flow.nodes.wren import WrenNode
|
||||
|
||||
__all__ = [
|
||||
"ChangeNode",
|
||||
"DebounceNode",
|
||||
"DelayNode",
|
||||
"ExecNode",
|
||||
"FileNode",
|
||||
@@ -35,7 +36,6 @@ __all__ = [
|
||||
"RESERVED_SETTINGS",
|
||||
"RbeNode",
|
||||
"SwitchNode",
|
||||
"TriggerNode",
|
||||
"WebSocketNode",
|
||||
"WrenNode",
|
||||
]
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
"""Trigger: send one thing now, another if nothing follows.
|
||||
"""Debounce: 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.
|
||||
Node-RED calls this node *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
|
||||
@@ -19,7 +19,7 @@ from fluksio.flow.nodes.base import Node
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class TriggerNode(Node):
|
||||
class DebounceNode(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
|
||||
@@ -67,17 +67,17 @@ class TriggerNode(Node):
|
||||
):
|
||||
self.cfg = self.Params.model_validate(params or {})
|
||||
super().__init__(
|
||||
f=self._trigger,
|
||||
f=self._debounce,
|
||||
requires=requires,
|
||||
provides=provides,
|
||||
params=params,
|
||||
name=name or "trigger",
|
||||
name=name or "debounce",
|
||||
)
|
||||
|
||||
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:
|
||||
def _debounce(self, params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None:
|
||||
if not kwargs or not self.output_ports:
|
||||
return None
|
||||
|
||||
@@ -166,6 +166,40 @@ class FlowStore:
|
||||
if not (self.root / ".git").exists():
|
||||
self._git("init", "-q")
|
||||
self._commit("Initialise flow store", allow_empty=True)
|
||||
self._migrate()
|
||||
|
||||
def _migrate(self) -> None:
|
||||
"""Rewrite node types stored under a name they no longer have.
|
||||
|
||||
The node type ``trigger`` became ``debounce`` on 2026-10-15, with no
|
||||
alias left for the old name. Each published and draft document is read
|
||||
as plain JSON and only the ``type`` of a matching node changes, so a
|
||||
field this version does not model survives, and the old document stays
|
||||
in git. The version goes up by one, so an editor still holding the old
|
||||
document gets a conflict on its next save instead of writing the old
|
||||
name back. A store with nothing to rename is left alone.
|
||||
"""
|
||||
changed = []
|
||||
paths = [*self.root.glob("*/flow.json"), *self.root.glob("*/flow.draft.json")]
|
||||
for path in sorted(paths):
|
||||
try:
|
||||
data = json.loads(path.read_text())
|
||||
nodes = [n for n in data["nodes"] if n.get("type") == "trigger"]
|
||||
except Exception:
|
||||
# Unreadable to the engine too, which reports it when it loads.
|
||||
continue
|
||||
if not nodes:
|
||||
continue
|
||||
for node in nodes:
|
||||
node["type"] = "debounce"
|
||||
data["version"] = data.get("version", 1) + 1
|
||||
path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n")
|
||||
changed.append(f"{path.parent.name}/{path.name}")
|
||||
if changed:
|
||||
logger.warning(
|
||||
"Renamed node type 'trigger' to 'debounce' in %s", ", ".join(changed)
|
||||
)
|
||||
self._commit("Rename node type 'trigger' to 'debounce'")
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# git
|
||||
|
||||
@@ -8,8 +8,8 @@ come up than to go down.
|
||||
|
||||
from fluksio.flow.messages import DType, MessageSpec
|
||||
from fluksio.flow.nodes import MqttNode
|
||||
from fluksio.flow.nodes.debounce import DebounceNode
|
||||
from fluksio.flow.nodes.mqtt import topic_matches
|
||||
from fluksio.flow.nodes.trigger import TriggerNode
|
||||
|
||||
# ── MQTT topic filters ────────────────────────────────────────────────────
|
||||
|
||||
@@ -72,7 +72,7 @@ def test_without_json_key_the_port_name_is_the_key_as_before():
|
||||
assert node.json_keys == {}
|
||||
|
||||
|
||||
# ── trigger: passthrough and a per-message wait ───────────────────────────
|
||||
# ── debounce: passthrough and a per-message wait ──────────────────────────
|
||||
|
||||
|
||||
class _Pipeline:
|
||||
@@ -87,8 +87,8 @@ class _Pipeline:
|
||||
return True
|
||||
|
||||
|
||||
def _trigger(**params) -> tuple[TriggerNode, _Pipeline]:
|
||||
node = TriggerNode(
|
||||
def _debounce(**params) -> tuple[DebounceNode, _Pipeline]:
|
||||
node = DebounceNode(
|
||||
requires=[
|
||||
MessageSpec(name="run", port="run", dtype=DType.STR),
|
||||
MessageSpec(name="run_for", port="run_for", dtype=DType.FLOAT),
|
||||
@@ -103,30 +103,30 @@ def _trigger(**params) -> tuple[TriggerNode, _Pipeline]:
|
||||
|
||||
|
||||
def test_the_incoming_value_passes_through_and_the_wait_comes_from_a_port():
|
||||
node, pipeline = _trigger()
|
||||
node, pipeline = _debounce()
|
||||
|
||||
assert node._trigger({}, run="DOWN", run_for=28.0) == {"cmd": "DOWN"}
|
||||
assert node._debounce({}, run="DOWN", run_for=28.0) == {"cmd": "DOWN"}
|
||||
(outputs, seconds, _guard) = pipeline.deferred[-1]
|
||||
assert outputs == {"shutters.cmd": "STOP"} and seconds == 28.0
|
||||
|
||||
node._trigger({}, run="UP", run_for=26.0)
|
||||
node._debounce({}, run="UP", run_for=26.0)
|
||||
assert pipeline.deferred[-1][1] == 26.0
|
||||
|
||||
|
||||
def test_a_second_command_invalidates_the_stop_the_first_one_scheduled():
|
||||
node, pipeline = _trigger()
|
||||
node._trigger({}, run="DOWN", run_for=28.0)
|
||||
node, pipeline = _debounce()
|
||||
node._debounce({}, run="DOWN", run_for=28.0)
|
||||
first = pipeline.deferred[-1][2]
|
||||
node._trigger({}, run="UP", run_for=26.0)
|
||||
node._debounce({}, run="UP", run_for=26.0)
|
||||
assert pipeline.deferred[-1][2] != first
|
||||
|
||||
|
||||
def test_a_wait_of_zero_sends_nothing_afterwards():
|
||||
"""STOP is commanded once. Nothing follows it, forever."""
|
||||
node, pipeline = _trigger()
|
||||
node._trigger({}, run="DOWN", run_for=28.0)
|
||||
node, pipeline = _debounce()
|
||||
node._debounce({}, run="DOWN", run_for=28.0)
|
||||
scheduled = len(pipeline.deferred)
|
||||
|
||||
assert node._trigger({}, run="STOP", run_for=0.0) == {"cmd": "STOP"}
|
||||
assert node._debounce({}, run="STOP", run_for=0.0) == {"cmd": "STOP"}
|
||||
assert len(pipeline.deferred) == scheduled
|
||||
assert node.recall("armed", 0) == 0
|
||||
|
||||
@@ -75,7 +75,7 @@ FIXTURES: dict[str, dict] = {
|
||||
],
|
||||
"provides": [MessageSpec(name="reading", dtype=DType.JSON)],
|
||||
},
|
||||
"trigger": {
|
||||
"debounce": {
|
||||
"params": {"first": True, "then": False, "wait": 120},
|
||||
"requires": [MessageSpec(name="motion", dtype=DType.BOOL)],
|
||||
"provides": [MessageSpec(name="light", dtype=DType.BOOL)],
|
||||
|
||||
@@ -131,3 +131,31 @@ def test_renaming_onto_an_existing_name_is_refused(store: FlowStore):
|
||||
store.rename_flow("heating", "warmth")
|
||||
|
||||
assert store.list_flows() == ["heating", "warmth"]
|
||||
|
||||
|
||||
def test_opening_the_store_renames_trigger_nodes_to_debounce(store: FlowStore):
|
||||
"""The old name has no alias, so what is stored under it is rewritten once."""
|
||||
store.write_flow(
|
||||
FlowDef(
|
||||
name="hall",
|
||||
nodes=[NodeDef(id="hold", type="trigger"), NodeDef(id="lamp")],
|
||||
)
|
||||
)
|
||||
store.write_draft(
|
||||
FlowDef(name="hall", nodes=[NodeDef(id="hold", type="trigger")]),
|
||||
base_version=1,
|
||||
)
|
||||
commits = commit_count(store)
|
||||
|
||||
reopened = FlowStore(store.root)
|
||||
|
||||
published = reopened.read_flow("hall")
|
||||
draft = reopened.read_flow("hall", draft=True)
|
||||
assert [node.type for node in published.nodes] == ["debounce", "python"]
|
||||
assert [node.type for node in draft.nodes] == ["debounce"]
|
||||
# Bumped, so an editor holding the old draft cannot save `trigger` back.
|
||||
assert (published.version, draft.version) == (2, 3)
|
||||
assert commit_count(store) == commits + 1
|
||||
|
||||
FlowStore(store.root)
|
||||
assert commit_count(store) == commits + 1
|
||||
|
||||
@@ -187,11 +187,12 @@ Timers wait until their deadline; busy workers can delay execution.
|
||||
Batch flows reject rate-limited non-streaming ports. Runs have no live
|
||||
timer to release a held value.
|
||||
|
||||
### Trigger
|
||||
### Debounce
|
||||
|
||||
**`trigger`** — send one value now and another once things go quiet. Despite
|
||||
the name, a debounce and hold rather than a scheduler: everything it sends
|
||||
starts from a value arriving. For a cron tick, see [`inject`](#inject).
|
||||
**`debounce`** — send one value now and another once things go quiet. A
|
||||
debounce and hold rather than a scheduler: everything it sends starts from a
|
||||
value arriving. For a cron tick, see [`inject`](#inject). Node-RED calls this
|
||||
node *trigger*.
|
||||
|
||||
| Setting | Default | Notes |
|
||||
|---|---|---|
|
||||
@@ -363,7 +364,7 @@ Each record in `actions` contains `message`, `value`, `applied`, `delay_s`,
|
||||
applied. Keep `apply=false` while checking how the model interprets commands.
|
||||
|
||||
A call with a delay ("in 10m") is held by the engine's work queue, like the
|
||||
Trigger node's second value: it survives a restart where the queue is in
|
||||
Debounce node's second value: it survives a restart where the queue is in
|
||||
Redis, and not otherwise.
|
||||
|
||||
The node adds default devices to the model's view for missing expected devices.
|
||||
|
||||
@@ -55,7 +55,7 @@ const NODE_ICONS = {
|
||||
change: Shuffle,
|
||||
rbe: Filter,
|
||||
join: Merge,
|
||||
trigger: Timer,
|
||||
debounce: Timer,
|
||||
exec: Terminal,
|
||||
file: FileText,
|
||||
ntfy: Bell,
|
||||
|
||||
Reference in New Issue
Block a user