Melvin StroblandClaude Fable 5
01af7787c1
- MQTT nodes failed to build: the topic map was renamed to talk in ports, but __slots__ still declared the old name, so every MQTT node raised AttributeError. Building one of each node type is now a test, since __slots__ makes this failure invisible until someone places the node. - Port names offer the messages already in play: everything published is worth reading, and an input nobody provides yet is worth publishing. A message only connects when both ends spell it the same way, so choosing beats typing. - Adding a port focuses its name field. - Dragging onto an input that already reads something offers the extra port as well as the replacement — an MQTT or InfluxDB node usually wants both. - The template's Item model, its routes, screens and table are gone. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016WzrvW7rjQbynnhF6pxh6i
1929 lines
68 KiB
Python
1929 lines
68 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import threading
|
|
import time
|
|
from collections.abc import Callable, Iterable
|
|
from enum import Enum
|
|
from typing import TYPE_CHECKING, Any, Literal
|
|
|
|
import httpx
|
|
import numpy as np
|
|
from pydantic import BaseModel, ConfigDict
|
|
|
|
from app.flow.messages import MessageSpec, qualify
|
|
|
|
if TYPE_CHECKING:
|
|
from fastapi import FastAPI
|
|
|
|
from app.flow.pipeline import Pipeline
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class Node:
|
|
"""
|
|
A pipeline node that wraps a function with typed inputs/outputs.
|
|
|
|
Nodes are the fundamental building blocks of a pipeline. Each node encapsulates
|
|
a function that processes data, with explicit message-based inputs (requires)
|
|
and outputs (provides). Nodes can be connected into a directed acyclic graph (DAG)
|
|
where data flows from upstream to downstream nodes.
|
|
|
|
:param f: The function to execute when the node runs.
|
|
:type f: Callable
|
|
:param requires: Input messages this node consumes. Can be a single Message
|
|
or list of Messages. Empty for source nodes.
|
|
:type requires: MessageSpec | list[MessageSpec]
|
|
:param provides: Output messages this node produces. Can be a single Message
|
|
or list of Messages.
|
|
:type provides: MessageSpec | list[MessageSpec]
|
|
:param params: Additional parameters passed to the function during execution.
|
|
:type params: dict
|
|
:param name: Optional name for the node. Defaults to function name.
|
|
:type name: str | None
|
|
|
|
:ivar synchronous: If True, node only executes when all required inputs have
|
|
new versions since last execution. Useful for synchronizing multiple streams.
|
|
:vartype synchronous: bool
|
|
|
|
:example:
|
|
>>> def process_temp(temperature, params):
|
|
... return {"celsius": temperature * 0.5 + 32}
|
|
>>>
|
|
>>> temp_node = Node(
|
|
... f=process_temp,
|
|
... requires=MessageSpec(name="temperature", dtype=DType.FLOAT),
|
|
... provides=MessageSpec(name="celsius", dtype=DType.FLOAT),
|
|
... params={},
|
|
... )
|
|
"""
|
|
|
|
__slots__ = (
|
|
"f",
|
|
"id",
|
|
"flow",
|
|
"name",
|
|
"input_ports",
|
|
"output_ports",
|
|
"requires",
|
|
"provides",
|
|
"params",
|
|
"_pipeline",
|
|
"synchronous",
|
|
)
|
|
|
|
def __init__(
|
|
self,
|
|
f: Callable,
|
|
requires: MessageSpec | Iterable[MessageSpec] = (),
|
|
provides: MessageSpec | Iterable[MessageSpec] = (),
|
|
params: dict[str, Any] | None = None,
|
|
name: str | None = None,
|
|
):
|
|
self.f = f
|
|
self._pipeline: Pipeline | None = None
|
|
self.params = dict(params) if params else {}
|
|
self.synchronous = bool(self.params.get("synchronous", False))
|
|
|
|
self.input_ports = self._normalize_ports(requires)
|
|
self.output_ports = self._normalize_ports(provides)
|
|
|
|
self.flow = ""
|
|
self.name = name or getattr(f, "__name__", "node")
|
|
self.id = self.name
|
|
self._index_ports()
|
|
|
|
@staticmethod
|
|
def _normalize_ports(
|
|
msgs: MessageSpec | Iterable[MessageSpec],
|
|
) -> list[MessageSpec]:
|
|
"""Accept a single spec or any iterable of them."""
|
|
if isinstance(msgs, MessageSpec):
|
|
return [msgs]
|
|
return list(msgs or ())
|
|
|
|
def _index_ports(self) -> None:
|
|
"""Index bound ports by message name; unbound ports have no wiring."""
|
|
self.requires = {s.name: s for s in self.input_ports if s.name}
|
|
self.provides = {s.name: s for s in self.output_ports if s.name}
|
|
|
|
def assign_flow(self, flow: str, node_id: str) -> None:
|
|
"""Place this node in a flow, qualifying its identity and messages.
|
|
|
|
Called once by the loader, before the pipeline is built.
|
|
"""
|
|
self.flow = flow
|
|
self.name = node_id
|
|
self.id = f"{flow}.{node_id}"
|
|
self.input_ports = [
|
|
s.model_copy(update={"name": qualify(flow, s.name)})
|
|
for s in self.input_ports
|
|
]
|
|
self.output_ports = [
|
|
s.model_copy(update={"name": qualify(flow, s.name)})
|
|
for s in self.output_ports
|
|
]
|
|
self._index_ports()
|
|
|
|
@property
|
|
def local_id(self) -> str:
|
|
"""The node's id within its flow."""
|
|
return self.name
|
|
|
|
def bind(self, pipeline: Pipeline) -> None:
|
|
"""
|
|
Bind this node to a pipeline for external triggering.
|
|
|
|
Once bound, the node can trigger downstream execution when called.
|
|
This is typically done automatically during pipeline construction.
|
|
|
|
:param pipeline: The pipeline to bind this node to.
|
|
:type pipeline: Pipeline
|
|
"""
|
|
self._pipeline = pipeline
|
|
|
|
def __repr__(self) -> str:
|
|
return self.id
|
|
|
|
def __hash__(self) -> int:
|
|
return hash(self.id)
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Port translation
|
|
#
|
|
# The graph speaks qualified message names; node functions speak ports.
|
|
# -------------------------------------------------------------------------
|
|
|
|
def _to_kwargs(self, inputs: dict) -> dict:
|
|
kwargs = {}
|
|
for msg_name, value in inputs.items():
|
|
spec = self.requires.get(msg_name)
|
|
if spec is None:
|
|
continue
|
|
spec.check(value)
|
|
kwargs[spec.port] = value
|
|
return kwargs
|
|
|
|
def _to_messages(self, retval: dict | None) -> dict | None:
|
|
"""Map a function's port-keyed return value onto message names."""
|
|
if not retval:
|
|
return None
|
|
by_port = {s.port: s for s in self.output_ports if s.name}
|
|
outputs = {}
|
|
for key, value in retval.items():
|
|
spec = by_port.get(key) or self.provides.get(key)
|
|
if spec is None:
|
|
continue
|
|
spec.check(value)
|
|
outputs[spec.name] = value
|
|
return outputs or None
|
|
|
|
def execute(self, inputs: dict | None = None) -> dict | None:
|
|
"""Run the node function and return its outputs by message name.
|
|
|
|
Downstream nodes are not triggered — the pipeline schedules those.
|
|
"""
|
|
kwargs = self._to_kwargs(inputs or {})
|
|
return self._to_messages(self.f(**kwargs, params=self.params))
|
|
|
|
def trigger(self, inputs: dict | None = None) -> dict | None:
|
|
"""
|
|
Trigger this node externally, executing downstream nodes if dependencies are met.
|
|
|
|
This method is for nodes that receive data from upstream dependencies.
|
|
For trigger/subscriber nodes that inject data into the pipeline, use :meth:`inject`.
|
|
|
|
:param inputs: Input values matching this node's ``requires``.
|
|
:type inputs: dict | None
|
|
:returns: Result of the node execution and downstream propagation.
|
|
:rtype: dict | None
|
|
:raises RuntimeError: If node is not bound to a pipeline.
|
|
"""
|
|
if self._pipeline is None:
|
|
raise RuntimeError("Node must be bound to a pipeline to trigger")
|
|
return self(inputs)
|
|
|
|
def inject(self, outputs: dict | None = None) -> dict | None:
|
|
"""
|
|
Inject data into the pipeline as if this node produced it.
|
|
|
|
This method is for trigger/subscriber nodes that receive external data
|
|
(e.g., HTTP requests, MQTT messages) and need to inject it into the pipeline.
|
|
The data is validated against this node's ``provides`` specification.
|
|
|
|
For source nodes (nodes with no ``requires``), if no outputs are provided,
|
|
the node's function will be executed to generate outputs.
|
|
|
|
:param outputs: Output values matching this node's ``provides``.
|
|
:type outputs: dict | None
|
|
:returns: Result of downstream propagation.
|
|
:rtype: dict | None
|
|
:raises RuntimeError: If node is not bound to a pipeline.
|
|
:raises TypeError: If output values don't match ``provides`` types.
|
|
:raises KeyError: If required output keys are missing.
|
|
"""
|
|
if self._pipeline is None:
|
|
raise RuntimeError("Node must be bound to a pipeline to inject")
|
|
|
|
outputs = outputs or {}
|
|
|
|
# A source node asked to inject nothing produces its own data.
|
|
if not outputs and not self.requires:
|
|
outputs = self.f(params=self.params) or {}
|
|
|
|
return self._pipeline.trigger(self, self._to_messages(outputs))
|
|
|
|
def __call__(self, inputs: dict | None = None) -> dict | None:
|
|
"""
|
|
Execute the node and trigger downstream nodes if bound to a pipeline.
|
|
|
|
Validates inputs against the node's ``requires`` specification, executes
|
|
the wrapped function, validates outputs, and triggers downstream execution
|
|
if the node is bound to a pipeline.
|
|
|
|
:param inputs: Input values keyed by message name. Must match the node's
|
|
``requires`` specification.
|
|
:type inputs: dict | None
|
|
:returns: Node outputs if successful, or pipeline execution results if bound.
|
|
:rtype: dict | None
|
|
"""
|
|
outputs = self.execute(inputs)
|
|
return self._pipeline.trigger(self, outputs) if self._pipeline else outputs
|
|
|
|
|
|
class MLPNode(Node):
|
|
"""
|
|
Multi-Layer Perceptron node for neural network processing in pipelines.
|
|
|
|
This node implements a simple single-layer neural network that applies
|
|
weights and biases to input values. Weights and biases are randomly
|
|
initialized using the provided random number generator.
|
|
|
|
The computation follows the standard neural network formula:
|
|
output = weights @ inputs + biases
|
|
|
|
:param requires: Input messages consumed by this node.
|
|
:type requires: MessageSpec | list[MessageSpec]
|
|
:param provides: Output messages produced by this node.
|
|
:type provides: MessageSpec | list[MessageSpec]
|
|
:param params: Parameters dict containing:
|
|
- ``rng`` (numpy.random.Generator): Random number generator for weight initialization
|
|
- Additional node parameters
|
|
:type params: dict
|
|
:param name: Name for this node.
|
|
:type name: str
|
|
|
|
:example:
|
|
>>> import numpy as np
|
|
>>> rng = np.random.default_rng(seed=42)
|
|
>>> mlp = MLPNode(
|
|
... requires=[MessageSpec(name="input1", dtype=DType.FLOAT), MessageSpec(name="input2", dtype=DType.FLOAT)],
|
|
... provides=[MessageSpec(name="output", dtype=DType.FLOAT)],
|
|
... params={"rng": rng},
|
|
... name="mlp_layer1",
|
|
... )
|
|
"""
|
|
|
|
class Params(BaseModel):
|
|
"""Weights are drawn from ``seed``, so a node reloads identically."""
|
|
|
|
model_config = ConfigDict(extra="allow")
|
|
|
|
seed: int = 0
|
|
|
|
def __init__(
|
|
self,
|
|
requires: MessageSpec | Iterable[MessageSpec] = (),
|
|
provides: MessageSpec | Iterable[MessageSpec] = (),
|
|
params: dict[str, Any] | None = None,
|
|
name: str | None = None,
|
|
):
|
|
super().__init__(
|
|
self._forward,
|
|
requires=requires,
|
|
provides=provides,
|
|
params=params,
|
|
name=name or "mlp",
|
|
)
|
|
cfg = self.Params.model_validate(self.params)
|
|
|
|
rng = np.random.default_rng(seed=cfg.seed)
|
|
num_inputs = max(1, len(self.input_ports))
|
|
num_outputs = max(1, len(self.output_ports))
|
|
self.weights = rng.normal(loc=1, size=(num_outputs, num_inputs))
|
|
self.biases = rng.normal(loc=0, size=(num_outputs,))
|
|
|
|
def _forward(self, params: dict, **kwargs) -> dict | None:
|
|
"""Apply ``weights @ inputs + biases`` to the incoming values."""
|
|
if not self.output_ports:
|
|
return None
|
|
|
|
logger.info(
|
|
"Executing MLP node in thread %s: %s",
|
|
threading.current_thread().name,
|
|
self.id,
|
|
)
|
|
|
|
if kwargs:
|
|
input_array = np.array([float(v) for v in kwargs.values()])
|
|
else:
|
|
input_array = np.array([1.0]) # Bias only, for source nodes.
|
|
|
|
outputs = np.dot(self.weights, input_array) + self.biases
|
|
return {p.port: float(outputs[i]) for i, p in enumerate(self.output_ports)}
|
|
|
|
|
|
class HttpNode(Node):
|
|
"""
|
|
HTTP node that can act as a trigger (receiver) or sender based on configuration.
|
|
|
|
This node integrates with FastAPI to either:
|
|
|
|
- **Trigger mode**: Receive incoming HTTP requests (GET/POST) and inject data
|
|
into the pipeline. Used when ``provides`` is specified but ``requires`` is empty.
|
|
- **Sender mode**: Make outgoing HTTP requests with pipeline data. Used when
|
|
``requires`` is specified.
|
|
|
|
:param url: The URL endpoint. For trigger mode, this is the route path
|
|
(e.g., "/sensors/temperature"). For sender mode, this is the full URL
|
|
to send requests to.
|
|
:type url: str
|
|
:param method: HTTP method - "GET" or "POST".
|
|
:type method: Literal["GET", "POST"]
|
|
:param requires: Messages required by this node (makes it a sender node).
|
|
:type requires: MessageSpec | list[MessageSpec] | None
|
|
:param provides: Messages provided by this node (makes it a trigger node).
|
|
:type provides: MessageSpec | list[MessageSpec] | None
|
|
:param params: Additional parameters for the node.
|
|
:type params: dict
|
|
:param name: Optional name for the node.
|
|
:type name: str | None
|
|
:param timeout: Request timeout in seconds (for sender mode).
|
|
:type timeout: float
|
|
:param headers: Additional HTTP headers.
|
|
:type headers: dict[str, str] | None
|
|
|
|
:raises ValueError: If both ``requires`` and ``provides`` are empty, or if
|
|
the configuration is invalid.
|
|
|
|
:example:
|
|
Trigger node (receives POST requests):
|
|
|
|
>>> trigger = HttpNode(
|
|
... url="/api/sensors/temperature",
|
|
... method="POST",
|
|
... provides=[MessageSpec(name="temperature", dtype=DType.FLOAT)],
|
|
... params={},
|
|
... )
|
|
|
|
Sender node (makes POST requests):
|
|
|
|
>>> sender = HttpNode(
|
|
... url="https://api.example.com/data",
|
|
... method="POST",
|
|
... requires=[MessageSpec(name="temperature", dtype=DType.FLOAT)],
|
|
... params={},
|
|
... )
|
|
"""
|
|
|
|
class Mode(Enum):
|
|
"""Operating mode of the HTTP node."""
|
|
|
|
TRIGGER = "trigger" # Receives HTTP requests
|
|
SENDER = "sender" # Sends HTTP requests
|
|
|
|
__slots__ = (
|
|
"url",
|
|
"method",
|
|
"mode",
|
|
"timeout",
|
|
"headers",
|
|
"_route_registered",
|
|
"_http_client",
|
|
)
|
|
|
|
class Params(BaseModel):
|
|
model_config = ConfigDict(extra="allow")
|
|
|
|
url: str
|
|
method: Literal["GET", "POST"] = "POST"
|
|
timeout: float = 30.0
|
|
headers: dict[str, str] = {}
|
|
|
|
def __init__(
|
|
self,
|
|
url: str | None = None,
|
|
method: Literal["GET", "POST"] | None = None,
|
|
requires: MessageSpec | Iterable[MessageSpec] = (),
|
|
provides: MessageSpec | Iterable[MessageSpec] = (),
|
|
params: dict[str, Any] | None = None,
|
|
name: str | None = None,
|
|
):
|
|
params = dict(params) if params else {}
|
|
if url is not None:
|
|
params["url"] = url
|
|
if method is not None:
|
|
params["method"] = method
|
|
cfg = self.Params.model_validate(params)
|
|
|
|
requires = Node._normalize_ports(requires)
|
|
provides = Node._normalize_ports(provides)
|
|
if not requires and not provides:
|
|
raise ValueError(
|
|
"An HTTP node needs either inputs (to send) or outputs (to receive)"
|
|
)
|
|
|
|
# Inputs mean this node sends data out; outputs mean it receives.
|
|
self.mode = HttpNode.Mode.SENDER if requires else HttpNode.Mode.TRIGGER
|
|
|
|
self.url = cfg.url
|
|
self.method = cfg.method.upper()
|
|
self.timeout = cfg.timeout
|
|
self.headers = cfg.headers
|
|
self._route_registered = False
|
|
self._http_client: httpx.AsyncClient | None = None
|
|
|
|
# Set default name based on mode and URL
|
|
if name is None:
|
|
safe_url = self.url.replace("/", "_").replace(":", "").strip("_")
|
|
name = f"http_{self.mode.value}_{safe_url}"
|
|
|
|
# Initialize parent with appropriate function
|
|
# For trigger mode, f is a no-op since data is injected via inject()
|
|
# For sender mode, f handles the outgoing HTTP request
|
|
super().__init__(
|
|
f=(
|
|
self._noop_trigger
|
|
if self.mode == HttpNode.Mode.TRIGGER
|
|
else self._sender_handler
|
|
),
|
|
requires=requires,
|
|
provides=provides,
|
|
params=params,
|
|
name=name,
|
|
)
|
|
|
|
@staticmethod
|
|
def _noop_trigger(params: dict, **kwargs) -> dict | None:
|
|
"""
|
|
No-op function for trigger mode nodes.
|
|
|
|
Trigger mode nodes inject data via :meth:`inject`, not :meth:`__call__`.
|
|
This function exists only to satisfy the Node interface.
|
|
"""
|
|
return None
|
|
|
|
def _sender_handler(self, params: dict, **kwargs) -> dict | None:
|
|
"""
|
|
Send HTTP request with pipeline data (sender mode).
|
|
|
|
This method is called when upstream dependencies are satisfied.
|
|
It sends the required data via HTTP request.
|
|
|
|
:param params: Node parameters.
|
|
:type params: dict
|
|
:param kwargs: Pipeline data to send (from required messages).
|
|
:type kwargs: Any
|
|
:returns: Response data if the endpoint returns JSON, None otherwise.
|
|
:rtype: dict | None
|
|
"""
|
|
# Run async request in sync context
|
|
try:
|
|
loop = asyncio.get_event_loop()
|
|
if loop.is_running():
|
|
# Already on an event loop, so run the request on its own.
|
|
import concurrent.futures
|
|
|
|
with concurrent.futures.ThreadPoolExecutor() as executor:
|
|
result = executor.submit(
|
|
asyncio.run, self._send_request(kwargs)
|
|
).result()
|
|
return result
|
|
else:
|
|
return loop.run_until_complete(self._send_request(kwargs))
|
|
except RuntimeError:
|
|
# No event loop, create one
|
|
return asyncio.run(self._send_request(kwargs))
|
|
|
|
async def _send_request(self, data: dict) -> dict | None:
|
|
"""
|
|
Send an async HTTP request.
|
|
|
|
:param data: Data to send in the request.
|
|
:type data: dict
|
|
:returns: Response JSON if available, None otherwise.
|
|
:rtype: dict | None
|
|
"""
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
try:
|
|
if self.method == "GET":
|
|
response = await client.get(
|
|
self.url, params=data, headers=self.headers
|
|
)
|
|
else: # POST
|
|
response = await client.post(
|
|
self.url, json=data, headers=self.headers
|
|
)
|
|
|
|
response.raise_for_status()
|
|
|
|
# Try to parse JSON response
|
|
try:
|
|
return response.json()
|
|
except Exception:
|
|
return None
|
|
|
|
except httpx.HTTPStatusError as e:
|
|
logger.error(
|
|
"HTTP error in node '%s': status=%s, url=%s",
|
|
self.name,
|
|
e.response.status_code,
|
|
self.url,
|
|
)
|
|
raise
|
|
except httpx.RequestError as e:
|
|
logger.error("Request error in node '%s': %s", self.name, e)
|
|
raise
|
|
|
|
def register_route(self, app: FastAPI) -> None:
|
|
"""
|
|
Register this node's HTTP endpoint with a FastAPI application.
|
|
|
|
This method should only be called for trigger mode nodes. It creates
|
|
a route that, when called, triggers the node in the pipeline.
|
|
|
|
:param app: The FastAPI application instance.
|
|
:type app: FastAPI
|
|
:raises RuntimeError: If called on a sender mode node.
|
|
|
|
:example:
|
|
>>> from fastapi import FastAPI
|
|
>>> app = FastAPI()
|
|
>>> trigger_node = HttpNode(
|
|
... url="/sensors/data",
|
|
... method="POST",
|
|
... provides=[MessageSpec(name="value", dtype=DType.FLOAT)],
|
|
... params={},
|
|
... )
|
|
>>> trigger_node.register_route(app)
|
|
"""
|
|
if self.mode != HttpNode.Mode.TRIGGER:
|
|
raise RuntimeError("Can only register routes for trigger mode nodes")
|
|
|
|
if self._route_registered:
|
|
return
|
|
|
|
from starlette.requests import Request
|
|
from starlette.responses import JSONResponse
|
|
from starlette.routing import Route
|
|
|
|
async def handle_request(request: Request) -> JSONResponse:
|
|
"""
|
|
Handle incoming HTTP request and trigger the pipeline.
|
|
|
|
:param request: The incoming Starlette request.
|
|
:type request: Request
|
|
:returns: JSON response with trigger result.
|
|
:rtype: JSONResponse
|
|
"""
|
|
try:
|
|
# Parse request data
|
|
if self.method == "GET":
|
|
data = dict(request.query_params)
|
|
else: # POST
|
|
content_type = request.headers.get("content-type", "")
|
|
if "application/json" in content_type:
|
|
data = await request.json()
|
|
elif "application/x-www-form-urlencoded" in content_type:
|
|
form = await request.form()
|
|
data = dict(form)
|
|
else:
|
|
data = await request.json() # Default to JSON
|
|
|
|
# Payload keys are port names; values arrive as text.
|
|
typed_data = {}
|
|
for spec in self.output_ports:
|
|
if spec.port in data:
|
|
typed_data[spec.port] = spec.coerce(data[spec.port])
|
|
|
|
# Inject data into the pipeline (trigger mode nodes inject via provides)
|
|
result = self.inject(typed_data)
|
|
|
|
return JSONResponse(
|
|
content={
|
|
"status": "triggered",
|
|
"node": self.name,
|
|
"data": typed_data,
|
|
"result": result if isinstance(result, dict) else None,
|
|
}
|
|
)
|
|
|
|
except ValueError as e:
|
|
return JSONResponse(
|
|
content={"error": str(e)},
|
|
status_code=400,
|
|
)
|
|
except Exception as e:
|
|
return JSONResponse(
|
|
content={"error": str(e)},
|
|
status_code=500,
|
|
)
|
|
|
|
# Create a Starlette Route and add it directly to the app's routes
|
|
route = Route(
|
|
self.url,
|
|
handle_request,
|
|
methods=[self.method],
|
|
name=self.id,
|
|
)
|
|
app.routes.append(route)
|
|
|
|
self._route_registered = True
|
|
logger.info(
|
|
"Registered %s route for node '%s': %s",
|
|
self.method,
|
|
self.name,
|
|
self.url,
|
|
)
|
|
|
|
def unregister_route(self, app: FastAPI) -> None:
|
|
"""
|
|
Unregister this node's HTTP endpoint from a FastAPI application.
|
|
|
|
:param app: The FastAPI application instance.
|
|
:type app: FastAPI
|
|
|
|
.. note::
|
|
FastAPI doesn't natively support route removal. This method
|
|
removes the route from the internal routes list, but the change
|
|
may not take effect until the application is restarted or
|
|
the OpenAPI schema is regenerated.
|
|
"""
|
|
if not self._route_registered:
|
|
return
|
|
|
|
# FastAPI doesn't have a clean way to remove routes
|
|
# We need to filter them out from the routes list
|
|
app.routes[:] = [
|
|
route
|
|
for route in app.routes
|
|
if not (hasattr(route, "name") and route.name == self.id)
|
|
]
|
|
|
|
self._route_registered = False
|
|
logger.info("Unregistered route for node '%s': %s", self.name, self.url)
|
|
|
|
|
|
class MqttNode(Node):
|
|
"""
|
|
MQTT node that can act as a subscriber (trigger) or publisher (sender).
|
|
|
|
This node integrates with an MQTT broker to either:
|
|
|
|
- **Trigger mode (Subscriber)**: Subscribe to MQTT topics and inject received
|
|
messages into the pipeline. Used when ``provides`` is specified but
|
|
``requires`` is empty.
|
|
- **Sender mode (Publisher)**: Publish pipeline data to MQTT topics. Used when
|
|
``requires`` is specified.
|
|
|
|
The ``topic`` parameter in ``params`` controls the mapping between pipeline
|
|
message names and MQTT topics:
|
|
|
|
- **dict**: Explicit mapping from message name to MQTT topic, e.g.
|
|
``{"temperature": "sensors/room1/temp", "humidity": "sensors/room1/hum"}``.
|
|
- **str** (legacy): A single topic string. All messages are mapped to this
|
|
one topic (subscriber receives from it, publisher sends to it).
|
|
|
|
:param requires: Messages required by this node (makes it a publisher node).
|
|
:type requires: MessageSpec | list[MessageSpec] | None
|
|
:param provides: Messages provided by this node (makes it a subscriber node).
|
|
:type provides: MessageSpec | list[MessageSpec] | None
|
|
:param params: Parameters dict containing:
|
|
- ``topic`` (str | dict): MQTT topic(s). A dict maps message names to
|
|
individual topics. A plain string uses that topic for all messages.
|
|
- ``broker_host`` (str): MQTT broker hostname (default: "localhost")
|
|
- ``broker_port`` (int): MQTT broker port (default: 1883)
|
|
- ``username`` (str | None): Optional username for authentication
|
|
- ``password`` (str | None): Optional password for authentication
|
|
- ``client_id`` (str | None): Optional client ID
|
|
- ``qos`` (int): Quality of Service level 0, 1, or 2 (default: 0)
|
|
- ``retain`` (bool): Retain flag for published messages (default: False)
|
|
- ``keepalive`` (int): Keepalive interval in seconds (default: 60)
|
|
:type params: dict
|
|
:param name: Optional name for the node.
|
|
:type name: str | None
|
|
|
|
:raises ValueError: If both ``requires`` and ``provides`` are empty.
|
|
|
|
:example:
|
|
Subscriber with per-message topics:
|
|
|
|
>>> subscriber = MqttNode(
|
|
... provides=[
|
|
... MessageSpec(name="inverter_input", dtype=DType.FLOAT),
|
|
... MessageSpec(name="inverter_output", dtype=DType.FLOAT),
|
|
... ],
|
|
... params={
|
|
... "topic": {
|
|
... "inverter_input": "sensors/pv",
|
|
... "inverter_output": "sensors/output",
|
|
... },
|
|
... "broker_host": "localhost",
|
|
... },
|
|
... )
|
|
|
|
Publisher with per-message topics:
|
|
|
|
>>> publisher = MqttNode(
|
|
... requires=[
|
|
... MessageSpec(name="target_temp", dtype=DType.FLOAT),
|
|
... MessageSpec(name="fan_speed", dtype=DType.INT),
|
|
... ],
|
|
... params={
|
|
... "topic": {
|
|
... "target_temp": "actuators/hvac/temp",
|
|
... "fan_speed": "actuators/hvac/fan",
|
|
... },
|
|
... "broker_host": "localhost",
|
|
... "qos": 1,
|
|
... },
|
|
... )
|
|
|
|
Legacy single-topic subscriber:
|
|
|
|
>>> subscriber = MqttNode(
|
|
... provides=[MessageSpec(name="temperature", dtype=DType.FLOAT)],
|
|
... params={"topic": "sensors/temperature", "broker_host": "localhost"},
|
|
... )
|
|
"""
|
|
|
|
class Mode(Enum):
|
|
"""Operating mode of the MQTT node."""
|
|
|
|
SUBSCRIBER = "subscriber" # Receives MQTT messages (trigger)
|
|
PUBLISHER = "publisher" # Sends MQTT messages (sender)
|
|
|
|
__slots__ = (
|
|
"topics",
|
|
"mode",
|
|
"broker_host",
|
|
"broker_port",
|
|
"username",
|
|
"password",
|
|
"client_id",
|
|
"qos",
|
|
"retain",
|
|
"keepalive",
|
|
"_topic_to_ports",
|
|
"_subscription_task",
|
|
"_mqtt_client",
|
|
"_stop_event",
|
|
)
|
|
|
|
class Params(BaseModel):
|
|
model_config = ConfigDict(extra="allow")
|
|
|
|
# One topic for every port, or a per-port mapping.
|
|
topic: str | dict[str, str] = "*"
|
|
broker_host: str = "localhost"
|
|
broker_port: int = 1883
|
|
username: str | None = None
|
|
password: str | None = None
|
|
client_id: str | None = None
|
|
qos: int = 0
|
|
retain: bool = False
|
|
keepalive: int = 60
|
|
|
|
def __init__(
|
|
self,
|
|
requires: MessageSpec | Iterable[MessageSpec] = (),
|
|
provides: MessageSpec | Iterable[MessageSpec] = (),
|
|
params: dict[str, Any] | None = None,
|
|
name: str | None = None,
|
|
):
|
|
cfg = self.Params.model_validate(params or {})
|
|
|
|
requires = Node._normalize_ports(requires)
|
|
provides = Node._normalize_ports(provides)
|
|
if not requires and not provides:
|
|
raise ValueError(
|
|
"An MQTT node needs either inputs (to publish) or outputs "
|
|
"(to subscribe)"
|
|
)
|
|
|
|
# Inputs mean this node publishes; outputs mean it subscribes.
|
|
self.mode = MqttNode.Mode.PUBLISHER if requires else MqttNode.Mode.SUBSCRIBER
|
|
|
|
ports = provides if self.mode == MqttNode.Mode.SUBSCRIBER else requires
|
|
if isinstance(cfg.topic, dict):
|
|
self.topics: dict[str, str] = dict(cfg.topic)
|
|
else:
|
|
self.topics = {spec.port: cfg.topic for spec in ports}
|
|
|
|
# Reverse lookup for routing incoming payloads back to ports.
|
|
self._topic_to_ports: dict[str, list[str]] = {}
|
|
for port, topic in self.topics.items():
|
|
self._topic_to_ports.setdefault(topic, []).append(port)
|
|
|
|
self.broker_host = cfg.broker_host
|
|
self.broker_port = cfg.broker_port
|
|
self.username = cfg.username
|
|
self.password = cfg.password
|
|
self.client_id = cfg.client_id
|
|
self.qos = cfg.qos
|
|
self.retain = cfg.retain
|
|
self.keepalive = cfg.keepalive
|
|
|
|
# Runtime state
|
|
self._subscription_task: asyncio.Task | None = None
|
|
self._mqtt_client = None
|
|
self._stop_event: asyncio.Event | None = None
|
|
|
|
# Set default name based on mode and topics
|
|
if name is None:
|
|
unique_topics = set(self.topics.values())
|
|
if len(unique_topics) == 1:
|
|
safe_topic = (
|
|
next(iter(unique_topics))
|
|
.replace("/", "_")
|
|
.replace("+", "x")
|
|
.replace("#", "all")
|
|
.strip("_")
|
|
)
|
|
else:
|
|
safe_topic = f"{len(unique_topics)}topics"
|
|
name = f"mqtt_{self.mode.value}_{safe_topic}"
|
|
|
|
# Initialize parent with appropriate function
|
|
# For subscriber mode, f is a no-op since data is injected via inject()
|
|
# For publisher mode, f handles the outgoing MQTT publish
|
|
super().__init__(
|
|
f=(
|
|
self._noop_subscriber
|
|
if self.mode == MqttNode.Mode.SUBSCRIBER
|
|
else self._publisher_handler
|
|
),
|
|
requires=requires,
|
|
provides=provides,
|
|
params=params,
|
|
name=name,
|
|
)
|
|
|
|
@staticmethod
|
|
def _noop_subscriber(params: dict, **kwargs) -> dict | None:
|
|
"""
|
|
No-op function for subscriber mode nodes.
|
|
|
|
Subscriber mode nodes inject data via :meth:`inject`, not :meth:`__call__`.
|
|
This function exists only to satisfy the Node interface and should not
|
|
be called directly.
|
|
|
|
:param params: Node parameters (unused).
|
|
:type params: dict
|
|
:param kwargs: Additional arguments (unused).
|
|
:type kwargs: Any
|
|
:returns: Always returns None.
|
|
:rtype: None
|
|
"""
|
|
return None
|
|
|
|
def _publisher_handler(self, params: dict, **kwargs) -> dict | None:
|
|
"""
|
|
Publish pipeline data to MQTT topic (publisher mode).
|
|
|
|
This method is called when upstream dependencies are satisfied.
|
|
It publishes the required data to the configured MQTT topic.
|
|
|
|
:param params: Node parameters.
|
|
:type params: dict
|
|
:param kwargs: Pipeline data to publish (from required messages).
|
|
:type kwargs: Any
|
|
:returns: None (publishing is fire-and-forget).
|
|
:rtype: dict | None
|
|
"""
|
|
|
|
# Run async publish in sync context
|
|
try:
|
|
loop = asyncio.get_event_loop()
|
|
if loop.is_running():
|
|
# Use thread to run async code when already in async context
|
|
import concurrent.futures
|
|
|
|
with concurrent.futures.ThreadPoolExecutor() as executor:
|
|
executor.submit(asyncio.run, self._publish_message(kwargs)).result()
|
|
else:
|
|
loop.run_until_complete(self._publish_message(kwargs))
|
|
except RuntimeError:
|
|
# No event loop, create one
|
|
asyncio.run(self._publish_message(kwargs))
|
|
|
|
return None
|
|
|
|
async def _publish_message(self, data: dict) -> None:
|
|
"""
|
|
Publish messages to their mapped MQTT topics.
|
|
|
|
Each message in *data* is published to its corresponding topic
|
|
from the ``topics`` mapping. Messages are sent as individual
|
|
JSON payloads per topic.
|
|
|
|
:param data: Data to publish, keyed by port name.
|
|
:type data: dict
|
|
"""
|
|
import json
|
|
|
|
import aiomqtt
|
|
|
|
try:
|
|
async with aiomqtt.Client(
|
|
hostname=self.broker_host,
|
|
port=self.broker_port,
|
|
username=self.username,
|
|
password=self.password,
|
|
identifier=self.client_id,
|
|
keepalive=self.keepalive,
|
|
) as client:
|
|
for port, value in data.items():
|
|
topic = self.topics.get(port)
|
|
if topic is None:
|
|
logger.warning(
|
|
"No topic mapping for port '%s' in node '%s', skipping",
|
|
port,
|
|
self.name,
|
|
)
|
|
continue
|
|
|
|
payload = json.dumps(value)
|
|
await client.publish(
|
|
topic,
|
|
payload=payload,
|
|
qos=self.qos,
|
|
retain=self.retain,
|
|
)
|
|
logger.info(
|
|
"Published to '%s' from node '%s': %s",
|
|
topic,
|
|
self.name,
|
|
payload,
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error("MQTT publish error in node '%s': %s", self.name, e)
|
|
raise
|
|
|
|
async def start_subscription(self) -> None:
|
|
"""
|
|
Start the MQTT subscription for trigger mode nodes.
|
|
|
|
This method starts a background task that listens for messages
|
|
on the subscribed topic and triggers the pipeline when messages arrive.
|
|
|
|
:raises RuntimeError: If called on a publisher mode node.
|
|
|
|
:example:
|
|
>>> subscriber = MqttNode(
|
|
... topic="sensors/#",
|
|
... provides=[MessageSpec(name="value", dtype=DType.FLOAT)],
|
|
... params={"broker_host": "localhost"},
|
|
... )
|
|
>>> await subscriber.start_subscription()
|
|
"""
|
|
if self.mode != MqttNode.Mode.SUBSCRIBER:
|
|
raise RuntimeError("Can only start subscription for subscriber mode nodes")
|
|
|
|
if self._subscription_task is not None:
|
|
return # Already running
|
|
|
|
self._stop_event = asyncio.Event()
|
|
self._subscription_task = asyncio.create_task(self._subscription_loop())
|
|
logger.info(
|
|
"Started MQTT subscription for node '%s' to topics %s",
|
|
self.name,
|
|
list(self._topic_to_ports.keys()),
|
|
)
|
|
|
|
async def stop_subscription(self) -> None:
|
|
"""
|
|
Stop the MQTT subscription.
|
|
|
|
Gracefully stops the background subscription task.
|
|
"""
|
|
if self._subscription_task is None:
|
|
return
|
|
|
|
if self._stop_event:
|
|
self._stop_event.set()
|
|
|
|
self._subscription_task.cancel()
|
|
try:
|
|
await self._subscription_task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
|
|
self._subscription_task = None
|
|
self._stop_event = None
|
|
logger.info(
|
|
"Stopped MQTT subscription for node '%s'",
|
|
self.name,
|
|
)
|
|
|
|
async def _subscription_loop(self) -> None:
|
|
"""
|
|
Background loop that listens for MQTT messages and triggers the pipeline.
|
|
|
|
Subscribes to all unique topics from the ``topics`` mapping and
|
|
uses the reverse lookup ``_topic_to_ports`` to route incoming
|
|
payloads to the correct pipeline message names.
|
|
"""
|
|
import json
|
|
|
|
import aiomqtt
|
|
|
|
while not (self._stop_event and self._stop_event.is_set()):
|
|
try:
|
|
async with aiomqtt.Client(
|
|
hostname=self.broker_host,
|
|
port=self.broker_port,
|
|
username=self.username,
|
|
password=self.password,
|
|
identifier=self.client_id,
|
|
keepalive=self.keepalive,
|
|
) as client:
|
|
# Subscribe to every unique topic
|
|
for topic in self._topic_to_ports:
|
|
await client.subscribe(topic, qos=self.qos)
|
|
logger.info("[%s] Subscribed to %s", self.name, topic)
|
|
|
|
async for message in client.messages:
|
|
if self._stop_event and self._stop_event.is_set():
|
|
break
|
|
|
|
try:
|
|
payload = message.payload.decode("utf-8")
|
|
incoming_topic = str(message.topic)
|
|
|
|
logger.info(
|
|
"[%s] Received on %s: %s",
|
|
self.name,
|
|
incoming_topic,
|
|
payload,
|
|
)
|
|
|
|
# Find which port(s) this topic feeds
|
|
ports = self._topic_to_ports.get(incoming_topic, [])
|
|
if not ports:
|
|
logger.debug(
|
|
"[%s] No mapping for topic '%s', ignoring",
|
|
self.name,
|
|
incoming_topic,
|
|
)
|
|
continue
|
|
|
|
# Parse the payload value
|
|
try:
|
|
parsed = json.loads(payload)
|
|
except json.JSONDecodeError:
|
|
parsed = payload
|
|
|
|
by_port = {s.port: s for s in self.output_ports}
|
|
typed_data = {}
|
|
for port in ports:
|
|
spec = by_port.get(port)
|
|
if spec is None:
|
|
continue
|
|
|
|
# A JSON object may carry the port as a key;
|
|
# anything else is the value itself.
|
|
if isinstance(parsed, dict) and port in parsed:
|
|
value = parsed[port]
|
|
else:
|
|
value = parsed
|
|
|
|
typed_data[port] = spec.coerce(value)
|
|
|
|
if typed_data:
|
|
await asyncio.to_thread(self.inject, typed_data)
|
|
|
|
except Exception as e:
|
|
logger.error(
|
|
"[%s] Error processing message: %s",
|
|
self.name,
|
|
e,
|
|
exc_info=True,
|
|
)
|
|
|
|
except aiomqtt.MqttError as e:
|
|
logger.warning(
|
|
"MQTT connection error in node '%s': %s. Retrying in 5s...",
|
|
self.name,
|
|
e,
|
|
)
|
|
if not (self._stop_event and self._stop_event.is_set()):
|
|
# Reconnect after a delay
|
|
await asyncio.sleep(5)
|
|
except asyncio.CancelledError:
|
|
break
|
|
except Exception as e:
|
|
logger.error(
|
|
"Unexpected error in MQTT subscription for node '%s': %s. Retrying in 5s...",
|
|
self.name,
|
|
e,
|
|
exc_info=True,
|
|
)
|
|
if not (self._stop_event and self._stop_event.is_set()):
|
|
await asyncio.sleep(5)
|
|
|
|
@property
|
|
def is_subscribed(self) -> bool:
|
|
"""
|
|
Check if the subscription is currently active.
|
|
|
|
:returns: True if subscription task is running.
|
|
:rtype: bool
|
|
"""
|
|
return (
|
|
self._subscription_task is not None and not self._subscription_task.done()
|
|
)
|
|
|
|
|
|
class InfluxDbNode(Node):
|
|
"""
|
|
InfluxDB node for writing to and reading from InfluxDB.
|
|
|
|
This node can perform both write and read operations independently:
|
|
|
|
- **Write operation**: Triggered when upstream dependencies are satisfied
|
|
(data flows in via ``requires``). Writes data points to InfluxDB based
|
|
on the ``writes`` configuration in params.
|
|
- **Read operation**: Performed when the node provides data to downstream
|
|
nodes via ``provides``, based on the ``queries`` configuration in params.
|
|
|
|
Both operations use a similar configuration pattern in params, making the
|
|
API consistent and the input/output data simple (just values).
|
|
|
|
:param requires: Messages to write to InfluxDB. The actual value from each
|
|
message is written according to the corresponding config in ``writes``.
|
|
:type requires: MessageSpec | list[MessageSpec] | None
|
|
:param provides: Messages to read from InfluxDB. Each message gets its value
|
|
from a query defined in ``queries``.
|
|
:type provides: MessageSpec | list[MessageSpec] | None
|
|
:param params: Parameters dict containing:
|
|
- ``url`` (str): InfluxDB server URL (required)
|
|
- ``token`` (str): Authentication token (required)
|
|
- ``org`` (str): Organization name (required)
|
|
- ``bucket`` (str): Bucket name (required)
|
|
- ``write_precision`` (str): Write precision ("ns", "us", "ms", "s"), default "ms"
|
|
- ``query_range`` (str): Default time range for queries, e.g., "-1h", "-24h"
|
|
- ``writes`` (dict): Write configurations keyed by message name, each with:
|
|
- ``measurement`` (str): Measurement name to write to
|
|
- ``field`` (str): Field name to write (default: "value")
|
|
- ``tags`` (dict): Static tags to add to each point
|
|
- ``queries`` (dict): Query configurations keyed by message name, each with:
|
|
- ``measurement`` (str): Measurement name to query
|
|
- ``field`` (str): Field name to retrieve (default: "value")
|
|
- ``tags`` (dict): Optional tag filters
|
|
- ``range`` (str): Optional time range override
|
|
- ``aggregation`` (str): Aggregation function ("mean", "last", "first", "max", "min")
|
|
:type params: dict
|
|
:param name: Optional name for the node.
|
|
:type name: str | None
|
|
|
|
:raises ValueError: If required params are missing or both requires and provides are empty.
|
|
|
|
:example:
|
|
Write-only node (writes temperature values):
|
|
|
|
>>> writer = InfluxDbNode(
|
|
... requires=[MessageSpec(name="temperature", dtype=DType.FLOAT)],
|
|
... params={
|
|
... "url": "http://localhost:8086",
|
|
... "token": "my-token",
|
|
... "org": "my-org",
|
|
... "bucket": "sensors",
|
|
... "writes": {
|
|
... "temperature": {
|
|
... "measurement": "environment",
|
|
... "field": "temp_celsius",
|
|
... "tags": {"location": "room1", "sensor": "dht22"},
|
|
... }
|
|
... },
|
|
... },
|
|
... )
|
|
|
|
Read-only node (queries average temperature):
|
|
|
|
>>> reader = InfluxDbNode(
|
|
... provides=[MessageSpec(name="avg_temperature", dtype=DType.FLOAT)],
|
|
... params={
|
|
... "url": "http://localhost:8086",
|
|
... "token": "my-token",
|
|
... "org": "my-org",
|
|
... "bucket": "sensors",
|
|
... "queries": {
|
|
... "avg_temperature": {
|
|
... "measurement": "environment",
|
|
... "field": "temp_celsius",
|
|
... "tags": {"location": "room1"},
|
|
... "range": "-1h",
|
|
... "aggregation": "mean",
|
|
... }
|
|
... },
|
|
... },
|
|
... )
|
|
|
|
Combined read/write node:
|
|
|
|
>>> node = InfluxDbNode(
|
|
... requires=[MessageSpec(name="raw_temp", dtype=DType.FLOAT)],
|
|
... provides=[MessageSpec(name="avg_temp", dtype=DType.FLOAT)],
|
|
... params={
|
|
... "url": "http://localhost:8086",
|
|
... "token": "my-token",
|
|
... "org": "my-org",
|
|
... "bucket": "sensors",
|
|
... "writes": {
|
|
... "raw_temp": {
|
|
... "measurement": "temperature",
|
|
... "field": "value",
|
|
... "tags": {"source": "sensor"},
|
|
... }
|
|
... },
|
|
... "queries": {
|
|
... "avg_temp": {
|
|
... "measurement": "temperature",
|
|
... "field": "value",
|
|
... "aggregation": "mean",
|
|
... "range": "-5m",
|
|
... }
|
|
... },
|
|
... },
|
|
... )
|
|
"""
|
|
|
|
__slots__ = (
|
|
"url",
|
|
"token",
|
|
"org",
|
|
"bucket",
|
|
"write_precision",
|
|
"query_range",
|
|
"writes",
|
|
"queries",
|
|
"_write_client",
|
|
"_query_client",
|
|
)
|
|
|
|
class Params(BaseModel):
|
|
model_config = ConfigDict(extra="allow")
|
|
|
|
url: str
|
|
token: str
|
|
org: str
|
|
bucket: str
|
|
write_precision: str = "ms"
|
|
query_range: str = "-1h"
|
|
# Per-port write and query configuration.
|
|
writes: dict[str, dict[str, Any]] = {}
|
|
queries: dict[str, dict[str, Any]] = {}
|
|
|
|
def __init__(
|
|
self,
|
|
requires: MessageSpec | Iterable[MessageSpec] = (),
|
|
provides: MessageSpec | Iterable[MessageSpec] = (),
|
|
params: dict[str, Any] | None = None,
|
|
name: str | None = None,
|
|
):
|
|
cfg = self.Params.model_validate(params or {})
|
|
|
|
requires = Node._normalize_ports(requires)
|
|
provides = Node._normalize_ports(provides)
|
|
if not requires and not provides:
|
|
raise ValueError(
|
|
"An InfluxDB node needs either inputs (to write) or outputs (to read)"
|
|
)
|
|
|
|
self.url = cfg.url
|
|
self.token = cfg.token
|
|
self.org = cfg.org
|
|
self.bucket = cfg.bucket
|
|
self.write_precision = cfg.write_precision
|
|
self.query_range = cfg.query_range
|
|
self.writes = cfg.writes
|
|
self.queries = cfg.queries
|
|
|
|
# Lazy-initialized clients
|
|
self._write_client = None
|
|
self._query_client = None
|
|
|
|
# Set default name
|
|
if name is None:
|
|
name = f"influxdb_{self.bucket}"
|
|
|
|
# Initialize parent
|
|
# The handler function depends on what operations are configured
|
|
super().__init__(
|
|
f=self._handler,
|
|
requires=requires,
|
|
provides=provides,
|
|
params=params,
|
|
name=name,
|
|
)
|
|
|
|
def _handler(self, params: dict, **kwargs) -> dict | None:
|
|
"""
|
|
Handle incoming data - write to InfluxDB and optionally query.
|
|
|
|
This method is called when upstream dependencies (requires) are satisfied.
|
|
It writes the incoming data to InfluxDB and can also perform reads.
|
|
|
|
:param params: Node parameters.
|
|
:type params: dict
|
|
:param kwargs: Incoming data from upstream nodes.
|
|
:type kwargs: Any
|
|
:returns: Query results if provides is configured, None otherwise.
|
|
:rtype: dict | None
|
|
"""
|
|
# Write incoming data
|
|
if kwargs:
|
|
self._write_points(kwargs)
|
|
|
|
# If we have provides, perform queries
|
|
if self.provides:
|
|
return self._query_data()
|
|
|
|
return None
|
|
|
|
def _write_points(self, data: dict) -> None:
|
|
"""
|
|
Write data points to InfluxDB using configuration from ``writes``.
|
|
|
|
The write configuration is looked up in ``self.writes`` by message name.
|
|
Each config specifies the measurement, field, and tags. The actual value
|
|
comes from the incoming data.
|
|
|
|
:param data: Data to write, keyed by port name. Values can be:
|
|
- Simple values (float, int, str, bool): Written using config from ``writes``
|
|
- List of values: Each value written as a separate point
|
|
- Dict with "value" key: Value extracted and written using config
|
|
- Dict with "value" and "tags" keys: Value written with merged tags
|
|
:type data: dict
|
|
"""
|
|
from influxdb_client import InfluxDBClient, Point, WritePrecision
|
|
from influxdb_client.client.write_api import SYNCHRONOUS
|
|
|
|
try:
|
|
with InfluxDBClient(url=self.url, token=self.token, org=self.org) as client:
|
|
write_api = client.write_api(write_options=SYNCHRONOUS)
|
|
|
|
precision_map = {
|
|
"ns": WritePrecision.NS,
|
|
"us": WritePrecision.US,
|
|
"ms": WritePrecision.MS,
|
|
"s": WritePrecision.S,
|
|
}
|
|
precision = precision_map.get(self.write_precision, WritePrecision.MS)
|
|
|
|
for msg_name, msg_value in data.items():
|
|
# Get write configuration for this message
|
|
write_config = self.writes.get(msg_name, {})
|
|
|
|
# Get measurement, field, and base tags from config
|
|
measurement = write_config.get("measurement", msg_name)
|
|
field = write_config.get("field", "value")
|
|
base_tags = write_config.get("tags", {})
|
|
|
|
# Handle list of values (batch write)
|
|
values_to_write = (
|
|
msg_value if isinstance(msg_value, list) else [msg_value]
|
|
)
|
|
|
|
for item in values_to_write:
|
|
# Extract value and optional runtime tags
|
|
if isinstance(item, dict):
|
|
value = item.get("value", item)
|
|
runtime_tags = item.get("tags", {})
|
|
# If no "value" key, treat the whole dict as invalid
|
|
if "value" not in item and not isinstance(
|
|
value, (int, float, str, bool)
|
|
):
|
|
logger.warning(
|
|
"Skipping invalid item in node '%s': %s",
|
|
self.name,
|
|
item,
|
|
)
|
|
continue
|
|
else:
|
|
value = item
|
|
runtime_tags = {}
|
|
|
|
if value is None:
|
|
logger.info(
|
|
"Skipping None value for '%s' in node '%s'",
|
|
msg_name,
|
|
self.name,
|
|
)
|
|
continue
|
|
|
|
# Merge base tags with runtime tags (runtime takes precedence)
|
|
tags = {**base_tags, **runtime_tags}
|
|
|
|
# Build the point
|
|
point = Point(measurement)
|
|
|
|
for tag_key, tag_value in tags.items():
|
|
point = point.tag(tag_key, str(tag_value))
|
|
|
|
point = point.field(field, value)
|
|
|
|
# Write the point
|
|
write_api.write(
|
|
bucket=self.bucket,
|
|
org=self.org,
|
|
record=point,
|
|
write_precision=precision,
|
|
)
|
|
|
|
logger.info(
|
|
"Wrote to InfluxDB from node '%s': %s.%s=%s, tags=%s",
|
|
self.name,
|
|
measurement,
|
|
field,
|
|
value,
|
|
tags,
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error("InfluxDB write error in node '%s': %s", self.name, e)
|
|
raise
|
|
|
|
def _query_data(self) -> dict:
|
|
"""
|
|
Query data from InfluxDB based on provides configuration.
|
|
|
|
:returns: Dict of port name to queried value.
|
|
:rtype: dict
|
|
"""
|
|
from influxdb_client import InfluxDBClient
|
|
|
|
results = {}
|
|
|
|
try:
|
|
with InfluxDBClient(url=self.url, token=self.token, org=self.org) as client:
|
|
query_api = client.query_api()
|
|
|
|
for spec in self.output_ports:
|
|
msg_name = spec.port
|
|
query_config = self.queries.get(msg_name, {})
|
|
|
|
measurement = query_config.get("measurement", msg_name)
|
|
field = query_config.get("field", "value")
|
|
tags = query_config.get("tags", {})
|
|
time_range = query_config.get("range", self.query_range)
|
|
aggregation = query_config.get("aggregation", "last")
|
|
|
|
# Build Flux query
|
|
flux_query = self._build_flux_query(
|
|
measurement=measurement,
|
|
field=field,
|
|
tags=tags,
|
|
time_range=time_range,
|
|
aggregation=aggregation,
|
|
)
|
|
|
|
logger.info(
|
|
"Executing InfluxDB query for '%s' in node '%s': %s",
|
|
msg_name,
|
|
self.name,
|
|
flux_query,
|
|
)
|
|
|
|
# Execute query
|
|
tables = query_api.query(flux_query, org=self.org)
|
|
|
|
# Extract result
|
|
value = self._extract_query_result(tables, spec)
|
|
|
|
if value is not None:
|
|
results[msg_name] = value
|
|
logger.info(
|
|
"Query result for '%s' in node '%s': %s",
|
|
msg_name,
|
|
self.name,
|
|
value,
|
|
)
|
|
else:
|
|
logger.info(
|
|
"No data found for '%s' in node '%s'",
|
|
msg_name,
|
|
self.name,
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error("InfluxDB query error in node '%s': %s", self.name, e)
|
|
raise
|
|
|
|
return results
|
|
|
|
def _build_flux_query(
|
|
self,
|
|
measurement: str,
|
|
field: str,
|
|
tags: dict,
|
|
time_range: str,
|
|
aggregation: str,
|
|
) -> str:
|
|
"""
|
|
Build a Flux query string.
|
|
|
|
:param measurement: Measurement name.
|
|
:type measurement: str
|
|
:param field: Field name.
|
|
:type field: str
|
|
:param tags: Tag filters.
|
|
:type tags: dict
|
|
:param time_range: Time range (e.g., "-1h").
|
|
:type time_range: str
|
|
:param aggregation: Aggregation function.
|
|
:type aggregation: str
|
|
:returns: Flux query string.
|
|
:rtype: str
|
|
"""
|
|
# Base query
|
|
query_parts = [
|
|
f'from(bucket: "{self.bucket}")',
|
|
f" |> range(start: {time_range})",
|
|
f' |> filter(fn: (r) => r["_measurement"] == "{measurement}")',
|
|
f' |> filter(fn: (r) => r["_field"] == "{field}")',
|
|
]
|
|
|
|
# Add tag filters
|
|
for tag_key, tag_value in tags.items():
|
|
query_parts.append(
|
|
f' |> filter(fn: (r) => r["{tag_key}"] == "{tag_value}")'
|
|
)
|
|
|
|
# Add aggregation
|
|
aggregation_map = {
|
|
"mean": "mean()",
|
|
"last": "last()",
|
|
"first": "first()",
|
|
"max": "max()",
|
|
"min": "min()",
|
|
"sum": "sum()",
|
|
"count": "count()",
|
|
}
|
|
|
|
if aggregation in aggregation_map:
|
|
query_parts.append(f" |> {aggregation_map[aggregation]}")
|
|
else:
|
|
# Default to last value
|
|
query_parts.append(" |> last()")
|
|
|
|
return "\n".join(query_parts)
|
|
|
|
def _extract_query_result(self, tables, spec: MessageSpec) -> Any:
|
|
"""
|
|
Extract a single value from query result tables.
|
|
|
|
:param tables: InfluxDB query result tables.
|
|
:param spec: The port the value is destined for.
|
|
:type spec: MessageSpec
|
|
:returns: Extracted and typed value, or None if no data.
|
|
:rtype: Any
|
|
"""
|
|
for table in tables:
|
|
for record in table.records:
|
|
value = record.get_value()
|
|
if value is not None:
|
|
try:
|
|
return spec.coerce(value)
|
|
except (ValueError, TypeError):
|
|
return value
|
|
|
|
return None
|
|
|
|
def inject(self, outputs: dict | None = None) -> dict | None:
|
|
"""
|
|
Inject queried data into the pipeline.
|
|
|
|
For InfluxDbNode, inject performs a query operation and injects
|
|
the results into the pipeline. This is useful for trigger-style
|
|
usage where you want to periodically query InfluxDB.
|
|
|
|
:param outputs: Optional pre-set outputs (usually None for queries).
|
|
:type outputs: dict | None
|
|
:returns: Query results injected into the pipeline.
|
|
:rtype: dict | None
|
|
"""
|
|
if self._pipeline is None:
|
|
raise RuntimeError("Node must be bound to a pipeline to inject")
|
|
|
|
# Without given values, injecting means running the configured queries.
|
|
if not outputs:
|
|
if not self.provides:
|
|
return None
|
|
outputs = self._query_data()
|
|
|
|
return self._pipeline.trigger(self, self._to_messages(outputs))
|
|
|
|
|
|
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")
|
|
|
|
delay: int = 0
|
|
interval: int = 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):
|
|
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 = None
|
|
self._stop_cron = 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, **kwargs):
|
|
"""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
|
|
|
|
# Apply fixed delay
|
|
if self.delay > 0:
|
|
time.sleep(self.delay)
|
|
|
|
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
|
|
}
|
|
|
|
logger.info("[%s] Sending %s", self.name, output)
|
|
return output
|
|
|
|
# -----------------------------------------------------------------
|
|
# Cron scheduler
|
|
# -----------------------------------------------------------------
|
|
|
|
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 = asyncio.create_task(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._cron_task is None:
|
|
return
|
|
|
|
if self._stop_cron:
|
|
self._stop_cron.set()
|
|
|
|
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):
|
|
"""Background loop that sleeps until the next cron tick and triggers."""
|
|
from datetime import datetime
|
|
|
|
try:
|
|
from croniter import croniter
|
|
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())
|
|
|
|
while not (self._stop_cron and self._stop_cron.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(self._stop_cron.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:
|
|
break
|
|
except Exception as e:
|
|
logger.error(
|
|
"Error in cron loop for node '%s': %s",
|
|
self.name,
|
|
e,
|
|
exc_info=True,
|
|
)
|
|
# Back off on error to avoid tight loops
|
|
await asyncio.sleep(60)
|
|
|
|
async def _trigger_cron(self):
|
|
"""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,
|
|
)
|