stroblmeandClaude Opus 5
ba707c8051
Four small things, each with a device behind it.
An MQTT filter now routes what it subscribed to. `+` and `#` reached the
broker and were then looked up in an exact-match dict, so every message a
wildcard subscription received was dropped in silence.
`json_key` lifts a value out of the object a device wraps it in — Victron
publishes `{"value": 47}` on every path, which was otherwise a Python node
per port.
The trigger node learned `passthrough` and `wait_port`, because how long to
wait can be a value rather than a constant: a rollershutter takes 26 seconds
up and 28 down. A wait of zero sends nothing afterwards and still cancels
what the last message scheduled, which is how a stop is commanded once
instead of forever.
The HTTP sender takes fixed `query` parameters, so an API key is a secret
reference rather than a message on the canvas, and `send_inputs` off for a
request whose inputs are only a trigger.
Also: `delay` accepts fractional seconds, and `TZ` reaches the container, so
a cron expression means local time. Left unset it is UTC, as before.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
334 lines
11 KiB
Python
334 lines
11 KiB
Python
"""Timing nodes: fixed delay, rate limit, and the cron schedule."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import time
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from pydantic import BaseModel, ConfigDict
|
|
|
|
from fluksio.flow.nodes.base import Node
|
|
|
|
if TYPE_CHECKING:
|
|
from fastapi import FastAPI
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class DelayNode(Node):
|
|
"""
|
|
A node that adds delay, rate-limiting, and/or cron-scheduled emissions.
|
|
|
|
This node can:
|
|
- Add a fixed delay before forwarding messages
|
|
- Rate-limit messages to a minimum interval between forwards
|
|
- Schedule emissions using crontab syntax (like a cron job)
|
|
|
|
The node passes through all input values to outputs with matching names.
|
|
If ``requires`` and ``provides`` have the same message names, the values are
|
|
forwarded directly. Otherwise, you can specify a ``mapping`` in params.
|
|
|
|
**Cron functionality:**
|
|
- Without ``requires``: Emits current timestamp at each cron tick
|
|
- With ``requires``: Emits last received input at each cron tick
|
|
|
|
**Execution order:** rate check → delay → forward
|
|
|
|
:param params: Parameters dict containing:
|
|
- ``delay`` (int): Fixed delay in seconds before forwarding (default: 0)
|
|
- ``interval`` (int): Minimum interval in seconds between forwards (default: 0)
|
|
- ``mapping`` (dict): Optional mapping from input names to output names
|
|
- ``cron`` (str): Crontab expression for scheduled emissions (optional).
|
|
Standard 5-field format: ``minute hour day-of-month month day-of-week``
|
|
:type params: dict
|
|
:param kwargs: Additional arguments passed to Node (requires, provides, name)
|
|
|
|
:example:
|
|
Simple passthrough with delay:
|
|
|
|
>>> delay_node = DelayNode(
|
|
... requires=[MessageSpec(name="temperature", dtype=DType.FLOAT)],
|
|
... provides=[MessageSpec(name="temperature", dtype=DType.FLOAT)],
|
|
... params={"delay": 1},
|
|
... )
|
|
|
|
Rate-limited forwarding:
|
|
|
|
>>> throttle_node = DelayNode(
|
|
... requires=[MessageSpec(name="sensor_data", dtype=DType.FLOAT)],
|
|
... provides=[MessageSpec(name="sensor_data", dtype=DType.FLOAT)],
|
|
... params={"interval": 5},
|
|
... )
|
|
|
|
Cron: emit timestamp every day at 08:30:
|
|
|
|
>>> alarm_node = DelayNode(
|
|
... provides=[MessageSpec(name="timestamp", dtype=DType.FLOAT)],
|
|
... params={"cron": "30 8 * * *"},
|
|
... )
|
|
|
|
Cron: emit stored input every 5 minutes:
|
|
|
|
>>> periodic_node = DelayNode(
|
|
... requires=[MessageSpec(name="value", dtype=DType.FLOAT)],
|
|
... provides=[MessageSpec(name="value", dtype=DType.FLOAT)],
|
|
... params={"cron": "*/5 * * * *"},
|
|
... )
|
|
|
|
Cron: emit on weekdays at 23:59:
|
|
|
|
>>> weekday_node = DelayNode(
|
|
... requires=[MessageSpec(name="daily_summary", dtype=DType.FLOAT)],
|
|
... provides=[MessageSpec(name="daily_summary", dtype=DType.FLOAT)],
|
|
... params={"cron": "59 23 * * 1-5"},
|
|
... )
|
|
"""
|
|
|
|
__slots__ = (
|
|
"delay",
|
|
"interval",
|
|
"ts",
|
|
"mapping",
|
|
"cron_expr",
|
|
"last_input",
|
|
"_cron_task",
|
|
"_stop_cron",
|
|
)
|
|
|
|
class Params(BaseModel):
|
|
model_config = ConfigDict(extra="allow")
|
|
|
|
# Seconds, fractional: a shutter's run-time is not a whole number.
|
|
delay: float = 0
|
|
interval: float = 0
|
|
# Input port → output port; ports are paired in order when empty.
|
|
mapping: dict[str, str] = {}
|
|
cron: str | None = None
|
|
|
|
def __init__(self, params: dict[str, Any] | None = None, **kwargs: Any) -> None:
|
|
cfg = self.Params.model_validate(params or {})
|
|
self.delay = cfg.delay
|
|
self.interval = cfg.interval
|
|
self.mapping = cfg.mapping
|
|
self.cron_expr = cfg.cron
|
|
self.ts = 0.0
|
|
self.last_input: dict[str, Any] = {}
|
|
self._cron_task: asyncio.Task[None] | None = None
|
|
self._stop_cron: asyncio.Event | None = None
|
|
|
|
super().__init__(f=self._f, params=params, **kwargs)
|
|
|
|
def _port_pairs(self) -> list[tuple[str, str]]:
|
|
"""Input port → output port pairs this node forwards along."""
|
|
if self.mapping:
|
|
return list(self.mapping.items())
|
|
return [
|
|
(i.port, o.port)
|
|
for i, o in zip(self.input_ports, self.output_ports, strict=False)
|
|
]
|
|
|
|
def _f(self, params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None:
|
|
"""Forward messages with optional delay, rate-limiting, and alarm."""
|
|
|
|
logger.info("[%s] Received %s", self.name, kwargs)
|
|
|
|
# Store last input for cron use
|
|
if kwargs:
|
|
self.last_input = dict(kwargs)
|
|
|
|
# Apply rate limiting first
|
|
if self.interval > 0:
|
|
ts = time.time()
|
|
if ts <= self.ts + self.interval:
|
|
logger.info("[%s] Stashing %s", self.name, kwargs)
|
|
return None
|
|
self.ts = ts
|
|
|
|
if not kwargs:
|
|
return None
|
|
|
|
output = {
|
|
out_port: kwargs[in_port]
|
|
for in_port, out_port in self._port_pairs()
|
|
if in_port in kwargs
|
|
}
|
|
|
|
if self.delay > 0:
|
|
# Hand the wait to the queue rather than sit on a worker thread:
|
|
# a handful of delay nodes would otherwise occupy the whole pool
|
|
# and the engine would stop dead until they woke up.
|
|
if self._pipeline is not None and self._pipeline.defer(
|
|
self, self._to_messages(output) or {}, self.delay
|
|
):
|
|
logger.info("[%s] Sending %s in %ss", self.name, output, self.delay)
|
|
return None
|
|
time.sleep(self.delay)
|
|
|
|
logger.info("[%s] Sending %s", self.name, output)
|
|
return output
|
|
|
|
# -----------------------------------------------------------------
|
|
# Cron scheduler
|
|
# -----------------------------------------------------------------
|
|
|
|
async def start(self, app: FastAPI | None = None) -> None:
|
|
"""Only a node with a schedule has anything to run on its own."""
|
|
if self.cron_expr:
|
|
await self.start_cron()
|
|
|
|
async def stop(self, app: FastAPI | None = None) -> None:
|
|
if self.cron_expr:
|
|
await self.stop_cron()
|
|
|
|
async def start_cron(self) -> None:
|
|
"""
|
|
Start the cron scheduler.
|
|
|
|
Requires the ``croniter`` package. The scheduler runs in the background
|
|
and triggers emissions according to the ``cron`` expression in params.
|
|
|
|
:raises ValueError: If no ``cron`` expression is configured.
|
|
"""
|
|
if not self.cron_expr:
|
|
logger.warning("No cron expression configured for node '%s'", self.name)
|
|
return
|
|
|
|
if self._cron_task is not None:
|
|
logger.info("Cron already running for node '%s'", self.name)
|
|
return
|
|
|
|
self._stop_cron = asyncio.Event()
|
|
self._cron_task = self._run_supervised("cron", self._cron_loop)
|
|
logger.info("Started cron for node '%s': %s", self.name, self.cron_expr)
|
|
|
|
async def stop_cron(self) -> None:
|
|
"""Stop the cron scheduler."""
|
|
if self._stop_cron is None:
|
|
return
|
|
|
|
self._stop_cron.set()
|
|
|
|
if self._cron_task is not None:
|
|
self._cron_task.cancel()
|
|
try:
|
|
await self._cron_task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
|
|
self._cron_task = None
|
|
self._stop_cron = None
|
|
logger.info("Stopped cron for node '%s'", self.name)
|
|
|
|
async def _cron_loop(self) -> None:
|
|
"""Background loop that sleeps until the next cron tick and triggers."""
|
|
from datetime import datetime
|
|
|
|
try:
|
|
from croniter import croniter # type: ignore[import-untyped]
|
|
except ImportError:
|
|
logger.error(
|
|
"croniter package is required for cron scheduling. "
|
|
"Install it with: pip install croniter"
|
|
)
|
|
return
|
|
|
|
if not croniter.is_valid(self.cron_expr):
|
|
logger.error(
|
|
"Invalid cron expression '%s' for node '%s'",
|
|
self.cron_expr,
|
|
self.name,
|
|
)
|
|
return
|
|
|
|
cron = croniter(self.cron_expr, datetime.now())
|
|
|
|
# Walrus so the body can use the event without re-reading a field
|
|
# stop_cron() may have cleared in the meantime.
|
|
while (stop := self._stop_cron) is not None and not stop.is_set():
|
|
try:
|
|
# Compute seconds until next tick
|
|
next_dt = cron.get_next(datetime)
|
|
now = datetime.now()
|
|
wait_seconds = max(0, (next_dt - now).total_seconds())
|
|
|
|
logger.info(
|
|
"Node '%s' cron: next tick at %s (in %.1fs)",
|
|
self.name,
|
|
next_dt.isoformat(),
|
|
wait_seconds,
|
|
)
|
|
|
|
# Sleep until next tick (wake up on stop signal)
|
|
try:
|
|
await asyncio.wait_for(stop.wait(), timeout=wait_seconds)
|
|
# If we get here, stop was requested
|
|
break
|
|
except asyncio.TimeoutError:
|
|
# Timeout means it's time to fire
|
|
pass
|
|
|
|
logger.info(
|
|
"Cron triggered for node '%s' (%s)",
|
|
self.name,
|
|
self.cron_expr,
|
|
)
|
|
await self._trigger_cron()
|
|
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception:
|
|
# One bad tick should not cost the schedule; the supervisor
|
|
# picks it up from here, backoff included.
|
|
logger.error(
|
|
"Error in cron loop for node '%s'", self.name, exc_info=True
|
|
)
|
|
raise
|
|
|
|
async def _trigger_cron(self) -> None:
|
|
"""Emit data into the pipeline on a cron tick."""
|
|
if self._pipeline is None:
|
|
logger.warning(
|
|
"Node '%s' not bound to pipeline, cannot trigger cron",
|
|
self.name,
|
|
)
|
|
return
|
|
|
|
try:
|
|
output = {}
|
|
|
|
if not self.input_ports:
|
|
# Nothing to forward — emit the current time on every output.
|
|
output = {spec.port: time.time() for spec in self.output_ports}
|
|
logger.info("Cron emission for node '%s': timestamp", self.name)
|
|
elif self.last_input:
|
|
output = {
|
|
out_port: self.last_input[in_port]
|
|
for in_port, out_port in self._port_pairs()
|
|
if in_port in self.last_input
|
|
}
|
|
logger.info(
|
|
"Cron emission for node '%s': stored input %s",
|
|
self.name,
|
|
output,
|
|
)
|
|
else:
|
|
logger.warning(
|
|
"Cron for node '%s' triggered but no input stored yet",
|
|
self.name,
|
|
)
|
|
return
|
|
|
|
if output:
|
|
await asyncio.to_thread(
|
|
self._pipeline.trigger, self, self._to_messages(output)
|
|
)
|
|
except Exception as e:
|
|
logger.error(
|
|
"Error triggering cron for node '%s': %s",
|
|
self.name,
|
|
e,
|
|
exc_info=True,
|
|
)
|