"""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", volatile: bool = False, ) -> 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. ``volatile`` is what a camera publishes with: the bytes go to a ring held in memory rather than to the data volume, are pushed to whatever screen is watching, and last seconds. Use it for a frame; leave it off for a recording somebody asked to keep. 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, volatile=volatile ) 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)