stroblme
4a776e52c4
The interval now rides the output ports, so the pipeline's own hold and flush keep what arrives inside the window instead of the node dropping it.
329 lines
11 KiB
Python
329 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, sending the
|
|
latest one held inside the window when the window ends
|
|
- 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:** delay → rate check → 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",
|
|
"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.last_input: dict[str, Any] = {}
|
|
self._cron_task: asyncio.Task[None] | None = None
|
|
self._stop_cron: asyncio.Event | None = None
|
|
|
|
if cfg.interval > 0:
|
|
# The engine's own rate limit, on every output: it keeps the latest
|
|
# value inside the window and publishes it when the window ends.
|
|
kwargs["provides"] = [
|
|
spec.model_copy(update={"interval": max(spec.interval, cfg.interval)})
|
|
for spec in self._normalize_ports(kwargs.get("provides", ()))
|
|
]
|
|
|
|
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, after a delay if one is set."""
|
|
|
|
logger.debug("[%s] Received %s", self.name, kwargs)
|
|
|
|
# Store last input for cron use
|
|
if kwargs:
|
|
self.last_input = dict(kwargs)
|
|
|
|
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.debug("[%s] Sending %s in %ss", self.name, output, self.delay)
|
|
return None
|
|
time.sleep(self.delay)
|
|
|
|
logger.debug("[%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:
|
|
await self._cancel_task(self._cron_task)
|
|
|
|
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 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,
|
|
)
|