simvx.core.assets.server

AssetServer: orchestrates Sources, Loaders, and a worker pool.

Public entry point for asset loading. Routes a URI through the right

class:

~simvx.core.assets.source.Source and :class:~simvx.core.assets.loaders.base.Loader, runs the I/O on a thread pool, and marshals completion notifications back onto the main thread via :meth:flush (called once per SceneTree.process(dt) tick).

Usage::

server = AssetServer.instance()
handle = server.load("pkg://game/textures/player.png")
handle.completed.connect(lambda h: print("loaded:", h.state))

batch = server.load_folder("pkg://game/textures/")
batch.item_completed.connect(lambda h: print("loaded one:", h.uri))
batch.completed.connect(lambda b: print("all done"))

The constructor is private-by-convention; use :meth:AssetServer.instance to obtain the engine-wide singleton, or instantiate directly for isolated tests.

Module Contents

Classes

AssetServer

Async typed-asset orchestrator with byte-budget caches per loader.

Data

__all__

API

simvx.core.assets.server.__all__

[‘AssetServer’]

class simvx.core.assets.server.AssetServer(*, max_workers: int = 2)

Async typed-asset orchestrator with byte-budget caches per loader.

Initialization

classmethod instance() → simvx.core.assets.server.AssetServer

Return the engine-wide singleton, creating it on first access.

classmethod reset_instance() → None

Tear down and clear the singleton (for tests).

shutdown() → None

register_source(source: simvx.core.assets.source.Source) → None

Install source for its declared scheme.

register_loader(loader: simvx.core.assets.loaders.base.Loader) → None

Install a typed loader. Order of registration is the lookup order.

get_source(scheme: str) → simvx.core.assets.source.Source

load(uri: str, *, cache: bool = True, loader: simvx.core.assets.loaders.base.Loader | None = None) → simvx.core.assets.handle.Handle

Submit uri for asynchronous loading.

Returns a :class:Handle that exposes progress, state, errors, and a completed Signal. Uses the registered Source for the URI’s scheme and the first Loader whose claims(uri) returns True (or the explicit loader= override). When cache is True, both lookup and store go through the loader’s LRU.

load_group(uris: collections.abc.Iterable[str], *, cache: bool = True, loader: simvx.core.assets.loaders.base.Loader | None = None) → simvx.core.assets.handle.BatchHandle

Load every URI in uris concurrently. Returns a :class:BatchHandle.

load_folder(uri: str, *, cache: bool = True, loader: simvx.core.assets.loaders.base.Loader | None = None) → simvx.core.assets.handle.BatchHandle

Enumerate uri via the Source’s list() and load each child.

Sources that cannot list (e.g. HTTP) raise :class:NotImplementedError; use :meth:load_manifest for those.

load_manifest(manifest_uri: str, *, cache: bool = True, loader: simvx.core.assets.loaders.base.Loader | None = None) → simvx.core.assets.handle.BatchHandle

Load the URI list from a JSON manifest, then load each entry.

The manifest must be a JSON array of URI strings, e.g.::

["pkg://game/level1/floor.png", "https://cdn.example.com/level1.ogg"]

This call blocks briefly to fetch + parse the manifest, then returns a BatchHandle for the listed assets.

flush() → None

Drain the completion queue, dispatching each on the calling thread.

SceneTree.process(dt) calls this once per frame so the completed Signal on each Handle / BatchHandle fires on the main thread, ≤1 frame after the worker actually finished.

invalidate(uri: str) → None

Drop uri from every loader’s cache.

clear_caches() → None

Drop every cached asset across every loader.