AssetServer¶

simvx.core.assets.AssetServer is the engine-wide asynchronous asset loader. It resolves a URI through a pluggable Source, parses bytes through a typed Loader, runs the I/O on a thread pool, and dispatches completion onto the main thread once per frame via SceneTree.tick().

Use it for:

single asset loads where you want a Signal/poll API instead of a blocking call;
loading screens (group / folder / manifest with combined progress);
network assets (HTTP with ETag-based revalidation);
web exports: the same code runs on Pyodide via a pyfetch shim.

For synchronous package-resource access (e.g. small config files, shipped fonts) prefer Resource / importlib.resources: no thread pool, no flush, no Handle.

Quick start¶

from simvx.core.assets import AssetServer

server = AssetServer.instance()

# One asset
handle = server.load("pkg://game/textures/player.png")
handle.completed.connect(lambda h: print(h.state, h.error))

# Many assets
batch = server.load_folder("pkg://game/textures/")
batch.item_completed.connect(lambda h: hud.set_status(f"loaded: {h.uri}"))
batch.completed.connect(lambda b: hud.show_main_menu())

The SceneTree calls AssetServer.instance().flush() once per tick(dt), which transitions handles to their final state and fires completed signals on the main thread.

URI schemes¶

Scheme	Source class	List support	Notes
`pkg://pkg/path`	`PkgSource`	yes	`importlib.resources`. Immutable per session.
`file:///abs/path`	`FileSource`	yes	mtime-based version hint.
`http(s)://...`	`HttpSource`	no	ETag revalidation. Web override uses `pyfetch`.
`mem://key`	`MemSource`	yes	In-memory; tests / dynamic data.

A bare relative path ("my/file.txt") routes through FileSource.

Handles¶

server.load(uri) returns a Handle:

Attr / method	Purpose
`state`	`”pending”
`progress`	`0.0` – `1.0`. Real for HTTP; `0→1` in one tick for pkg/file.
`error`	`BaseException
`is_done`	`True` once in a terminal state.
`result(timeout=None)`	Block until done; raise `error` on failure.
`cancel()`	Mark for cancellation; worker stops on its next progress check.
`completed`	`Signal(Handle)`: emits once on terminal state.

server.load_group([...]), server.load_folder(uri), and server.load_manifest(uri) return a BatchHandle:

Attr / method	Purpose
`total`, `completed_count`, `failed_count`, `cancelled_count`	Counts.
`progress`	Sum of children’s progress / total (partial credit for in-flight HTTP).
`handles`	List of child Handles.
`is_done`	True when every child is done.
`results(timeout=None)`	`dict[uri, value]`; blocks until all done; raises if any failed.
`cancel()`	Cancel every unfinished child.
`item_completed`	`Signal(Handle)`: emits per child as it finishes.
`completed`	`Signal(BatchHandle)`: emits once when all children are done.

Cache and freshness¶

Each typed loader owns a byte-budget LRU cache keyed on (uri, version_hint). The version hint is provided by the Source: mtime for files, ETag (or Last-Modified fallback) for HTTP, and None for pkg://. Defaults to 256 MB per loader.

# Per-call opt out (fresh fetch, no cache write)
h = server.load("https://cdn.example.com/level.json", cache=False)

# Explicit eviction
server.invalidate("https://cdn.example.com/level.json")
server.clear_caches()

The HTTP source revalidates with a HEAD request: if the server’s ETag matches the cached version, the in-flight load resolves from cache; otherwise it refetches. This makes auto-update straightforward: just ship the ETag.

Typed loaders¶

from simvx.core.assets.loaders.base import Loader

class CsvLoader(Loader):
    suffixes = (".csv",)

    def parse(self, raw: bytes, uri: str):
        import csv
        return list(csv.reader(raw.decode("utf-8").splitlines()))

server.register_loader(CsvLoader())

handle = server.load("pkg://game/data/items.csv")

The first registered loader whose claims(uri) returns True wins; the fallback BytesLoader returns raw bytes if no typed loader matches.

Built-in: BytesLoader (raw), JsonLoader (.json → Python dict/list).

Threading¶

Workers run on a small ThreadPoolExecutor (2 workers default, configurable via AssetServer(max_workers=...)). Each worker enqueues (handle, kind, payload) onto a thread-safe deque; the SceneTree drains it on the main thread once per frame. Handler code running off completed always runs on the main thread, ≤1 frame after the worker actually finished.

Don’t subscribe to completed from a worker thread; subscribe from node code that runs on the main thread.

Web¶

simvx.web.app.WebApp calls simvx.web.assets.install_web_sources() during construction, which swaps the desktop HttpSource for WebHttpSource: a pyodide.http.pyfetch shim. User code is identical to desktop:

batch = server.load_group([
    "https://cdn.example.com/textures/0.png",
    "https://cdn.example.com/textures/1.png",
])
batch.completed.connect(lambda b: print("all loaded"))

The SceneTree’s per-frame flush still drives completion; the deque just gets fed by the JS event loop instead of a real thread.

Loading-screen pattern¶

class LoadingScreen(Control):
    def on_ready(self):
        self.batch = AssetServer.instance().load_folder("pkg://game/level1/")
        self.batch.item_completed.connect(self._on_item)
        self.batch.completed.connect(self._on_done)

    def _on_item(self, handle):
        self.status_label.text = (
            f"Loading {self.batch.completed_count}/{self.batch.total}: "
            f"{handle.uri.rsplit('/', 1)[-1]}"
        )
        self.progress_bar.value = self.batch.progress

    def _on_done(self, batch):
        self.tree.change_scene(MainGame())

Caveats¶

result(timeout=...) blocks the calling thread. If you call it from a node lifecycle hook that runs on the main thread (where flush also runs), you’ll deadlock waiting for a completion that can’t arrive. Prefer the completed signal for in-engine code; result() is for tests and tools.
MemSource versions by content length, so updating a key in place with the same number of bytes will hit the cache. Tests that rely on freshness should pass cache=False or call server.invalidate(uri).