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 `