"""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: 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, )