stroblmeandClaude Opus 5
70e542ec3c
The poll loop remembered what it read rather than what it published, so a value the node could not publish counted as said: the next poll skipped it, succeeded, and health went back to ok with the port still dark. Remember it only after inject returns, and report ok last. A node reporting itself down is now derived into its flow's issues on read and counted on the health summary, so the canvas marks it and Home says so. Being down does not stop the flow, and the issue clears by itself when the node reports well again. The repeating poll warning is logged once per outage rather than once per tick. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01K1moruzue2kTJd3uVisgNk
208 lines
8.0 KiB
Python
208 lines
8.0 KiB
Python
"""The contract a connector node is written against.
|
|
|
|
A connector is the device-facing node class: it talks to something outside the
|
|
engine — a device, a service, a protocol — and publishes what it finds as
|
|
ordinary typed messages. Third parties write these, so this surface is the one
|
|
part of the engine that has to stay stable; it is versioned by
|
|
:data:`CONTRACT_VERSION` and a connector declares which version it was written
|
|
for.
|
|
|
|
What a connector gets from the base class:
|
|
|
|
* a polling loop that runs :meth:`ConnectorNode.poll` on a schedule, publishes
|
|
only the ports whose value changed, and reports health around it;
|
|
* :meth:`Node.report_health`, so a connection problem shows on the node rather
|
|
than only in the log;
|
|
* the lifecycle hooks the controller drives, so nothing device-specific has to
|
|
be known by the engine;
|
|
* :meth:`ConnectorNode.write`, the other direction — values arriving on the
|
|
node's input ports, for a connector that commands something rather than only
|
|
reading it;
|
|
* :meth:`ConnectorNode.save_artifact`, for a device whose readings are bytes —
|
|
a camera frame, a recorded clip — which travel as a reference on a media
|
|
port rather than as the message itself.
|
|
|
|
The message schemas and the parameter model are the rest of the contract, and
|
|
they are the same ones the built-in nodes use. See ``docs/connectors/`` for the
|
|
authoring guide.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from typing import TYPE_CHECKING, Any, ClassVar
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
from fluksio.flow.nodes import Node
|
|
|
|
if TYPE_CHECKING:
|
|
from fastapi import FastAPI
|
|
|
|
from fluksio.flow.artifacts import ArtifactStore
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
#: Bumped when a change would break connectors written against the old surface.
|
|
#: The loader refuses a connector declaring anything else.
|
|
CONTRACT_VERSION = 1
|
|
|
|
|
|
class ConnectorNode(Node):
|
|
"""Base class for device- and service-facing nodes.
|
|
|
|
A subclass declares the contract version it was written for, describes
|
|
itself for the editor, and implements :meth:`poll`, :meth:`start`, or both::
|
|
|
|
class RandomSensor(ConnectorNode):
|
|
contract = CONTRACT_VERSION
|
|
title = "Random sensor"
|
|
description = "Emits a random reading, for trying the contract out."
|
|
|
|
class Params(ConnectorNode.Params):
|
|
ceiling: float = 1.0
|
|
|
|
async def poll(self):
|
|
return {"reading": random.random() * self.config.ceiling}
|
|
"""
|
|
|
|
#: Declared explicitly by every connector; inheriting it does not count.
|
|
contract: ClassVar[int]
|
|
title: ClassVar[str] = ""
|
|
description: ClassVar[str] = ""
|
|
|
|
class Params(BaseModel):
|
|
"""Settings the editor renders a form for.
|
|
|
|
Subclass it to add your own. A field holding a credential should carry
|
|
``json_schema_extra={"x-secret": True}``, which makes the editor offer
|
|
the stored secrets instead of a text box.
|
|
"""
|
|
|
|
poll_interval: float = Field(
|
|
default=0,
|
|
ge=0,
|
|
description="Seconds between polls; 0 polls never.",
|
|
)
|
|
|
|
__slots__ = (
|
|
"config",
|
|
"_poll_task",
|
|
"_stop_event",
|
|
"_last_published",
|
|
"_artifacts",
|
|
"_down",
|
|
)
|
|
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
super().__init__(f=self._dispatch, **kwargs)
|
|
self.config = type(self).Params(**self.params)
|
|
self._poll_task: asyncio.Task[None] | None = None
|
|
self._stop_event: asyncio.Event | None = None
|
|
self._last_published: dict[str, Any] = {}
|
|
self._artifacts: ArtifactStore | None = None
|
|
self._down = False
|
|
|
|
def _dispatch(self, params: dict[str, Any], **ports: Any) -> dict[str, Any] | None:
|
|
"""The scheduler's entry point. Settings are already on ``self.config``."""
|
|
return self.write(**ports)
|
|
|
|
# -------------------------------------------------------------------------
|
|
# What a connector implements
|
|
# -------------------------------------------------------------------------
|
|
|
|
async def poll(self) -> dict[str, Any] | None:
|
|
"""Read the device once and return values keyed by output port.
|
|
|
|
Return ``None`` when there is nothing new. Raising is reported as a
|
|
health problem and retried on the next tick.
|
|
"""
|
|
return None
|
|
|
|
def write(self, **ports: Any) -> dict[str, Any] | None:
|
|
"""Send incoming values to the device. Values arrive keyed by input port.
|
|
|
|
A connector that only reads leaves this alone — the default discards
|
|
whatever reaches it, which is what a node with no inputs gets anyway.
|
|
Return ``None`` unless the device answers something worth publishing,
|
|
in which case return it keyed by output port like :meth:`poll` does.
|
|
|
|
This runs on the scheduler's thread, so it must not block for long.
|
|
"""
|
|
return None
|
|
|
|
# -------------------------------------------------------------------------
|
|
# What the engine drives
|
|
# -------------------------------------------------------------------------
|
|
|
|
def save_artifact(
|
|
self,
|
|
data: bytes,
|
|
name: str = "",
|
|
media_type: str = "application/octet-stream",
|
|
) -> dict[str, Any]:
|
|
"""Store bytes and return the reference to publish on a media port.
|
|
|
|
A camera frame or a recorded clip is far too big to be a message, so a
|
|
connector publishing one publishes this instead: the bytes go to the
|
|
store and the reference names them, which is what an ``image``,
|
|
``audio`` or ``video`` port carries.
|
|
|
|
Only available once the node has started — the store belongs to the
|
|
engine, and is handed over then.
|
|
"""
|
|
if self._artifacts is None:
|
|
raise RuntimeError(
|
|
"no artifact store: a connector can only save bytes once it has started"
|
|
)
|
|
return self._artifacts.put([data], name=name, media_type=media_type)
|
|
|
|
async def start(self, app: FastAPI | None = None) -> None:
|
|
self._artifacts = getattr(app.state, "artifact_store", None) if app else None
|
|
if self.config.poll_interval > 0 and self._stop_event is None:
|
|
self._stop_event = asyncio.Event()
|
|
self._poll_task = self._run_supervised("poll", self._poll_loop)
|
|
|
|
async def stop(self, app: FastAPI | None = None) -> None:
|
|
if self._stop_event is None:
|
|
return
|
|
self._stop_event.set()
|
|
if self._poll_task is not None:
|
|
await self._cancel_task(self._poll_task)
|
|
self._poll_task = None
|
|
self._stop_event = None
|
|
self._last_published = {}
|
|
|
|
async def _poll_loop(self) -> None:
|
|
"""Poll, publish what changed, and say how the connection is doing.
|
|
|
|
Only changed ports are published: a device polled every few seconds is
|
|
usually saying the same thing, and every publication wakes everything
|
|
downstream of it. What is remembered is what was *published*, not what
|
|
the poll returned — a publication that raised is retried next tick
|
|
rather than counting as said.
|
|
"""
|
|
while not (self._stop_event and self._stop_event.is_set()):
|
|
try:
|
|
values = await self.poll()
|
|
changed = {
|
|
port: value
|
|
for port, value in (values or {}).items()
|
|
if self._last_published.get(port, object()) != value
|
|
}
|
|
if changed:
|
|
# inject runs the graph, which is blocking work.
|
|
await asyncio.to_thread(self.inject, changed)
|
|
self._last_published.update(changed)
|
|
self.report_health("ok")
|
|
self._down = False
|
|
except asyncio.CancelledError:
|
|
break
|
|
except Exception as exc:
|
|
if not self._down:
|
|
logger.warning("Connector '%s' failed to poll: %s", self.id, exc)
|
|
self._down = True
|
|
self.report_health("down", f"{type(exc).__name__}: {exc}")
|
|
await asyncio.sleep(self.config.poll_interval)
|