From 0ffcabfdb92ef12d738b047d743daffc9e47519f Mon Sep 17 00:00:00 2001 From: stroblme Date: Wed, 26 Aug 2026 23:44:55 +0200 Subject: [PATCH] Media dtypes: image, audio and video as narrowed artifact references MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A port may now declare `image`, `audio` or `video`. Each is the artifact reference the engine already had, narrowed by the `media_type` on it, so a speech recogniser declares what it eats rather than taking any bytes at all and finding out. Bytes still never travel as a message and nothing on the wire stops being JSON: a camera publishes one reference per frame, a microphone one per chunk, and a reference may carry a `meta` dict nothing here interprets. Streaming media is therefore an ordinary streaming port — with one change to what that means. An emission used to journal an item with no payload, so downstream read whatever was current when the item was claimed; a consumer slower than its producer saw only the newest chunk and the ones between were lost. That is right for a training curve and wrong for a second of speech, so an emission now journals a `kind="emission"` item carrying its values, and the executor hands them to the nodes reading that message instead of writing them to state again. The value in state stays the latest, which is what everything else reads, and the wave is filtered by what actually changed rather than walking everything reachable. No queue serialization change — the existing `outputs` field carries it. Continuous media makes the store's missing GC a real problem, so this closes it: `sweep_artifacts` runs hourly, keeps every digest a `run_artifact` row records or a live message holds, spares anything written in the last hour, and stands aside entirely while a run is in flight, since a node may store a checkpoint long before it returns the reference to it. That also collects the orphans a deleted flow has always left behind. `ARTIFACT_GC_INTERVAL_S=0` turns it off. Around the edges: `GET /artifacts/{digest}` serves the media type the caller passes and answers ranged requests, so a browser plays a clip rather than downloading it; `PUT` spools to disk instead of holding the whole body in memory, as does `save_artifact` given a path; a Media widget draws whatever its message points at, and a wall panel may fetch the bytes its own tiles are showing and nothing else; and a connector gets `save_artifact`, for a device whose readings are bytes. What this cannot do is live video: a frame every second or two is a glance, and the honest answer above that is the camera's own stream, which the widget takes as a URL and the browser plays from source. Co-Authored-By: Claude Opus 5 (1M context) --- backend/fluksio/api/deps.py | 33 +++- backend/fluksio/api/routes/artifacts.py | 53 ++++-- backend/fluksio/core/config.py | 8 + backend/fluksio/flow/artifacts.py | 20 ++- backend/fluksio/flow/connector.py | 33 +++- backend/fluksio/flow/dashboards.py | 5 + backend/fluksio/flow/executor.py | 13 +- backend/fluksio/flow/messages.py | 46 +++++- backend/fluksio/flow/pipeline.py | 70 ++++++-- backend/fluksio/flow/queue.py | 9 +- backend/fluksio/flow/runs.py | 60 +++++++ backend/fluksio/main.py | 31 +++- backend/fluksio/sdk/__init__.py | 15 +- backend/tests/api/routes/test_panels.py | 42 +++++ backend/tests/flow/test_connector.py | 40 ++++- backend/tests/flow/test_dashboards.py | 11 ++ backend/tests/flow/test_messages.py | 49 ++++++ backend/tests/flow/test_queue.py | 91 +++++++++++ backend/tests/flow/test_workers.py | 72 +++++++++ backend/tests/test_artifacts.py | 110 +++++++++++++ docs/code/connectors.md | 15 ++ docs/code/nodes.md | 33 ++++ docs/concepts/flows.md | 22 ++- docs/interface/dashboards.md | 21 +++ docs/reference/configuration.md | 8 + docs/reference/connector-contract.md | 23 +++ docs/reference/payload-types.md | 34 ++++ frontend/scripts/capture-screenshots.mjs | 36 +++++ frontend/src/client/schemas.gen.ts | 13 +- frontend/src/client/sdk.gen.ts | 15 +- frontend/src/client/types.gen.ts | 14 +- .../src/components/Dashboard/MediaWidget.tsx | 153 ++++++++++++++++++ frontend/src/components/Dashboard/panels.tsx | 47 ++++++ frontend/src/components/Dashboard/widgets.tsx | 14 ++ frontend/src/components/Flow/NodePanel.tsx | 9 +- frontend/src/components/Flow/ValuePreview.tsx | 11 +- worker/fluksio_worker/worker_main.py | 54 +++++-- 37 files changed, 1271 insertions(+), 62 deletions(-) create mode 100644 backend/tests/test_artifacts.py create mode 100644 frontend/src/components/Dashboard/MediaWidget.tsx diff --git a/backend/fluksio/api/deps.py b/backend/fluksio/api/deps.py index ceae5d5..6f9940e 100644 --- a/backend/fluksio/api/deps.py +++ b/backend/fluksio/api/deps.py @@ -15,6 +15,7 @@ from fluksio.core import security from fluksio.core.config import settings from fluksio.core.db import engine from fluksio.flow import panels +from fluksio.flow.artifacts import is_reference from fluksio.flow.controller import FlowController from fluksio.flow.dashboards import DashboardStore from fluksio.flow.workers import PythonWorkerPool @@ -78,13 +79,36 @@ def _panel_messages(panel_id: str, request: Request) -> set[str]: return panels.messages_for(panel_id, store) +def _panel_digests(panel_id: str, request: Request) -> set[str]: + """The artifacts this panel's messages are pointing at right now. + + A media tile fetches the bytes its message names, so the messages already + bounding the panel bound this too — one step further along, through + whatever those messages currently hold. + """ + controller: FlowController | None = getattr( + request.app.state, "flow_controller", None + ) + if controller is None: + return set() + names = _panel_messages(panel_id, request) + if not names: + return set() + found: set[str] = set() + for value in controller.state.get_present(sorted(names)).values(): + if is_reference(value): + found.add(str(value["digest"])) + return found + + def _panel_may(payload: dict[str, Any], request: Request) -> None: """Refuse anything a wall panel has no business asking for. A panel credential names the account that approved the pairing, so without this it would be that person's session hanging on a wall. What a panel genuinely needs is small and worth writing out: the dashboards it was - assigned, its own definition, and the messages its own widgets bind to. + assigned, its own definition, the messages its own widgets bind to, and the + bytes those messages currently point at, for a tile drawing a camera frame. Publishing is in the list because a panel cannot be strictly read-only — a control on a panel is the point of putting one there, and a querying chart @@ -128,6 +152,13 @@ def _panel_may(payload: dict[str, Any], request: Request) -> None: else: name = "" allowed = bool(name) and name in _panel_messages(panel.id, request) + elif method == "GET" and path.startswith(f"{api}/artifacts/"): + # The bytes behind a media message a tile on this panel is drawing. + # Scoped to what those messages hold *now*, which is exactly what a + # live widget asks for — a screen has no business reading an artifact + # off an old run because it happens to know the digest. + digest = unquote(path[len(f"{api}/artifacts/") :]) + allowed = "/" not in digest and digest in _panel_digests(panel.id, request) if not allowed: raise HTTPException( diff --git a/backend/fluksio/api/routes/artifacts.py b/backend/fluksio/api/routes/artifacts.py index 7cb98e9..605887e 100644 --- a/backend/fluksio/api/routes/artifacts.py +++ b/backend/fluksio/api/routes/artifacts.py @@ -5,19 +5,28 @@ worker cannot — and having one path rather than two is what keeps a flow's code the same wherever it runs. """ +import re +import tempfile +from pathlib import Path from typing import Any from fastapi import APIRouter, Depends, HTTPException, Query, Request -from fastapi.responses import StreamingResponse +from fastapi.responses import FileResponse from jwt.exceptions import InvalidTokenError from pydantic import BaseModel from sqlmodel import Session +from starlette.concurrency import run_in_threadpool from fluksio.api.deps import user_from_token from fluksio.core import security from fluksio.core.db import engine from fluksio.flow.artifacts import ArtifactStore +#: What may be echoed back as a response content type. The caller holds the +#: reference and passes its media type, so this guards a header rather than +#: trusting one — anything else is served as bytes. +_MEDIA_TYPE = re.compile(r"^[\w.+-]+/[\w.+-]+$") + def artifact_caller(request: Request) -> str: """Who may move artifacts: a signed-in person, or an attached worker. @@ -71,21 +80,45 @@ async def put_artifact( name: str = Query(default=""), media_type: str = Query(default=""), ) -> Any: - """Store the request body and answer with the reference to it.""" + """Store the request body and answer with the reference to it. + + Spooled to disk as it arrives rather than buffered: a video segment is as + legitimate a body here as a checkpoint, and neither should have to fit in + memory twice. + """ store = _store(request) - body = await request.body() - return store.put([body], name=name, media_type=media_type) + handle = tempfile.NamedTemporaryFile(dir=store.root, delete=False) + try: + with handle: + async for chunk in request.stream(): + handle.write(chunk) + return await run_in_threadpool( + store.put_file, Path(handle.name), media_type, name + ) + finally: + Path(handle.name).unlink(missing_ok=True) @router.get("/{digest}") -def get_artifact(digest: str, request: Request) -> Any: - """Stream one artifact back.""" +def get_artifact( + digest: str, request: Request, media_type: str = Query(default="") +) -> Any: + """Serve one artifact back. + + The caller passes the media type off the reference it holds, which is what + lets a browser play a clip rather than download it; the store itself keeps + only bytes. Ranged requests are answered because an audio or video element + scrubbing through a file asks for them. + """ store = _store(request) path = store.path(digest) if path is None: raise HTTPException(status_code=404, detail="No such artifact") - return StreamingResponse( - store.read(digest), - media_type="application/octet-stream", - headers={"Content-Length": str(path.stat().st_size)}, + return FileResponse( + path, + media_type=( + media_type if _MEDIA_TYPE.match(media_type) else "application/octet-stream" + ), + # The digest is the content, so it is also the perfect validator. + headers={"ETag": f'"{digest}"'}, ) diff --git a/backend/fluksio/core/config.py b/backend/fluksio/core/config.py index 39a8012..e5f6339 100644 --- a/backend/fluksio/core/config.py +++ b/backend/fluksio/core/config.py @@ -113,6 +113,14 @@ class Settings(BaseSettings): FLOW_GPUS: int = 0 # How long the engine's own metrics, events and run records are kept. OBS_RETENTION_DAYS: int = 30 + # How often artifact bytes nothing refers to any more are swept away; 0 + # never sweeps. A flow streaming media writes one artifact per frame, so + # without this the store only grows. + ARTIFACT_GC_INTERVAL_S: int = 3600 + # How long a freshly written artifact is spared, whatever refers to it. + # Storing bytes and recording the reference are two steps; this is the + # window between them. + ARTIFACT_GC_GRACE_S: int = 3600 # Without a Redis host the engine keeps its state in memory. REDIS_HOST: str | None = None REDIS_PORT: int = 6379 diff --git a/backend/fluksio/flow/artifacts.py b/backend/fluksio/flow/artifacts.py index 61acedd..616e1ec 100644 --- a/backend/fluksio/flow/artifacts.py +++ b/backend/fluksio/flow/artifacts.py @@ -19,6 +19,7 @@ import hashlib import logging import os import tempfile +import time from collections.abc import Iterable, Iterator from pathlib import Path from typing import Any @@ -96,11 +97,13 @@ class ArtifactStore: "name": name, } - def put_file(self, path: Path, media_type: str = "") -> dict[str, Any]: + def put_file( + self, path: Path, media_type: str = "", name: str = "" + ) -> dict[str, Any]: with path.open("rb") as handle: return self.put( iter(lambda: handle.read(CHUNK), b""), - name=path.name, + name=name or path.name, media_type=media_type, ) @@ -119,20 +122,29 @@ class ArtifactStore: while chunk := handle.read(CHUNK): yield chunk - def collect(self, keep: set[str]) -> int: - """Delete what no run refers to any more. Returns how many went. + def collect(self, keep: set[str], grace_s: float = 0.0) -> int: + """Delete what nothing refers to any more. Returns how many went. The caller passes every digest still recorded; anything else in the store was produced by a run that has since been pruned, or never got a row at all because the run failed between writing and recording. + + ``grace_s`` spares anything written that recently. Storing bytes and + recording the reference to them are two steps, and a sweep landing + between them would take an artifact its run is about to name — so + recent files are left for the next pass, by which time they are either + referenced or genuinely orphaned. """ removed = 0 + cutoff = time.time() - grace_s for entry in self.root.glob("*/*"): if not entry.is_file(): continue if DIGEST_PREFIX + entry.name in keep: continue try: + if grace_s > 0 and entry.stat().st_mtime > cutoff: + continue entry.unlink() removed += 1 except OSError: diff --git a/backend/fluksio/flow/connector.py b/backend/fluksio/flow/connector.py index b55c23e..ac49e4b 100644 --- a/backend/fluksio/flow/connector.py +++ b/backend/fluksio/flow/connector.py @@ -17,7 +17,10 @@ What a connector gets from the base class: 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. + 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 @@ -37,6 +40,8 @@ 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. @@ -81,7 +86,7 @@ class ConnectorNode(Node): description="Seconds between polls; 0 polls never.", ) - __slots__ = ("config", "_poll_task", "_stop_event", "_last_published") + __slots__ = ("config", "_poll_task", "_stop_event", "_last_published", "_artifacts") def __init__(self, **kwargs: Any) -> None: super().__init__(f=self._dispatch, **kwargs) @@ -89,6 +94,7 @@ class ConnectorNode(Node): 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 def _dispatch(self, params: dict[str, Any], **ports: Any) -> dict[str, Any] | None: """The scheduler's entry point. Settings are already on ``self.config``.""" @@ -122,7 +128,30 @@ class ConnectorNode(Node): # 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) diff --git a/backend/fluksio/flow/dashboards.py b/backend/fluksio/flow/dashboards.py index 688a64d..40db80c 100644 --- a/backend/fluksio/flow/dashboards.py +++ b/backend/fluksio/flow/dashboards.py @@ -63,6 +63,7 @@ WidgetType = Literal[ "icon", "forecast", "clock", + "media", # Input "button", "switch", @@ -100,6 +101,10 @@ WIDGET_DTYPES: dict[str, set[str]] = { # Either shape a colour can travel as; which of the two this widget means # is its ``format``, checked against ``COLOR_DTYPES`` below. "color": {"list", "str"}, + # A camera frame, a clip, a segment. What it draws follows the type it is + # bound to; a plain artifact is taken as well, since the bytes may be + # anything and the media type on the reference is what says what they are. + "media": {"image", "audio", "video", "artifact"}, # An icon maps weather strings, bool hints and numbers alike, and a clock # binds nothing at all, so neither has a row to be held to. } diff --git a/backend/fluksio/flow/executor.py b/backend/fluksio/flow/executor.py index 77ab476..f933a8e 100644 --- a/backend/fluksio/flow/executor.py +++ b/backend/fluksio/flow/executor.py @@ -398,8 +398,18 @@ class ExecutionService: ) replay = item.deliveries > 1 + emission = item.kind == "emission" try: - published = pipeline.apply_outputs(node, item.outputs or None) + # An emission's values went into state when the node produced them; + # this item carries them so its readers get the chunk that caused + # the wave rather than whichever is newest by the time they run. + # Applying them again would let a mid-node emission overwrite what + # the node returned at the end. + published = ( + set(item.outputs) + if emission + else pipeline.apply_outputs(node, item.outputs or None) + ) pipeline.run_downstream( node, entry_id=item.entry_id, @@ -408,6 +418,7 @@ class ExecutionService: # an item with no payload is the value already being in state. # Neither can say what changed, so neither filters on it. changed=None if replay or not item.outputs else published, + overrides=item.outputs if emission else None, ) finally: # Paired, or a cascade that raised — state backend gone, say — is a diff --git a/backend/fluksio/flow/messages.py b/backend/fluksio/flow/messages.py index 261e85d..3629e87 100644 --- a/backend/fluksio/flow/messages.py +++ b/backend/fluksio/flow/messages.py @@ -33,6 +33,13 @@ class DType(str, Enum): thirty-megabyte checkpoint never sits in Redis. Inline codecs would only be needed for payloads too small to be worth a round trip, and nothing asks for that yet. + + ``image``, ``audio`` and ``video`` are that same reference narrowed to a + media family, so a speech recogniser declares what it eats rather than + taking any bytes at all and finding out. Bytes still never travel as a + message: a camera publishes one reference per frame, a microphone one per + chunk. A reference may carry a ``meta`` dict — sample rate, dimensions, a + sequence number — which nothing here interprets. """ FLOAT = "float" @@ -51,6 +58,12 @@ class DType(str, Enum): #: A reference to stored bytes: #: ``{"digest": "sha256:…", "size": int, "media_type": str, "name": str}``. ARTIFACT = "artifact" + #: An artifact reference whose ``media_type`` is ``image/*``. + IMAGE = "image" + #: An artifact reference whose ``media_type`` is ``audio/*``. + AUDIO = "audio" + #: An artifact reference whose ``media_type`` is ``video/*``. + VIDEO = "video" _JSON_TYPES = (dict, list, str, int, float, bool, type(None)) @@ -65,6 +78,14 @@ _ITEM_TYPES = frozenset( {DType.FLOAT, DType.INT, DType.STR, DType.BOOL, DType.JSON, DType.RECORD} ) +#: The media dtypes, and the ``media_type`` family a reference must declare to +#: satisfy each. A stream of them is a stream of references, one per chunk. +MEDIA_FAMILIES = { + DType.IMAGE: "image/", + DType.AUDIO: "audio/", + DType.VIDEO: "video/", +} + #: How deep to look for a non-finite number. Deeper than any payload that #: reads well on a canvas, and a bound on a value that refers to itself. @@ -147,6 +168,11 @@ def _is_artifact(value: Any) -> bool: ) +def _media_family(value: Any) -> str: + """The ``media_type`` an artifact reference declares, lowercased.""" + return str(value.get("media_type") or "").lower() + + def _matches(dtype: DType, value: Any) -> bool: """Whether one value satisfies a scalar or record type.""" if dtype is DType.BOOL: @@ -161,6 +187,10 @@ def _matches(dtype: DType, value: Any) -> bool: return _is_record(value) if dtype is DType.ARTIFACT: return _is_artifact(value) + if dtype in MEDIA_FAMILIES: + return _is_artifact(value) and _media_family(value).startswith( + MEDIA_FAMILIES[dtype] + ) return isinstance(value, _JSON_TYPES) @@ -243,6 +273,14 @@ class MessageSpec(BaseModel): else: ok = _matches(self.dtype, value) if not ok: + if self.dtype in MEDIA_FAMILIES and _is_artifact(value): + # It is a reference, just to the wrong kind of bytes — saying so + # beats "expected audio, got dict" on a media_type typo. + raise TypeError( + f"{where}: expected {self.dtype.value} " + f"({MEDIA_FAMILIES[self.dtype]}*), got an artifact of " + f"'{_media_family(value) or 'no media type'}'" + ) raise TypeError( f"{where}: expected {self.dtype.value}, got {type(value).__name__}" ) @@ -263,7 +301,13 @@ class MessageSpec(BaseModel): return str(value).lower() in ("true", "1", "yes", "on") if self.dtype is DType.STR: return value if isinstance(value, str) else json.dumps(value) - if self.dtype in (DType.SERIES, DType.RECORD, DType.LIST, DType.ARTIFACT): + if self.dtype in ( + DType.SERIES, + DType.RECORD, + DType.LIST, + DType.ARTIFACT, + *MEDIA_FAMILIES, + ): # A structured payload arriving as text is the same hint a numeric # one is; the shape itself is still checked afterwards. return json.loads(value) if isinstance(value, str) else value diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py index 7bace47..beddd25 100644 --- a/backend/fluksio/flow/pipeline.py +++ b/backend/fluksio/flow/pipeline.py @@ -928,7 +928,11 @@ class Pipeline: return True, outputs def _execute_node( - self, node: Node, state: StateBackend, entry_id: str = "" + self, + node: Node, + state: StateBackend, + entry_id: str = "", + overrides: dict[str, Any] | None = None, ) -> dict[str, Any] | None: """Run one node and record its outputs. Never raises.""" started = time.perf_counter() @@ -938,6 +942,17 @@ class Pipeline: # single bulk read does not, and it was the engine's one global # mutex — every node of every cascade queued behind it. inputs = state.get_present(list(node.requires)) + if overrides: + # The value this wave is delivering wins over whatever state + # holds by now. Not written back: the newest value is still the + # one everything else reads. + inputs.update( + { + name: value + for name, value in overrides.items() + if name in node.requires + } + ) key = "" if self.run_cache is not None and node.fingerprint: @@ -1074,13 +1089,16 @@ class Pipeline: return self._record_outputs(node, passed, self._state) if self._queue is not None: - # Journalled with no payload of its own: the value is already in - # state, published in the order it was produced. An item carrying - # it would re-apply that value whenever it happened to be claimed, - # which is how an emission from the middle of a node overwrites the - # one it returned at the end. Downstream reads what is current, - # which is what "the latest value wins" has always meant here. - self._enqueue_cascade(node, None) + # Journalled carrying the emitted values, as an ``emission`` item: + # the executor hands them to the nodes reading them instead of + # writing them to state a second time. That distinction is the + # whole of it — re-applying would let a mid-node emission overwrite + # the value the node returned at the end, while reading state + # instead means a consumer slower than its producer sees only the + # newest chunk and the ones between are lost. A frame of video or a + # second of speech is worth delivering; the value in state stays + # the latest, which is what everything else reads. + self._enqueue_cascade(node, passed, kind="emission") def _observe(self, outcome: NodeOutcome) -> None: """Tell the run watching this pipeline, if there is one.""" @@ -1166,6 +1184,7 @@ class Pipeline: entry_id: str = "", replay: bool = False, changed: set[str] | None = None, + overrides: dict[str, Any] | None = None, ) -> StateBackend: """Execute nodes concurrently, scheduling each as its inputs arrive. @@ -1173,6 +1192,10 @@ class Pipeline: whose triggering inputs are all untouched is completed without being run, so what is downstream of *it* is judged on the same footing. None runs everything the subset holds, which is what a manual run means. + + ``overrides`` reaches the nodes that read those names directly, and no + further: a value carried by this wave is what its readers should see, + while everything past them reads what those readers produced. """ # One view of the graph for the whole wave: a flow replaced halfway # through must not have this wave asking the new dependencies about a @@ -1248,7 +1271,7 @@ class Pipeline: continue submitted.add(n) node_futures[n] = executor.submit( - self._execute_node, n, state, entry_id + self._execute_node, n, state, entry_id, overrides ) def drain(executor: ThreadPoolExecutor) -> None: @@ -1348,6 +1371,7 @@ class Pipeline: entry_id: str = "", replay: bool = False, changed: set[str] | None = None, + overrides: dict[str, Any] | None = None, ) -> StateBackend: """Run everything downstream of a node that has just published. @@ -1361,6 +1385,10 @@ class Pipeline: already read, which is how one node came to publish 619 messages a minute off inputs that changed six times. ``None`` means walk everything reachable, which is what a manual run wants. + + ``overrides`` carries values to hand to whoever reads them instead of + what state holds — an emission delivering the chunk that caused this + wave rather than whichever one is newest by the time it runs. """ if changed is not None and not changed: # Everything the cascade carried was held back by a rate limit, so @@ -1376,6 +1404,7 @@ class Pipeline: entry_id=entry_id, replay=replay, changed=changed, + overrides=overrides, ) def trigger( @@ -1439,13 +1468,21 @@ class Pipeline: return False def _run_here( - self, node: Node, outputs: dict[str, Any] | None, cause: str = "manual" + self, + node: Node, + outputs: dict[str, Any] | None, + cause: str = "manual", + delivered: bool = False, ) -> StateBackend: """Run a cascade in this thread, under a run id of its own. The queued path gets its run id from the journal entry. A run that never went through the queue still belongs in the history, so it makes one — marked as such, because it is no one's idempotency key. + + ``delivered`` marks an emission the queue could not take: its values are + already in state, so they are handed to their readers rather than + written again. """ run_id = f"{MANUAL_RUN_PREFIX}{uuid.uuid4().hex[:12]}" self._publish( @@ -1460,13 +1497,16 @@ class Pipeline: } ) try: - published = self.apply_outputs(node, outputs) + published = ( + set(outputs or {}) if delivered else self.apply_outputs(node, outputs) + ) state = self.run_downstream( node, entry_id=run_id, # No payload means the value is already in state and this is a # wake-up, which has nothing to name as changed. changed=published if outputs else None, + overrides=outputs if delivered else None, ) finally: # Paired, or a cascade that raised leaves the run open until the @@ -1564,12 +1604,14 @@ class Pipeline: logger.error("Could not defer work for '%s': %s", node.id, exc) return False - def _enqueue_cascade(self, node: Node, outputs: dict[str, Any] | None) -> None: + def _enqueue_cascade( + self, node: Node, outputs: dict[str, Any] | None, kind: str = "cascade" + ) -> None: """Journal a trigger, or fall back to running it here if that fails.""" from fluksio.flow.queue import WorkItem item = WorkItem( - kind="cascade", + kind=kind, node=node.id, flow=node.flow, outputs=outputs or {}, @@ -1591,7 +1633,7 @@ class Pipeline: } ) # Losing the value outright would be worse than running it here. - self._run_here(node, outputs, cause="external") + self._run_here(node, outputs, cause="external", delivered=kind == "emission") def values(self, flow: str | None = None) -> dict[str, dict[str, Any]]: """Last value and timestamp of every message, optionally one flow's. diff --git a/backend/fluksio/flow/queue.py b/backend/fluksio/flow/queue.py index 1953db4..b9a9b12 100644 --- a/backend/fluksio/flow/queue.py +++ b/backend/fluksio/flow/queue.py @@ -40,12 +40,15 @@ class WorkItem: """One unit of journaled work. :param kind: ``cascade`` replays a node's outputs and runs what is - downstream; ``flush`` lets out what a node's rate limits held back; - ``run`` is a whole batch run, and carries only its id. + downstream; ``emission`` is a value a node published while still + running, already in state, carried so its readers get *that* value + rather than whichever is newest when they run; ``flush`` lets out what + a node's rate limits held back; ``run`` is a whole batch run, and + carries only its id. :param node: The node the item is about — the source for a cascade, the one whose rate limits are released for a flush. :param flow: The flow that node belongs to, so gating needs no lookup. - :param outputs: What the source node emitted (cascade only). + :param outputs: What the source node emitted (cascade and emission). :param cause: Where the work came from, for logs and debugging. :param not_before: Epoch seconds before which the item must not run. :param guard: ``(key, value)`` the target node must still remember for diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py index 1e21a75..811ea3b 100644 --- a/backend/fluksio/flow/runs.py +++ b/backend/fluksio/flow/runs.py @@ -295,6 +295,66 @@ def _from_digest( } +#: How deep a stored value is walked looking for artifact references. A +#: reference nested past this is not something any port declares. +_REF_DEPTH = 8 +#: How many state keys are read back at a time by the sweep. +_SWEEP_BATCH = 500 + + +def _references_in(value: Any, found: set[str], depth: int = 0) -> None: + """Every artifact digest inside one stored value.""" + if is_reference(value): + found.add(str(value["digest"])) + return + if depth >= _REF_DEPTH: + return + if isinstance(value, dict): + items: Any = value.values() + elif isinstance(value, (list, tuple)): + items = value + else: + return + for item in items: + _references_in(item, found, depth + 1) + + +def sweep_artifacts( + store: ArtifactStore, state: StateBackend, grace_s: float = 0.0 +) -> int: + """Remove artifact bytes nothing refers to any more. Returns how many went. + + Two things refer to an artifact: a run that recorded it, and a message + currently holding it. The second is what makes a stream of media + affordable — a camera publishing a frame a second replaces the reference + each time, so yesterday's frames are unreferenced by definition and the + store does not grow without bound. What a node *returns* is recorded + against its run and kept; what it emits along the way is not. + + Skipped entirely while anything is running: a node that stores a checkpoint + an hour before it returns has neither a row nor a message naming it yet, + and a sweep in that window would take the bytes out from under it. + """ + with Session(db_engine) as session: + active = session.exec( + select(Run.id).where(col(Run.status).in_(("running", "queued"))).limit(1) + ).first() + if active is not None: + logger.debug("Artifact sweep skipped: run %s is in flight", active) + return 0 + keep = set(session.exec(select(col(RunArtifact.digest)).distinct()).all()) + + keys = state.keys() + for start in range(0, len(keys), _SWEEP_BATCH): + for value in state.get_multi(keys[start : start + _SWEEP_BATCH]).values(): + _references_in(value, keep) + + removed = store.collect(keep, grace_s) + if removed: + logger.info("Artifact sweep removed %d unreferenced artifacts", removed) + return removed + + def seed_values( flow: FlowDef, params: dict[str, Any], seed: int | None = None ) -> dict[str, Any]: diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py index 7456ddd..e1c9def 100644 --- a/backend/fluksio/main.py +++ b/backend/fluksio/main.py @@ -1,5 +1,6 @@ import asyncio import contextlib +import logging from collections.abc import AsyncIterator from contextlib import AbstractAsyncContextManager, asynccontextmanager @@ -32,13 +33,15 @@ from fluksio.flow.plugins import load_plugins from fluksio.flow.queue import MemoryWorkQueue, RedisWorkQueue, WorkQueue from fluksio.flow.remote import RemoteWorkerHub from fluksio.flow.resources import ResourceAccountant, fair_share_env -from fluksio.flow.runs import RUN_STATE_TTL, RunService +from fluksio.flow.runs import RUN_STATE_TTL, RunService, sweep_artifacts from fluksio.flow.secrets import init_secrets from fluksio.flow.state import MemoryState, RedisState, StateBackend from fluksio.flow.store import FlowStore from fluksio.flow.watchdog import LoopWatchdog from fluksio.flow.workers import PythonWorkerPool +logger = logging.getLogger(__name__) + def custom_generate_unique_id(route: APIRoute) -> str: return f"{route.tags[0]}-{route.name}" @@ -80,6 +83,28 @@ def _run_state(namespace: str) -> StateBackend: return MemoryState() +async def _sweep_artifacts(store: ArtifactStore, controller: FlowController) -> None: + """Take unreferenced artifact bytes off the disk, on a slow loop. + + A flow streaming media writes one artifact per frame, so a store nothing + prunes only grows. Runs in a thread: it walks a directory and reads state. + """ + interval = settings.ARTIFACT_GC_INTERVAL_S + if interval <= 0: + return + while True: + await asyncio.sleep(interval) + try: + await run_in_threadpool( + sweep_artifacts, + store, + controller.state, + settings.ARTIFACT_GC_GRACE_S, + ) + except Exception: + logger.exception("Artifact sweep failed") + + def _mcp_sessions() -> AbstractAsyncContextManager[None]: """The MCP session manager's run scope, or nothing when MCP is off.""" if not settings.MCP_ENABLED: @@ -181,6 +206,9 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: metrics_task = asyncio.create_task( MetricsCollector(event_bus).run(), name="metrics-collector" ) + gc_task = asyncio.create_task( + _sweep_artifacts(artifacts, controller), name="artifact-gc" + ) await controller.start() run_service.start() # Optional, and off unless someone enrolled this installation: the @@ -208,6 +236,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: watchdog_task.cancel() alerts_task.cancel() metrics_task.cancel() + gc_task.cancel() enrol_task.cancel() # Re-read from app.state: enrolling at runtime replaces this. running_cloud = getattr(app.state, "cloud_task", None) or cloud_task diff --git a/backend/fluksio/sdk/__init__.py b/backend/fluksio/sdk/__init__.py index e41f1aa..951cb56 100644 --- a/backend/fluksio/sdk/__init__.py +++ b/backend/fluksio/sdk/__init__.py @@ -39,7 +39,20 @@ F = TypeVar("F", bound=Callable[..., Any]) #: Mirrors :class:`fluksio.flow.messages.DType`. Mirrored rather than imported: #: importing it would pull the engine into a research process. DTYPES = frozenset( - {"float", "int", "str", "bool", "json", "series", "record", "list", "artifact"} + { + "float", + "int", + "str", + "bool", + "json", + "series", + "record", + "list", + "artifact", + "image", + "audio", + "video", + } ) #: What a list may hold, mirroring ``messages._ITEM_TYPES``. ITEM_TYPES = frozenset({"float", "int", "str", "bool", "json", "record"}) diff --git a/backend/tests/api/routes/test_panels.py b/backend/tests/api/routes/test_panels.py index 0ca689c..b6f3bee 100644 --- a/backend/tests/api/routes/test_panels.py +++ b/backend/tests/api/routes/test_panels.py @@ -764,3 +764,45 @@ def test_a_panels_socket_carries_only_what_it_draws( # A person's credential is not bounded at all. assert panel_scope(superuser_token_headers["Authorization"][7:], client.app) is None + + +def test_a_panel_fetches_only_the_media_its_tiles_are_showing( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + """A media tile needs the bytes, and only the ones its message points at.""" + _dashboard_with( + client, + superuser_token_headers, + "panel_camera", + [ + { + "id": "w_media", + "type": "media", + "config": {"message": "demo.frame", "dtype": "image"}, + } + ], + ) + _panels( + client, + superuser_token_headers, + {"panels": [{"id": "porch", "dashboards": ["panel_camera"]}]}, + ) + panel_headers = _pair(client, superuser_token_headers, "porch") + + store = client.app.state.artifact_store + shown = store.put([b"the frame on the wall"], media_type="image/png") + elsewhere = store.put([b"an artifact of some other run"]) + client.app.state.flow_controller.state.update({"demo.frame": shown}) + + artifacts = f"{settings.API_V1_STR}/artifacts" + assert ( + client.get(f"{artifacts}/{shown['digest']}", headers=panel_headers).status_code + == 200 + ) + # Knowing a digest is not being entitled to it: a screen reads what it draws. + assert ( + client.get( + f"{artifacts}/{elsewhere['digest']}", headers=panel_headers + ).status_code + == 403 + ) diff --git a/backend/tests/flow/test_connector.py b/backend/tests/flow/test_connector.py index a2473ed..ed44191 100644 --- a/backend/tests/flow/test_connector.py +++ b/backend/tests/flow/test_connector.py @@ -1,8 +1,12 @@ """The connector contract: polling, deduplication, health and discovery.""" import asyncio +from types import SimpleNamespace from typing import Any +import pytest + +from fluksio.flow.artifacts import ArtifactStore from fluksio.flow.connector import CONTRACT_VERSION, ConnectorNode from fluksio.flow.controller import NODE_TYPES from fluksio.flow.messages import DType, MessageSpec @@ -44,11 +48,11 @@ def a_sensor(readings: list[Any], **params: Any) -> Sensor: return node -def run_briefly(node: ConnectorNode, seconds: float = 0.12) -> None: +def run_briefly(node: ConnectorNode, seconds: float = 0.12, app: Any = None) -> None: """Start the poll loop, let it tick a few times, stop it.""" async def cycle() -> None: - await node.start() + await node.start(app) await asyncio.sleep(seconds) await node.stop() @@ -188,3 +192,35 @@ def test_a_connector_may_not_take_over_a_built_in_type(monkeypatch): ) assert load_plugins() == [] assert NODE_TYPES["mqtt"].plugin is None + + +def test_a_connector_publishes_bytes_as_a_media_reference(tmp_path): + """A camera's reading is bytes, and bytes never travel as a message.""" + + class Camera(ConnectorNode): + contract = CONTRACT_VERSION + + async def poll(self) -> dict[str, Any] | None: + return {"frame": self.save_artifact(b"\x89PNG...", "f.png", "image/png")} + + node = Camera( + provides=[MessageSpec(name="frame", dtype=DType.IMAGE)], + params={"poll_interval": 0.01}, + ) + node.assign_flow("demo", "camera") + pipeline = Pipeline(nodes=[node]) + + # Before it starts there is no store to write to, and saying so beats an + # AttributeError from inside somebody's connector. + with pytest.raises(RuntimeError, match="artifact store"): + node.save_artifact(b"x") + + store = ArtifactStore(tmp_path / "artifacts") + app = SimpleNamespace(state=SimpleNamespace(artifact_store=store)) + run_briefly(node, app=app) + + reference = pipeline.state["demo.frame"] + # It typechecks against the port it was published on, which is the whole + # point of a media dtype. + MessageSpec(name="demo.frame", dtype=DType.IMAGE).check(reference) + assert store.path(reference["digest"]).read_bytes() == b"\x89PNG..." diff --git a/backend/tests/flow/test_dashboards.py b/backend/tests/flow/test_dashboards.py index cfa582e..67ce519 100644 --- a/backend/tests/flow/test_dashboards.py +++ b/backend/tests/flow/test_dashboards.py @@ -241,6 +241,17 @@ def test_the_structured_widgets_bind_their_shapes(): WidgetDef(id="f", type="forecast", config={"message": "a.b", "dtype": "json"}) +def test_a_media_widget_binds_media_and_nothing_else(): + for dtype in ("image", "audio", "video", "artifact"): + WidgetDef(id="m", type="media", config={"message": "cam.frame", "dtype": dtype}) + + for dtype in ("float", "record"): + with pytest.raises(ValueError): + WidgetDef( + id="m", type="media", config={"message": "cam.frame", "dtype": dtype} + ) + + def test_a_bar_nests_a_second_number(): WidgetDef( id="b", diff --git a/backend/tests/flow/test_messages.py b/backend/tests/flow/test_messages.py index a4de9ce..8081dd7 100644 --- a/backend/tests/flow/test_messages.py +++ b/backend/tests/flow/test_messages.py @@ -91,6 +91,55 @@ def test_a_list_holds_one_declared_level(): MessageSpec(name="x", dtype=DType.LIST, item=item) +def _ref(media_type: str) -> dict: + return { + "digest": "sha256:" + "ab" * 32, + "size": 12, + "media_type": media_type, + "name": "chunk", + } + + +def test_a_media_port_checks_the_family(): + spec = MessageSpec(name="speech", dtype=DType.AUDIO) + spec.check(_ref("audio/wav")) + for wrong in (_ref("video/mp4"), _ref(""), {"a": 1}, "sha256:x"): + with pytest.raises(TypeError): + spec.check(wrong) + + +def test_a_media_mismatch_names_the_media_type(): + spec = MessageSpec(name="speech", dtype=DType.AUDIO) + with pytest.raises(TypeError, match="video/mp4"): + spec.check(_ref("video/mp4")) + + +def test_an_artifact_port_accepts_a_media_reference(): + # Media narrows artifact, so the wider port still takes it; the reverse is + # what the family check refuses. + MessageSpec(name="blob", dtype=DType.ARTIFACT).check(_ref("image/png")) + with pytest.raises(TypeError): + MessageSpec(name="frame", dtype=DType.IMAGE).check( + {"digest": "sha256:" + "cd" * 32, "size": 1} + ) + + +def test_a_media_reference_may_carry_meta(): + spec = MessageSpec(name="frame", dtype=DType.IMAGE) + spec.check({**_ref("image/png"), "meta": {"width": 640, "seq": 3}}) + + +def test_a_list_refuses_media_items(): + for item in (DType.IMAGE, DType.AUDIO, DType.VIDEO): + with pytest.raises(ValueError): + MessageSpec(name="x", dtype=DType.LIST, item=item) + + +def test_coerce_parses_a_media_reference_from_text(): + spec = MessageSpec(name="frame", dtype=DType.IMAGE) + assert spec.coerce(json.dumps(_ref("image/png"))) == _ref("image/png") + + def test_coerce_parses_structured_text(): spec = MessageSpec(name="notice", dtype=DType.RECORD) assert spec.coerce('{"title": "Boiler"}') == {"title": "Boiler"} diff --git a/backend/tests/flow/test_queue.py b/backend/tests/flow/test_queue.py index 25c3a99..18d08ee 100644 --- a/backend/tests/flow/test_queue.py +++ b/backend/tests/flow/test_queue.py @@ -431,3 +431,94 @@ def test_no_more_is_claimed_than_the_pool_can_run(): finally: release.set() service.stop() + + +# ----------------------------------------------------------------------------- +# Emissions: values a node publishes while it is still running +# ----------------------------------------------------------------------------- + + +def _emitting_pipeline() -> tuple[Pipeline, Node, MemoryState, list]: + """A source with a streaming port, and a consumer that records every value.""" + seen: list[float] = [] + + def consume(chunk, params): + seen.append(chunk) + return None + + source = Node( + f=lambda params: None, + provides=[ + MessageSpec(name="chunk", port="chunk", dtype=DType.FLOAT, stream=True) + ], + name="source", + ) + consumer = Node( + f=consume, + requires=[MessageSpec(name="chunk", port="chunk", dtype=DType.FLOAT)], + name="consumer", + ) + source.assign_flow("f", "source") + consumer.assign_flow("f", "consumer") + + state = MemoryState() + queue = MemoryWorkQueue() + pipeline = Pipeline(nodes=[source, consumer], state=state, work_queue=queue) + return pipeline, source, state, seen + + +def test_every_emitted_chunk_reaches_the_consumer(): + """A consumer slower than its producer must not skip what it missed. + + Reading state instead would give whichever chunk is newest by the time the + item runs — fine for a temperature, lossy for a second of speech. + """ + pipeline, source, state, seen = _emitting_pipeline() + service = ExecutionService(pipeline._queue) + service.bind(pipeline) + + # Both emitted before either item is claimed, so state has moved on. + pipeline.publish_emission(source, {"f.chunk": 1.0}) + pipeline.publish_emission(source, {"f.chunk": 2.0}) + assert state["f.chunk"] == 2.0 + + for item in pipeline._queue.claim(10, 10): + service._run_item(item) + + assert seen == [1.0, 2.0] + + +def test_a_delivered_emission_does_not_write_state_a_second_time(): + """The value in state stays the newest one, whenever an item is claimed.""" + pipeline, source, state, seen = _emitting_pipeline() + service = ExecutionService(pipeline._queue) + service.bind(pipeline) + + pipeline.publish_emission(source, {"f.chunk": 1.0}) + # What the node returned at the end, after the emission it made on the way. + pipeline.apply_outputs(source, {"f.chunk": 9.0}) + + for item in pipeline._queue.claim(10, 10): + service._run_item(item) + + assert seen[0] == 1.0 + # Re-applying the carried chunk here is what would undo the final value. + assert state["f.chunk"] == 9.0 + + +def test_a_throttled_emission_wakes_nothing(): + pipeline, source, _state, seen = _emitting_pipeline() + source.provides["f.chunk"] = MessageSpec( + name="f.chunk", port="chunk", dtype=DType.FLOAT, stream=True, interval=60 + ) + service = ExecutionService(pipeline._queue) + service.bind(pipeline) + + pipeline.publish_emission(source, {"f.chunk": 1.0}) + pipeline.publish_emission(source, {"f.chunk": 2.0}) + for item in pipeline._queue.claim(10, 10): + service._run_item(item) + + # The first is let through; the second is held by the interval, and a value + # nothing published is nothing to wake on. + assert seen == [1.0] diff --git a/backend/tests/flow/test_workers.py b/backend/tests/flow/test_workers.py index a06b0ed..4d8c9a0 100644 --- a/backend/tests/flow/test_workers.py +++ b/backend/tests/flow/test_workers.py @@ -455,6 +455,78 @@ def test_a_node_saves_and_loads_an_artifact(tmp_path): pool.stop() +def test_a_node_passes_audio_to_the_next_one(tmp_path): + # The media path end to end: one node writes a clip and declares what kind + # of bytes it is, the next opens it. Only the reference travels. + store = ArtifactStore(tmp_path / "artifacts") + pool = PythonWorkerPool( + python=sys.executable, size=1, env={ARTIFACT_DIR_ENV: str(store.root)} + ) + pool.start() + try: + ref = pool.run( + "demo", + "speak", + "import fluksio, io, math, struct, wave\n" + "def process():\n" + " buffer = io.BytesIO()\n" + " with wave.open(buffer, 'wb') as out:\n" + " out.setnchannels(1)\n" + " out.setsampwidth(2)\n" + " out.setframerate(8000)\n" + " out.writeframes(b''.join(\n" + " struct.pack(' ArtifactStore: + return ArtifactStore(tmp_path / "artifacts") + + +def _run_row(status: str, digest: str = "") -> str: + """A run, and optionally the artifact it recorded. Returns its id.""" + run_id = new_run_id() + with Session(db_engine) as session: + session.add( + Run(id=run_id, flow="f", status=status, created_at=datetime.now(UTC)) + ) + if digest: + session.add( + RunArtifact( + run_id=run_id, + name="f.out", + node="n", + digest=digest, + size=3, + ) + ) + session.commit() + return run_id + + +def _forget(run_id: str) -> None: + with Session(db_engine) as session: + rows = session.exec( + select(RunArtifact).where(col(RunArtifact.run_id) == run_id) + ).all() + for row in rows: + session.delete(row) + session.delete(session.get(Run, run_id)) + session.commit() + + +def test_the_grace_window_spares_a_fresh_artifact(store): + ref = store.put([b"new"]) + assert store.collect(set(), grace_s=3600) == 0 + assert store.path(ref["digest"]) is not None + assert store.collect(set(), grace_s=0) == 1 + assert store.path(ref["digest"]) is None + + +def test_a_sweep_keeps_what_a_run_recorded_and_what_a_message_holds(store): + recorded = store.put([b"kept by a run"]) + held = store.put([b"kept by a message"]) + nested = store.put([b"kept inside a payload"]) + orphan = store.put([b"referred to by nothing"]) + + run_id = _run_row("ok", recorded["digest"]) + state = MemoryState() + state.update( + { + "cam.frame": held, + # A reference inside a json payload counts as much as a bare one. + "cam.report": {"latest": {"clip": nested}}, + "cam.count": 3, + } + ) + + try: + assert sweep_artifacts(store, state) == 1 + finally: + _forget(run_id) + + assert store.path(orphan["digest"]) is None + for kept in (recorded, held, nested): + assert store.path(kept["digest"]) is not None + + +def test_a_sweep_stands_aside_while_a_run_is_in_flight(store): + orphan = store.put([b"mid-run checkpoint"]) + run_id = _run_row("running") + try: + assert sweep_artifacts(store, MemoryState()) == 0 + finally: + _forget(run_id) + assert store.path(orphan["digest"]) is not None + + +def test_a_streamed_chunk_falls_out_once_the_message_moves_on(store): + """What makes a media stream affordable: only the current frame is held.""" + state = MemoryState() + first = store.put([b"frame one"], media_type="image/png") + state.update({"cam.frame": first}) + assert sweep_artifacts(store, state) == 0 + + second = store.put([b"frame two"], media_type="image/png") + state.update({"cam.frame": second}) + # Old enough to be swept: the sweep only spares what grace covers. + os.utime(store.path(first["digest"]), (time.time() - 10, time.time() - 10)) + assert sweep_artifacts(store, state, grace_s=5) == 1 + assert store.path(first["digest"]) is None + assert store.path(second["digest"]) is not None diff --git a/docs/code/connectors.md b/docs/code/connectors.md index aa111b1..c9df560 100644 --- a/docs/code/connectors.md +++ b/docs/code/connectors.md @@ -105,6 +105,21 @@ Keep the first version off the wire. A boolean setting the code checks before it sends — `artnet`'s `transmit` is the example — lets a flow be built and watched in the logs before anything physically moves. +## If the reading is bytes + +A camera or a microphone publishes a reference rather than the bytes: + +```python + async def poll(self): + jpeg = await asyncio.to_thread(self._grab) + return {"frame": self.save_artifact(jpeg, "f.jpg", media_type="image/jpeg")} +``` + +Place the node with an `image`-typed output port and a Media widget draws each +frame as it lands. `fluksio-connector-test-media` publishes test frames and +tones this way, so the whole path can be wired up with no camera in the room — +copy it if yours is a media device. + ## Try it without a device Give `poll` something predictable first and confirm the values reach the canvas diff --git a/docs/code/nodes.md b/docs/code/nodes.md index 375e6cc..249151b 100644 --- a/docs/code/nodes.md +++ b/docs/code/nodes.md @@ -110,6 +110,39 @@ Because the address is the content's hash, a sweep whose fifty configs share one preprocessed input stores it once, and a reference stays valid wherever the store is reachable from — including on another machine. +### Media + +Say what the bytes are and the port can be typed for them: + +```python +def process(speech): # an `audio` port + clip = fluksio.load_artifact(speech) + words = transcribe(clip) + return {"transcript": words} # a `str` port +``` + +```python +def process(camera_url): + for index, jpeg in enumerate(grab(camera_url)): # a generator + frame = fluksio.save_artifact( + jpeg, f"frame-{index:05d}.jpg", media_type="image/jpeg" + ) + frame["meta"] = {"seq": index} + yield {"frame": frame} # an `image` stream port +``` + +An `image`, `audio` or `video` port is an artifact reference whose media type +has to match, so a node declaring `audio` never receives a video by accident. +See [Payload types](../reference/payload-types.md#image-audio-video) for what +each carries and what rates are realistic. + +!!! warning "Emitted media is not kept; returned media is" + + Only what a node *returns* is recorded against its run. Frames yielded + along the way are replaced in state by the next one, and the artifact sweep + removes bytes nothing refers to any more — which is what stops a camera + filling the disk. If a particular frame matters, return it. + ## Printing `print` works and is captured. The first 16 KB per call is kept and shown in diff --git a/docs/concepts/flows.md b/docs/concepts/flows.md index 5aa7caf..a73db87 100644 --- a/docs/concepts/flows.md +++ b/docs/concepts/flows.md @@ -79,8 +79,8 @@ schema, so they all behave the same way. The full list is in ### Ports are typed A port declares a `dtype`: `float`, `int`, `str`, `bool`, `json`, `record`, -`list`, `series` or `artifact`. Every value that passes through is checked -against it. +`list`, `series`, `artifact`, or one of the media types `image`, `audio` and +`video`. Every value that passes through is checked against it. Types are not decoration. They are what lets the dashboard editor offer you only the messages a gauge can actually draw, and what lets the canvas refuse a @@ -88,7 +88,23 @@ binding before anything runs. See [Payload types](../reference/payload-types.md) Everything on the wire is JSON. Bytes — a checkpoint, an image, a model — travel as an `artifact`: the bytes go to a content-addressed store and the -message carries a small reference to them. +message carries a small reference to them. The media types are that same +reference, saying what kind of bytes are behind it. + +### Streaming ports + +A port marked `stream` produces repeatedly *during* one execution rather than +once at the end: a training loss, a progress fraction, a frame from a camera, a +second of speech. A node publishes on one by being a generator and yielding, or +by calling `fluksio.emit`. + +Each value is delivered to the nodes reading it, in the order it was produced — +so a recogniser slower than the microphone in front of it still sees every +chunk rather than only the newest. What is in state remains the latest value, +which is what everything else reads, and what a run keeps is the whole series. + +An `interval` on a streaming port thins what reaches the canvas without +thinning the run's record of it. ### Nodes are pure diff --git a/docs/interface/dashboards.md b/docs/interface/dashboards.md index 9e849a1..ca48fd1 100644 --- a/docs/interface/dashboards.md +++ b/docs/interface/dashboards.md @@ -34,6 +34,7 @@ and dragging is off. Picking a widget and editing its settings still works. | **Agenda** | `list` | upcoming items, e.g. from a calendar connector | | **Forecast** | `list` | a short outlook strip | | **Notification** | `record` | title, body and severity — what an alert channel writes | +| **Media** | `image`, `audio`, `video` | a camera frame, a clip; see *Media tiles* below | | **Clock** | — | the time, in a size a wall can read | Every widget carries a **title**, and **Show title** decides whether the panel @@ -91,6 +92,26 @@ it, so swapping the store is a change to one flow and nothing else. The answer also states what it was computed for, so an answer to a different question is ignored rather than two charts overwriting each other's picture. +## Media tiles + +A media widget draws what its message points at: a picture, a clip with +controls, a video. Media does not travel as a message — a reference to it does +— so the tile fetches the bytes behind whichever reference the message holds, +and redraws when a new one arrives. + +**Crop or fit** decides how a picture fills the tile. **Play as it arrives** +starts a clip by itself, though a browser only plays sound once somebody has +touched the page, so a screen nobody has tapped stays silent. + +Rate is the thing to get right. A frame every second or two is a glance at a +door, and works; through the portal, make that every few seconds. Live video is +not something to push through the message plane at all — put the camera's own +address in **Live stream** and the browser plays it from source, leaving the +messages to carry the occasional still that a flow can actually react to. + +Panels see media the same way, and only their own: a screen may fetch the bytes +its own tiles are showing and nothing else. + ## Dashboard settings Most of what a dashboard carries is a widget: a tile bound to a message. A diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index e00d15f..1c61895 100644 --- a/docs/reference/configuration.md +++ b/docs/reference/configuration.md @@ -128,6 +128,14 @@ warning into a refusal to start. | `FLOW_CPUS` | `0` | cores nodes that declare `resources` may be given; 0 works it out as every core but two, which are what keeps the engine answering while the machine is busy | | `FLOW_GPUS` | `0` | GPUs on this machine, each held by one node at a time. Not detected — say how many there are | | `OBS_RETENTION_DAYS` | `30` | how long metrics, events and run records are kept | +| `ARTIFACT_GC_INTERVAL_S` | `3600` | how often artifact bytes nothing refers to are swept away; 0 never sweeps | +| `ARTIFACT_GC_GRACE_S` | `3600` | how long a freshly written artifact is spared, whatever refers to it | + +An artifact is referred to by a run that recorded it or by a message currently +holding it; anything else is what a camera published four hours ago, and the +sweep is what keeps a flow streaming media from filling the disk. It stands +aside entirely while a run is in flight, since a node may store a checkpoint +long before it returns the reference to it. A node that declares nothing is not accounted against `FLOW_CPUS`; it runs on the shared pool and is given `FLOW_CPUS / FLOW_MAX_WORKERS` as a thread cap, so diff --git a/docs/reference/connector-contract.md b/docs/reference/connector-contract.md index 7f3a348..93cfec4 100644 --- a/docs/reference/connector-contract.md +++ b/docs/reference/connector-contract.md @@ -166,6 +166,29 @@ def write(self, **ports: Any) -> dict[str, Any] | None: A write is a command, not a value: set `idempotent = False` on the class so a redelivery after a crash does not undo a newer command that already landed. +## Devices whose readings are bytes + +A camera frame or a recorded clip is far too big to be a message, so a +connector publishes a reference to it instead: + +```python +async def poll(self) -> dict[str, Any] | None: + jpeg = await asyncio.to_thread(self._grab) + return { + "frame": self.save_artifact(jpeg, "frame.jpg", media_type="image/jpeg") + } +``` + +`save_artifact` stores the bytes and returns what an `image`, `audio` or +`video` port carries — the media type has to match the port's type. It only +works once the node has started, since the store is the engine's and is handed +over then. + +Each reading is a new artifact, which the poll loop publishes because its +digest differs from the last. Set `poll_interval` to what somebody actually +wants to look at: a frame every second or two is a glance, and live video +belongs on the camera's own stream rather than in the graph. + ## Lifecycle ```python diff --git a/docs/reference/payload-types.md b/docs/reference/payload-types.md index 4e69969..0b0b1e5 100644 --- a/docs/reference/payload-types.md +++ b/docs/reference/payload-types.md @@ -99,6 +99,39 @@ stays valid wherever the store is reachable from, including on another machine. Node code produces one with `fluksio.save_artifact` and opens one with `fluksio.load_artifact`. See [Writing node code](../code/nodes.md#bytes-artifacts). +### `image`, `audio`, `video` + +The same reference, narrowed to a kind of media by its `media_type`. + +```json +{"digest": "sha256:…", "size": 61344, "media_type": "image/jpeg", "name": "frame.jpg", + "meta": {"width": 1280, "height": 720, "seq": 41}} +``` + +An `audio` port takes `audio/*` and refuses anything else, so a speech +recogniser declares what it eats rather than taking any bytes at all and +finding out. An `artifact` port still accepts all three — media narrows +artifact, not the other way round. + +`meta` is optional and nothing here reads it: sample rates, dimensions and +sequence numbers are for whoever consumes the media. + +Bytes still never travel as a message. A camera publishes one reference per +frame and a microphone one per chunk — which makes a media stream an ordinary +[streaming port](../concepts/flows.md#streaming-ports), and each frame an +artifact. What that costs is worth knowing before pointing a camera at it: + +| Rate | Where it works | +|---|---| +| A clip a second (speech) | anywhere, including through the portal | +| A frame every second or two (a glance at a door) | locally; through the portal, every few seconds | +| Live video, 15–30 fps | not here — see below | + +Real-time video is not a message-plane problem: every frame would be an +artifact, an event and a fetch. Point a media widget's **stream URL** at +whatever the camera already serves and the browser plays it from source; the +messages then carry the occasional still, and the flow reacts to those. + ### `json` Anything JSON-serializable. The escape hatch, and the right answer when a @@ -129,6 +162,7 @@ carry one of them literally is asking for a value this engine reads as a name. | Notification | `record` | | Value | anything | | Icon | weather strings, booleans and numbers alike | +| Media | `image`, `audio`, `video`, `artifact` | | Clock, Text | nothing — they bind to no message | Enforced on the server as well as in the editor. diff --git a/frontend/scripts/capture-screenshots.mjs b/frontend/scripts/capture-screenshots.mjs index 1666609..d66c41f 100644 --- a/frontend/scripts/capture-screenshots.mjs +++ b/frontend/scripts/capture-screenshots.mjs @@ -89,6 +89,7 @@ for (const theme of ["light", "dark"]) { await captureFlows(page, dir) await captureDashboards(page, dir) + await captureMedia(page, dir) await captureRuns(page, dir) await context.close() @@ -192,6 +193,41 @@ async function captureDashboards(page, dir) { await page.screenshot({ path: `${dir}/app-panel.png` }) } +/** + * A media tile drawing what a camera published, where there is one. + * + * Skipped unless the media example is seeded (root `make seed-example-media`), + * since it is the one shot that needs a source of frames. The bytes arrive as + * a blob — the tile fetches them with the session's credential, which no `img` + * could carry on its own — so a `blob:` source is the proof the whole path ran + * rather than that a picture is merely present. + */ +async function captureMedia(page, dir) { + const answer = await page.goto(`${APP_URL}/view/camera`, { + waitUntil: "networkidle", + }) + if (!answer?.ok()) return + + const picture = page.locator("img[alt='Test camera']") + try { + await picture.waitFor({ timeout: 15000 }) + await page.waitForFunction( + () => + document + .querySelector("img[alt='Test camera']") + ?.src?.startsWith("blob:") ?? false, + { timeout: 15000 }, + ) + } catch { + console.warn( + " media tile drew nothing — is `make seed-example-media` run?", + ) + return + } + await page.waitForTimeout(500) + await page.screenshot({ path: `${dir}/app-media.png` }) +} + /** * The flow editor, empty-handed if the instance has no flows yet: seeds one * with a node so the canvas and the node panel are both worth looking at. diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index 09dbe2c..7ce7f7b 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -373,7 +373,7 @@ export const ChannelSchema = { export const DTypeSchema = { type: 'string', - enum: ['float', 'int', 'str', 'bool', 'json', 'series', 'record', 'list', 'artifact'], + enum: ['float', 'int', 'str', 'bool', 'json', 'series', 'record', 'list', 'artifact', 'image', 'audio', 'video'], title: 'DType', description: `Serializable payload types. @@ -388,7 +388,14 @@ them. That keeps everything on the wire JSON, which is what the state backend, the queue and the worker protocol all rely on, and it means a thirty-megabyte checkpoint never sits in Redis. Inline codecs would only be needed for payloads too small to be worth a round trip, and nothing asks -for that yet.` +for that yet. + +\`\`image\`\`, \`\`audio\`\` and \`\`video\`\` are that same reference narrowed to a +media family, so a speech recogniser declares what it eats rather than +taking any bytes at all and finding out. Bytes still never travel as a +message: a camera publishes one reference per frame, a microphone one per +chunk. A reference may carry a \`\`meta\`\` dict — sample rate, dimensions, a +sequence number — which nothing here interprets.` } as const; export const DashboardDef_InputSchema = { @@ -3327,7 +3334,7 @@ export const WidgetDefSchema = { }, type: { type: 'string', - enum: ['stat', 'gauge', 'chart', 'markdown', 'agenda', 'notification', 'bar', 'icon', 'forecast', 'clock', 'button', 'switch', 'slider', 'input', 'dropdown', 'color'], + enum: ['stat', 'gauge', 'chart', 'markdown', 'agenda', 'notification', 'bar', 'icon', 'forecast', 'clock', 'media', 'button', 'switch', 'slider', 'input', 'dropdown', 'color'], title: 'Type' }, title: { diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index a9103c9..68f02c8 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -65,6 +65,10 @@ export class ArtifactsService { /** * Put Artifact * Store the request body and answer with the reference to it. + * + * Spooled to disk as it arrives rather than buffered: a video segment is as + * legitimate a body here as a checkpoint, and neither should have to fit in + * memory twice. * @param data The data for the request. * @param data.name * @param data.mediaType @@ -87,9 +91,15 @@ export class ArtifactsService { /** * Get Artifact - * Stream one artifact back. + * Serve one artifact back. + * + * The caller passes the media type off the reference it holds, which is what + * lets a browser play a clip rather than download it; the store itself keeps + * only bytes. Ranged requests are answered because an audio or video element + * scrubbing through a file asks for them. * @param data The data for the request. * @param data.digest + * @param data.mediaType * @returns unknown Successful Response * @throws ApiError */ @@ -100,6 +110,9 @@ export class ArtifactsService { path: { digest: data.digest }, + query: { + media_type: data.mediaType + }, errors: { 422: 'Validation Error' } diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index d011026..f0d9507 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -182,8 +182,15 @@ export type DeadLetter = { * thirty-megabyte checkpoint never sits in Redis. Inline codecs would only be * needed for payloads too small to be worth a round trip, and nothing asks * for that yet. + * + * ``image``, ``audio`` and ``video`` are that same reference narrowed to a + * media family, so a speech recogniser declares what it eats rather than + * taking any bytes at all and finding out. Bytes still never travel as a + * message: a camera publishes one reference per frame, a microphone one per + * chunk. A reference may carry a ``meta`` dict — sample rate, dimensions, a + * sequence number — which nothing here interprets. */ -export type DType = 'float' | 'int' | 'str' | 'bool' | 'json' | 'series' | 'record' | 'list' | 'artifact'; +export type DType = 'float' | 'int' | 'str' | 'bool' | 'json' | 'series' | 'record' | 'list' | 'artifact' | 'image' | 'audio' | 'video'; /** * Something wired into this flow that is not a node in it. @@ -1156,7 +1163,7 @@ export type ValidationResult = { */ export type WidgetDef = { id: string; - type: 'stat' | 'gauge' | 'chart' | 'markdown' | 'agenda' | 'notification' | 'bar' | 'icon' | 'forecast' | 'clock' | 'button' | 'switch' | 'slider' | 'input' | 'dropdown' | 'color'; + type: 'stat' | 'gauge' | 'chart' | 'markdown' | 'agenda' | 'notification' | 'bar' | 'icon' | 'forecast' | 'clock' | 'media' | 'button' | 'switch' | 'slider' | 'input' | 'dropdown' | 'color'; title?: string; layout?: { [key: string]: Placement; @@ -1166,7 +1173,7 @@ export type WidgetDef = { }; }; -export type type = 'stat' | 'gauge' | 'chart' | 'markdown' | 'agenda' | 'notification' | 'bar' | 'icon' | 'forecast' | 'clock' | 'button' | 'switch' | 'slider' | 'input' | 'dropdown' | 'color'; +export type type = 'stat' | 'gauge' | 'chart' | 'markdown' | 'agenda' | 'notification' | 'bar' | 'icon' | 'forecast' | 'clock' | 'media' | 'button' | 'switch' | 'slider' | 'input' | 'dropdown' | 'color'; export type WorkerInfo = { name: string; @@ -1202,6 +1209,7 @@ export type ArtifactsPutArtifactResponse = (ArtifactRef); export type ArtifactsGetArtifactData = { digest: string; + mediaType?: string; }; export type ArtifactsGetArtifactResponse = (unknown); diff --git a/frontend/src/components/Dashboard/MediaWidget.tsx b/frontend/src/components/Dashboard/MediaWidget.tsx new file mode 100644 index 0000000..97976c6 --- /dev/null +++ b/frontend/src/components/Dashboard/MediaWidget.tsx @@ -0,0 +1,153 @@ +import { useEffect, useState } from "react" + +import { OpenAPI } from "@/client" +import { apiToken } from "@/lib/portal" +import { cn } from "@/lib/utils" +import { useBoundValue } from "./dataContext" +import { config, text } from "./ui/core/config" +import type { WidgetProps } from "./widgets" + +/** An artifact reference, as a message carries one. */ +type MediaRef = { + digest?: string + media_type?: string + name?: string + size?: number +} + +const isRef = (value: unknown): value is MediaRef => + typeof value === "object" && + value !== null && + typeof (value as MediaRef).digest === "string" && + (value as MediaRef).digest!.startsWith("sha256:") + +/** + * A local URL for an artifact's bytes, refreshed whenever the digest changes. + * + * Not the endpoint itself: `/artifacts/{digest}` takes a bearer token, and no + * `` or `