stroblmeandClaude Opus 5
640654bd66
A wheel whose top-level module is `app` collides with anything else in a user's venv, so the package that is about to be published takes the name it is published under. Only the Python package moves; the repo, the Docker WORKDIR and the compose project keep theirs. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
170 lines
6.6 KiB
Python
170 lines
6.6 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.
|
|
|
|
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
|
|
|
|
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")
|
|
|
|
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] = {}
|
|
|
|
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
|
|
# -------------------------------------------------------------------------
|
|
|
|
async def start(self, app: FastAPI | None = None) -> 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:
|
|
self._poll_task.cancel()
|
|
try:
|
|
await self._poll_task
|
|
except (asyncio.CancelledError, Exception): # noqa: B014 - shutting down
|
|
pass
|
|
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.
|
|
"""
|
|
while not (self._stop_event and self._stop_event.is_set()):
|
|
try:
|
|
values = await self.poll()
|
|
self.report_health("ok")
|
|
changed = {
|
|
port: value
|
|
for port, value in (values or {}).items()
|
|
if self._last_published.get(port, object()) != value
|
|
}
|
|
if changed:
|
|
self._last_published.update(changed)
|
|
# inject runs the graph, which is blocking work.
|
|
await asyncio.to_thread(self.inject, changed)
|
|
except asyncio.CancelledError:
|
|
break
|
|
except Exception as exc:
|
|
logger.warning("Connector '%s' failed to poll: %s", self.id, exc)
|
|
self.report_health("down", f"{type(exc).__name__}: {exc}")
|
|
await asyncio.sleep(self.config.poll_interval)
|