stroblmeandClaude Opus 5
d471614e6a
The rate the media dtypes could carry was one frame every second or two: each
was a file on the data volume, an event on the socket, and a request back for
the bytes. This closes both halves of that, and they are one feature.
`save_artifact(..., volatile=True)` writes to a `VolatileStore` — the same
content-addressed store, in `/dev/shm`, bounded by size with the oldest falling
out (`ARTIFACT_VOLATILE_BYTES`, 48 MB under the container's raised `shm_size`).
Nothing sweeps it: a frame nobody kept is not worth walking the store to find.
`ArtifactStore.path` falls through to it, which is what lets a volatile frame be
an ordinary reference everywhere else — the dtype check, a panel's digest scope,
`load_artifact` in a node, and the widget's own fetch all work on one unchanged.
`adopt` copies one into the store when a run records it, so "returned media is
kept, emitted media is not" stays true.
The bytes then go down the flows websocket as a length-prefixed binary frame,
sent just ahead of the `message_value` naming them, so a tile has the frame when
it hears the value moved. Nothing is pushed unasked: a client names the messages
it is drawing (`{"type":"media","names":[…]}`), a panel's list is intersected
with the scope it already had, and only the newest frame per name in a batch is
sent — a client that fell behind is not handed frames it would draw over. The
tunnel relays text only, so a screen reached through a portal falls back to
fetching, which is why the rate table now has two rows.
Around the edges: the remote worker's fetch cache is bounded at last
(`FLUKSIO_ARTIFACT_CACHE_BYTES`), since content addressing means nothing in it
ever expires and a media stream fills it with chunks nothing asks for twice; a
port carrying an image draws the frame in the node panel rather than only
saying `image/png · frame.png · 1.79kB`; and an edge chip says that much instead
of a line of hash. The media screenshot stops waiting for `networkidle` — a
camera is a socket that never goes quiet, which is the point of it.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01YC4u66vjzW54fnHu5Juhh9
76 lines
2.7 KiB
Python
76 lines
2.7 KiB
Python
"""The Fluksio engine, and the decorators that declare flows in your own code.
|
|
|
|
Two audiences, one import name. In a research repository, ``from fluksio
|
|
import Port, node, Flow`` is the authoring API: it says which of your existing
|
|
functions are nodes and which nodes make up a flow, and ``fluksio sync``
|
|
uploads what it finds. Those names come from :mod:`fluksio.sdk`, which imports
|
|
nothing but the standard library.
|
|
|
|
Inside a node, ``import fluksio`` is not this package at all: the worker
|
|
installs a reporter of its own under that name before any node code runs, so
|
|
what a node gets is :mod:`fluksio_worker.worker_main`'s ``emit`` and
|
|
``save_artifact`` — and inert copies of the decorators, since a module the
|
|
node imports may well declare them at its top. The stubs below stand in the
|
|
same place everywhere else, and say so rather than failing as a missing
|
|
attribute.
|
|
|
|
Nothing is imported from the rest of the package here. Every ``from fluksio.x
|
|
import y`` in the engine passes through this module, and an import cycle or a
|
|
second of start-up cost would both begin here.
|
|
"""
|
|
|
|
import logging
|
|
from importlib.metadata import PackageNotFoundError
|
|
from importlib.metadata import version as _version
|
|
from typing import Any
|
|
|
|
from fluksio.sdk import Flow, Port, node, use
|
|
|
|
try:
|
|
__version__ = _version("fluksio")
|
|
except PackageNotFoundError: # pragma: no cover - a checkout that was never installed
|
|
__version__ = "0.0.0+unknown"
|
|
|
|
__all__ = [
|
|
"Flow",
|
|
"Port",
|
|
"emit",
|
|
"load_artifact",
|
|
"node",
|
|
"save_artifact",
|
|
"use",
|
|
]
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_OUTSIDE = (
|
|
"fluksio.{name}() only works inside a node: the worker running it installs "
|
|
"the real one. There is nothing to {verb} out here."
|
|
)
|
|
|
|
|
|
def emit(**ports: Any) -> None:
|
|
"""Publish values on a node's declared output ports, mid-run.
|
|
|
|
Outside a node there is nowhere to publish to, and this does nothing on
|
|
purpose: a node function called directly — in a test, in a notebook, under
|
|
a debugger — is ordinary Python, and having it die on a progress report
|
|
would defeat the point of the code staying yours.
|
|
"""
|
|
logger.debug("fluksio.emit(%s) outside a node", ", ".join(ports))
|
|
|
|
|
|
def save_artifact(
|
|
source: Any,
|
|
name: str = "",
|
|
media_type: str = "application/octet-stream",
|
|
volatile: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Put bytes in the artifact store and return a reference to them."""
|
|
raise RuntimeError(_OUTSIDE.format(name="save_artifact", verb="save to"))
|
|
|
|
|
|
def load_artifact(ref: dict[str, Any]) -> str:
|
|
"""Fetch what an artifact reference points at, and return a path to it."""
|
|
raise RuntimeError(_OUTSIDE.format(name="load_artifact", verb="load from"))
|