app/backend/fluksio/flow/connector.py

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