"""Texture loading and bindless descriptor array management.
One canonical ``TextureManager`` serves both the Vulkan and web backends:
the only thing that differs between them is the ``_TextureRegistrar`` that
turns RGBA pixels into a backend-specific texture id. On the web path the
manager also retains pixel data so ``prepare_2d_overlays`` can re-ship it
over the drain channel; on the desktop path that's skipped (``retain_pixels=False``)
to avoid doubling VRAM.
"""
import hashlib
import io
import logging
import struct
import zlib
from pathlib import Path
from typing import Any, Protocol
import numpy as np
__all__ = ["TextureManager"]
log = logging.getLogger(__name__)
# Accepted texture source types at the public API surface. Kept as a tuple
# so it doubles as an isinstance() argument.
TextureSource = str | Path | bytes | np.ndarray
# Formats already warned about (deduped one-time WARNING per (path, format)).
_warned_compressed: set[tuple[str, int]] = set()
# Encoded raster suffixes a native-decode backend can hand straight to the
# platform image decoder (browser ``createImageBitmap``). Block-compressed
# containers (.dds/.ktx2) are deliberately excluded: they are not raster images
# and keep their dedicated transcode/block paths.
_ENCODED_RASTER_SUFFIXES = frozenset({".jpg", ".jpeg", ".png", ".webp", ".avif", ".bmp", ".gif"})
class _TextureRegistrar(Protocol):
"""Minimal contract every rendering backend exposes for pixel uploads.
Satisfied by ``Engine`` (Vulkan), ``WebRenderer`` (WebGPU drain channel),
and the tiny ``_PendingTextureRenderer`` shim used by 2D-only web exports.
"""
def upload_texture_pixels(self, pixels: np.ndarray, width: int, height: int) -> int: ...
class _EncodedImageRegistrar(Protocol):
"""Optional capability: hand RAW encoded image bytes to a native decoder.
A backend whose platform already ships JPEG/PNG/WebP/AVIF decoders (the web
backend: the browser decodes via ``createImageBitmap``) advertises
``supports_encoded_image_upload = True`` and implements
:meth:`upload_encoded_image`, so the shared ``TextureManager`` skips CPU
decode entirely. Desktop's ``Engine`` does NOT implement this, so it decodes
via PIL exactly as before (byte-identical). The decode resolves one frame
late, mirroring the existing mesh-texture / compressed-texture drain paths.
"""
supports_encoded_image_upload: bool
def upload_encoded_image(self, data: bytes, *, srgb: bool, filter: str) -> int: ...
class _CompressedRegistrar(Protocol):
"""Sibling contract for block-compressed (BC) uploads.
Distinct from :class:`_TextureRegistrar`: the blocks path has its OWN
contract (a non-empty ``list[bytes]`` of mips + an 8/16-byte block size) and
never goes through the strict RGBA8 (H, W, 4) uint8 validation. Satisfied by
the desktop ``Engine``; web registrars do not implement it yet.
"""
def supports_compressed_format(self, vk_format: int) -> bool: ...
def compressed_caps(self) -> dict[str, bool]: ...
def format_supported(self, vk_format: int) -> bool: ...
def upload_texture_blocks(
self, blocks: list[bytes], width: int, height: int, fmt: int, block_size: int,
) -> int: ...
def _sniff_encoded_raster(data: bytes) -> bool:
"""True when ``data`` begins with a browser-decodable raster magic.
Recognises PNG / JPEG / GIF / BMP / WebP / AVIF. Used by the native-decode
capability path to decide whether to ship raw bytes to the platform decoder.
DDS / KTX2 are intentionally NOT matched here: they are routed to the
compressed path before this is ever reached.
"""
if len(data) < 12:
return False
if data[:8] == b"\x89PNG\r\n\x1a\n": # PNG
return True
if data[:3] == b"\xff\xd8\xff": # JPEG
return True
if data[:6] in (b"GIF87a", b"GIF89a"): # GIF
return True
if data[:2] == b"BM": # BMP
return True
if data[:4] == b"RIFF" and data[8:12] == b"WEBP": # WebP
return True
if data[4:8] == b"ftyp" and data[8:12] in (b"avif", b"avis"): # AVIF
return True
return False
def _decode_png_rgba(data: bytes) -> tuple[np.ndarray, int, int] | None:
"""Pure-python decoder for filter-0 RGBA PNGs: used when PIL isn't available
(e.g. slim Pyodide installs). Returns None if the bytes aren't such a PNG.
"""
if data[:8] != b"\x89PNG\r\n\x1a\n":
return None
pos = 8
width = height = 0
idat_chunks: list[bytes] = []
while pos < len(data):
(length,) = struct.unpack_from(">I", data, pos)
ctype = data[pos + 4:pos + 8]
chunk_data = data[pos + 8:pos + 8 + length]
pos += 12 + length
if ctype == b"IHDR":
width, height = struct.unpack_from(">II", chunk_data, 0)
elif ctype == b"IDAT":
idat_chunks.append(chunk_data)
elif ctype == b"IEND":
break
if not idat_chunks or width == 0:
return None
raw = zlib.decompress(b"".join(idat_chunks))
stride = 1 + width * 4
pixels = np.empty((height, width, 4), dtype=np.uint8)
for y in range(height):
row_start = y * stride + 1 # skip filter byte
pixels[y] = np.frombuffer(raw[row_start:row_start + width * 4], dtype=np.uint8).reshape(width, 4)
return pixels, width, height
def _load_pixels_from_bytes(data: bytes) -> tuple[np.ndarray, int, int]:
"""Decode PNG/JPG bytes to RGBA uint8 pixels. Prefers PIL when installed,
falls back to the pure-python PNG decoder otherwise.
"""
try:
from PIL import Image # type: ignore[import-not-found]
except ImportError:
decoded = _decode_png_rgba(data)
if decoded is None:
raise ValueError("Texture bytes are not a filter-0 RGBA PNG and PIL is unavailable") # noqa: B904
return decoded
img = Image.open(io.BytesIO(data)).convert("RGBA")
pixels = np.ascontiguousarray(np.array(img, dtype=np.uint8))
return pixels, img.width, img.height
def _load_pixels_from_path(path: Path) -> tuple[np.ndarray, int, int]:
"""Decode a texture on disk. PIL is required for JPEG; for PNG we fall
back to the pure-python decoder so headless / Pyodide installs work.
"""
try:
from PIL import Image # type: ignore[import-not-found]
except ImportError:
return _load_pixels_from_bytes(path.read_bytes())
img = Image.open(str(path)).convert("RGBA")
pixels = np.ascontiguousarray(np.array(img, dtype=np.uint8))
return pixels, img.width, img.height
def _release_texture(manager_ref: Any, idx: int,
cache_key: str, source_id: int) -> None:
"""Top-level weakref.finalize callback: delegates to manager.release."""
manager = manager_ref()
if manager is None:
return
try:
manager.release(idx, cache_key=cache_key, source_id=source_id)
except Exception:
log.exception("TextureManager.release failed during finalize")
[docs]
class TextureManager:
"""Loads textures via a backend registrar and caches by source identity.
Desktop Vulkan: ``TextureManager(engine)``: pixels uploaded and forgotten.
Web: ``TextureManager(renderer, retain_pixels=True)``: pixels retained so
``prepare_2d_overlays`` can re-ship them over the drain channel on demand.
"""
def __init__(self, registrar: _TextureRegistrar, *, retain_pixels: bool = False) -> None:
self._registrar = registrar
self._retain_pixels = retain_pixels
self._cache: dict[str, int] = {} # source key → tex_id
self._sizes: dict[int, tuple[int, int]] = {} # tex_id → (w, h)
self._pixels_by_id: dict[int, np.ndarray] = {} # only populated when retain_pixels
# weakref.finalize handles, keyed by tex_id, so array-sourced textures
# reclaim their backend slot when the owning ndarray is GC'd. Only
# populated for ``load_from_array``: file/bytes sources are content-
# hashed and outlive the caller's reference intentionally.
self._finalizers: dict[int, Any] = {}
# ------------------------------------------------------------------
# Canonical entry point
# ------------------------------------------------------------------
[docs]
def resolve(self, source: TextureSource | None, *,
filter: str = "linear",
premultiply_alpha: bool = False,
colour_space: str = "srgb",
mipmaps: bool = False) -> int:
"""Resolve any supported texture source to a backend texture index.
Returns -1 for ``None``, empty strings, or sources that cannot be
resolved (e.g. a path that does not exist). All callers that accept
a user-provided ``texture`` property should go through this method.
Supported sources:
* ``str`` / ``pathlib.Path``: file on disk (PNG / JPG / ...)
* ``bytes``: raw encoded image data (PNG / JPG)
* ``numpy.ndarray``: RGBA uint8 pixels, shape ``(H, W, 4)``
``filter`` selects the sampler bound at the bindless slot:
``"linear"`` (default) or ``"nearest"``. Each (source, filter,
premultiply_alpha) tuple gets its own slot so the same source can be
drawn smoothly somewhere, crisply elsewhere, with or without alpha
premultiplication, all without re-uploading pixels.
Args:
filter: Sampler filter mode: ``"linear"`` or ``"nearest"``.
premultiply_alpha: When True, multiply RGB by alpha before GPU
upload (matches ``image_loader.premultiply_alpha_rgba``).
Fixes halo artefacts on alpha-blended PNGs whose transparent
pixels carry stale RGB. Default False: existing visual
snapshots use straight alpha.
mipmaps: When True, ask the registrar to generate a full runtime
mip chain (desktop blit chain; web fullscreen-sample
chain). Registrars without the capability fall back to
a single mip. Default False keeps every existing upload
byte-identical.
"""
if source is None:
return -1
if isinstance(source, np.ndarray):
return self.load_from_array(source, filter=filter, premultiply_alpha=premultiply_alpha,
colour_space=colour_space, mipmaps=mipmaps)
if isinstance(source, bytes | bytearray | memoryview):
return self.load_from_bytes(bytes(source), filter=filter, premultiply_alpha=premultiply_alpha,
colour_space=colour_space, mipmaps=mipmaps)
if isinstance(source, str) and source == "":
return -1
return self.load_if_exists(source, filter=filter, premultiply_alpha=premultiply_alpha,
colour_space=colour_space, mipmaps=mipmaps)
# ------------------------------------------------------------------
# Individual loaders
# ------------------------------------------------------------------
[docs]
def load(self, path: str | Path, *,
filter: str = "linear",
premultiply_alpha: bool = False,
colour_space: str = "srgb",
mipmaps: bool = False) -> int:
"""Load a texture from disk. Cached by (resolved path, filter, premul, colour_space).
``.dds`` and ``.ktx2`` files take the block-compressed path (BC1-BC7) on
backends that expose a compressed registrar; everything else takes the
RGBA8 path. ``colour_space`` (``"srgb"`` default / ``"linear"``) selects
the sampled view format on the RGBA8 path; compressed files carry their
own sRGB flag in the container format.
"""
resolved = Path(path).resolve()
suffix = resolved.suffix.lower()
if suffix == ".dds":
key = self._cache_key(f"dds:{filter}:{resolved}", premultiply_alpha)
if key in self._cache:
return self._cache[key]
idx = self._load_dds(resolved.read_bytes(), str(resolved), filter=filter)
self._cache[key] = idx
return idx
if suffix == ".ktx2":
key = self._cache_key(f"ktx2:{filter}:{resolved}", premultiply_alpha)
if key in self._cache:
return self._cache[key]
idx = self._load_ktx2(resolved.read_bytes(), str(resolved), filter=filter)
self._cache[key] = idx
return idx
key = self._cache_key(f"path:{filter}:{resolved}", premultiply_alpha, colour_space, mipmaps)
if key in self._cache:
return self._cache[key]
# Native-decode capability: hand the raw encoded bytes to the backend's
# platform decoder (browser) instead of CPU-decoding here. Skipped when
# premultiply_alpha is requested (the native path can't premultiply for
# us) so those sources still take the CPU decode + premultiply route.
if not premultiply_alpha and suffix in _ENCODED_RASTER_SUFFIXES:
idx = self._try_encoded_upload(resolved.read_bytes(), filter=filter, colour_space=colour_space,
mipmaps=mipmaps)
if idx is not None:
self._cache[key] = idx
log.debug("TextureManager: %s [native-decode %s cs=%s] -> index %d",
resolved.name, filter, colour_space, idx)
return idx
pixels, width, height = _load_pixels_from_path(resolved)
if premultiply_alpha:
from ..assets.image_loader import premultiply_alpha_rgba
pixels = premultiply_alpha_rgba(pixels)
idx = self._register(pixels, width, height, filter=filter, colour_space=colour_space, mipmaps=mipmaps)
self._cache[key] = idx
log.debug("TextureManager: %s [%s premul=%s cs=%s] → index %d",
resolved.name, filter, premultiply_alpha, colour_space, idx)
return idx
[docs]
def load_from_bytes(self, data: bytes, *,
filter: str = "linear",
premultiply_alpha: bool = False,
colour_space: str = "srgb",
mipmaps: bool = False) -> int:
"""Load a texture from in-memory image bytes. Cached by (content, filter, premul).
A leading ``b'DDS '`` magic or the 12-byte KTX2 identifier routes to the
block-compressed path; PNG/JPG bytes take the RGBA8 path unchanged.
"""
if data[:4] == b"DDS ":
key = self._cache_key(f"dds:{filter}:{hashlib.sha256(data).hexdigest()}", premultiply_alpha)
if key in self._cache:
return self._cache[key]
idx = self._load_dds(data, "<bytes>", filter=filter)
self._cache[key] = idx
return idx
if data[:12] == b"\xab\x4b\x54\x58\x20\x32\x30\xbb\x0d\x0a\x1a\x0a":
key = self._cache_key(f"ktx2:{filter}:{hashlib.sha256(data).hexdigest()}", premultiply_alpha)
if key in self._cache:
return self._cache[key]
idx = self._load_ktx2(data, "<bytes>", filter=filter)
self._cache[key] = idx
return idx
key = self._cache_key(f"bytes:{filter}:{hashlib.sha256(data).hexdigest()}", premultiply_alpha, colour_space,
mipmaps)
if key in self._cache:
return self._cache[key]
# Native-decode capability for embedded glTF images (the .glb JPEG case):
# ship raw encoded bytes to the browser decoder. Same premultiply caveat
# as the path branch. _sniff_encoded_raster excludes DDS/KTX2 (already
# routed above) and anything the magic doesn't recognise as a raster image.
if not premultiply_alpha and _sniff_encoded_raster(data):
idx = self._try_encoded_upload(data, filter=filter, colour_space=colour_space, mipmaps=mipmaps)
if idx is not None:
self._cache[key] = idx
log.debug("TextureManager: embedded [native-decode %s cs=%s] -> index %d",
filter, colour_space, idx)
return idx
try:
pixels, width, height = _load_pixels_from_bytes(data)
except ValueError:
log.warning("Failed to decode texture from bytes")
return -1
if premultiply_alpha:
from ..assets.image_loader import premultiply_alpha_rgba
pixels = premultiply_alpha_rgba(pixels)
idx = self._register(pixels, width, height, filter=filter, colour_space=colour_space, mipmaps=mipmaps)
self._cache[key] = idx
log.debug("TextureManager: embedded %dx%d [%s premul=%s cs=%s] → index %d",
width, height, filter, premultiply_alpha, colour_space, idx)
return idx
[docs]
def load_from_array(self, pixels: np.ndarray, *,
filter: str = "linear",
premultiply_alpha: bool = False,
colour_space: str = "srgb",
mipmaps: bool = False) -> int:
"""Upload an RGBA uint8 ndarray of shape ``(H, W, 4)``.
Cached by ``(id(pixels), filter, premultiply_alpha)``: the same
ndarray with the same triple returns the same index; any difference
allocates a fresh slot so a single source can be drawn through
multiple sampler/premultiplication paths without collision.
"""
if pixels.ndim != 3 or pixels.shape[2] != 4:
raise ValueError(f"Expected RGBA pixels with shape (H, W, 4); got {pixels.shape}")
if pixels.dtype != np.uint8:
raise ValueError(f"Expected uint8 pixels; got {pixels.dtype}")
h, w = pixels.shape[:2]
key = self._cache_key(f"array:{filter}:{id(pixels)}:{w}x{h}:{pixels.dtype}",
premultiply_alpha, colour_space, mipmaps)
if key in self._cache:
return self._cache[key]
contig = np.ascontiguousarray(pixels)
if premultiply_alpha:
from ..assets.image_loader import premultiply_alpha_rgba
contig = premultiply_alpha_rgba(contig)
idx = self._register(contig, w, h, filter=filter, colour_space=colour_space, mipmaps=mipmaps)
self._cache[key] = idx
# Hook the lifetime of the *source* ndarray so the texture slot is
# reclaimed when the caller's reference drops. Engine.unregister_texture
# pushes the slot back onto its free list; TextureManager.release drops
# its own bookkeeping. The finalizer holds only a weakref to self so
# the manager itself can still be GC'd.
import weakref
if idx not in self._finalizers:
self._finalizers[idx] = weakref.finalize(
pixels, _release_texture,
weakref.ref(self), idx, key, id(pixels),
)
return idx
[docs]
def load_if_exists(self, path: str | Path, *,
filter: str = "linear",
premultiply_alpha: bool = False,
colour_space: str = "srgb",
mipmaps: bool = False) -> int:
"""Load a texture if the file exists. Returns -1 if not found."""
p = Path(path)
if not p.exists():
return -1
return self.load(p, filter=filter, premultiply_alpha=premultiply_alpha, colour_space=colour_space,
mipmaps=mipmaps)
@staticmethod
def _cache_key(base: str, premultiply_alpha: bool, colour_space: str = "srgb", mipmaps: bool = False) -> str:
"""Compose a cache key including the premultiplied-alpha flag, colour space, and mip request.
Same source loaded both ways gets two distinct slots: the on-GPU
pixel data (premul), the sampled view format (colour space), or the
mip chain (mipmaps) differs, so they must not collide. ``filter`` is
already embedded in the ``base`` segment by the individual loaders.
"""
return (base + (":premul" if premultiply_alpha else "")
+ (":lin" if colour_space != "srgb" else "")
+ (":mips" if mipmaps else ""))
# ------------------------------------------------------------------
# Queries
# ------------------------------------------------------------------
[docs]
def get_texture_size(self, tex_idx: int) -> tuple[int, int]:
"""Return (width, height) for a loaded texture index. (0, 0) if unknown."""
return self._sizes.get(tex_idx, (0, 0))
[docs]
def get_pixels(self, tex_id: int) -> np.ndarray | None:
"""Return retained RGBA pixels for ``tex_id``, or None.
Only populated when the manager was constructed with ``retain_pixels=True``.
Used by the web runtime to re-ship 2D overlay pixels over the drain
channel without the browser having to fetch them back out.
"""
return self._pixels_by_id.get(tex_id)
[docs]
@property
def count(self) -> int:
"""Number of unique textures loaded."""
return len(self._cache)
[docs]
def destroy(self) -> None:
"""Clear all caches (GPU resources are owned by the backend)."""
self._cache.clear()
self._sizes.clear()
self._pixels_by_id.clear()
# ------------------------------------------------------------------
# Internals
# ------------------------------------------------------------------
def _load_dds(self, data: bytes, source_name: str, *, filter: str) -> int:
"""Parse DDS bytes and upload via the compressed-texture path.
Returns the bindless index, or -1 (the "couldn't resolve" sentinel) on a
parse failure, after a deduped one-time WARNING. Never raises.
"""
from ..assets.dds_loader import load_dds
try:
tex = load_dds(data)
except ValueError as exc:
self._warn_compressed(source_name, -1, str(exc))
return -1
return self._load_compressed(tex, source_name, filter=filter)
def _load_ktx2(self, data: bytes, source_name: str, *, filter: str) -> int:
"""Parse KTX2 bytes and upload via the compressed-texture path.
Same contract as :meth:`_load_dds`: a -1 sentinel + one-time WARNING on a
parse failure (bad magic, cubemap/array/3D, vkFormat 0, BasisLZ/ZLIB, or
a missing ``zstandard`` package on a Zstd file). Never raises.
"""
# Web-aware UASTC interception: a registrar that ships UASTC to the
# browser (``upload_compressed_source``) gets the RAW untranscoded levels
# so the device picks its own block format. Falls through to the normal
# parse for explicit-BC / ETC1S KTX2s (the ValueError path) and for every
# non-web backend, which has no such method. The desktop loader is untouched.
ship_uastc = getattr(self._registrar, "upload_compressed_source", None)
if callable(ship_uastc):
from ..assets.ktx2_loader import load_ktx2_uastc_source
try:
src = load_ktx2_uastc_source(data)
except ValueError:
src = None
if src is not None:
idx = ship_uastc(src.mips, src.width, src.height, srgb=src.srgb, filter=filter)
self._sizes[idx] = (src.width, src.height)
log.debug("TextureManager: %s [web-uastc %s] %dx%d srgb=%s mips=%d -> index %d",
source_name, filter, src.width, src.height, src.srgb, len(src.mips), idx)
return idx
from ..assets.ktx2_loader import load_ktx2
target = self._choose_uastc_target()
try:
tex = load_ktx2(data, target=target) if target else load_ktx2(data)
except ValueError as exc:
self._warn_compressed(source_name, -1, str(exc))
# A vkFormat-0 (UASTC) file on a GPU with no block family at all:
# fall through to the CPU-decode path so it still resolves.
return self._load_uastc_cpu_fallback(data, source_name, filter=filter)
return self._load_compressed(tex, source_name, filter=filter)
def _choose_uastc_target(self) -> str | None:
"""Pick the UASTC transcode target for this device, or None.
Order: BC7 -> ASTC-4x4 -> ETC2, each gated by BOTH the coarse family
feature (``compressed_caps``) AND the per-format SAMPLED check
(``format_supported``). Returns None when no block family is usable, so
the caller degrades to a CPU decode of mip 0.
"""
caps = getattr(self._registrar, "compressed_caps", None)
fmt_ok = getattr(self._registrar, "format_supported", None)
if not callable(caps) or not callable(fmt_ok):
# A registrar without the caps surface (e.g. an older/test stub):
# preserve the historical BC7-only behaviour.
return "bc7"
c = caps()
import vulkan as vk # noqa: PLC0415 (call-time only, never at import)
candidates = [
("bc7", "texture_compression_bc", vk.VK_FORMAT_BC7_UNORM_BLOCK),
("astc4x4", "texture_compression_astc_ldr", vk.VK_FORMAT_ASTC_4x4_UNORM_BLOCK),
("etc2", "texture_compression_etc2", vk.VK_FORMAT_ETC2_R8G8B8A8_UNORM_BLOCK),
]
for target, feature, fmt in candidates:
if c.get(feature) and fmt_ok(int(fmt)):
return target
return None
def _load_uastc_cpu_fallback(self, data: bytes, source_name: str, *, filter: str) -> int:
"""Decode UASTC mip 0 to RGBA8 on the CPU when no GPU block family fits.
Transcodes the top mip to BC7 bytes via the native transcoder, then runs
the texture2ddecoder BC7 decoder to RGBA8 and uploads through the
standard pixel path. Returns -1 (the existing -1 was already warned) if
the transcoder or decoder is unavailable.
"""
from ..assets.ktx2_loader import load_ktx2_uastc_source
try:
src = load_ktx2_uastc_source(data)
except ValueError:
return -1
from .._native import basis_transcoder as bt
if not bt.is_available():
bt.ensure_built()
if not bt.is_available():
return -1
import vulkan as vk # noqa: PLC0415 (call-time only, never at import)
from ..assets.block_decode import decode_blocks_to_rgba8
bc7 = bt.transcode_uastc_to_bc7(src.mips[0], src.width, src.height)
fmt = int(vk.VK_FORMAT_BC7_SRGB_BLOCK if src.srgb else vk.VK_FORMAT_BC7_UNORM_BLOCK)
rgba = decode_blocks_to_rgba8(fmt, src.width, src.height, bc7)
if rgba is None:
return -1
idx = self._register(rgba, src.width, src.height, filter=filter)
log.debug("TextureManager: %s [uastc->cpu-decode %s] %dx%d -> index %d",
source_name, filter, src.width, src.height, idx)
return idx
def _load_compressed(self, tex: Any, source_name: str, *, filter: str) -> int:
"""Upload a parsed compressed texture (DDSTexture or KTX2Texture).
Three outcomes, in order: (1) no compressed registrar -> warn + -1
(web has no BC path yet); (2) the GPU natively supports the BC format ->
upload the block mips through the strict compressed path; (3) the GPU
lacks the format -> CPU-decode mip 0 to RGBA8 and upload it through the
standard RGBA8 path (the universal fallback). Only when both the GPU
lacks the format AND no decoder/dep is available do we warn + -1.
``tex`` is duck-typed across DDSTexture / KTX2Texture: both expose
``vk_format``, ``width``, ``height``, ``block_size``, ``mips``.
"""
upload = getattr(self._registrar, "upload_texture_blocks", None)
supports = getattr(self._registrar, "supports_compressed_format", None)
if not callable(upload):
self._warn_compressed(source_name, -1,
"this backend has no compressed-texture path (web BC is a later phase)")
return -1
if callable(supports) and supports(tex.vk_format):
idx = self._register_blocks(tex.mips, tex.width, tex.height, tex.vk_format, tex.block_size, filter=filter)
log.debug("TextureManager: %s [compressed %s] %dx%d fmt=%d -> index %d",
source_name, filter, tex.width, tex.height, tex.vk_format, idx)
return idx
# Unsupported on this GPU: try a CPU decode of mip 0 to RGBA8.
from ..assets.block_decode import decode_blocks_to_rgba8
rgba = decode_blocks_to_rgba8(tex.vk_format, tex.width, tex.height, tex.mips[0])
if rgba is not None:
idx = self._register(rgba, tex.width, tex.height, filter=filter)
log.debug("TextureManager: %s [compressed->cpu-decode %s] %dx%d fmt=%d -> index %d",
source_name, filter, tex.width, tex.height, tex.vk_format, idx)
return idx
self._warn_compressed(source_name, tex.vk_format,
"GPU lacks this BC format and no CPU decoder/dep available")
return -1
@staticmethod
def _warn_compressed(source_name: str, vk_format: int, reason: str) -> None:
"""One-time WARNING per (source, format), mirroring the MDI fallback log."""
marker = (source_name, vk_format)
if marker in _warned_compressed:
return
_warned_compressed.add(marker)
log.warning("Compressed texture %s (VkFormat=%d) not loaded: %s", source_name, vk_format, reason)
def _register_blocks(
self, mips: list[bytes], width: int, height: int, vk_format: int, block_size: int, *,
filter: str = "linear",
) -> int:
"""Upload block-compressed mips through the compressed registrar.
Validates the blocks path's OWN contract (the strict RGBA8 (H, W, 4)
uint8 assertions are never touched): a non-empty ``list[bytes]`` and an
8- or 16-byte block size.
"""
if not mips or not all(isinstance(m, bytes | bytearray) for m in mips):
raise ValueError("Compressed upload requires a non-empty list[bytes] of mips")
if block_size not in (8, 16):
raise ValueError(f"block_size must be 8 or 16; got {block_size}")
upload = self._registrar.upload_texture_blocks
try:
idx = upload(mips, width, height, vk_format, block_size, filter=filter)
except TypeError:
idx = upload(mips, width, height, vk_format, block_size)
self._sizes[idx] = (width, height)
return idx
def _try_encoded_upload(self, data: bytes, *, filter: str, colour_space: str, mipmaps: bool = False) -> int | None:
"""Hand raw encoded image bytes to a native-decode registrar, if it has one.
Returns the minted texture id, or ``None`` when the registrar lacks the
capability (desktop) so the caller falls back to CPU decode. The decode
resolves one frame late on the backend, exactly like the mesh-texture and
compressed-texture drain paths. ``colour_space == "srgb"`` selects the
hardware sRGB sampled view, matching ``_register``. ``mipmaps=True``
requests runtime mip generation after the decode; probed with
the same try/except idiom as ``_register`` so registrars without the
capability fall back to a single mip instead of erroring.
"""
if not getattr(self._registrar, "supports_encoded_image_upload", False):
return None
srgb = colour_space == "srgb"
idx: int | None = None
if mipmaps:
try:
idx = self._registrar.upload_encoded_image(data, srgb=srgb, filter=filter, mipmaps=True)
except TypeError:
idx = None # registrar has no runtime mipgen: plain upload below
if idx is None:
idx = self._registrar.upload_encoded_image(data, srgb=srgb, filter=filter)
# Size is unknown until the browser decodes; recorded as (0, 0) like any
# async-resolved texture. Callers that need pixel dims should sample the
# decoded texture, not the manager (web never CPU-holds these pixels).
self._sizes.setdefault(idx, (0, 0))
return idx
def _register(
self, pixels: np.ndarray, width: int, height: int, *,
filter: str = "linear", colour_space: str = "srgb", mipmaps: bool = False,
) -> int:
# Backends that don't yet accept ``filter``/``colour_space``/``mipmaps``
# (older registrars, web stub) silently fall through to the default
# sampler, colour space, and single mip: Sprite2D's contract is "linear
# works everywhere, nearest works where supported". Probe via try/except
# instead of inspect to keep the hot path tight. ``colour_space`` selects
# the sampled view format (sRGB for perceptual colour, linear/UNORM for
# data + the 3D albedo path); ``mipmaps`` requests runtime mip generation
# (desktop blit chain; web fullscreen-sample chain).
upload = self._registrar.upload_texture_pixels
idx: int | None = None
if mipmaps:
try:
idx = upload(pixels, width, height, filter=filter, colour_space=colour_space, mipmaps=True)
except TypeError:
idx = None # registrar has no runtime mipgen: plain upload below
if idx is None:
try:
idx = upload(pixels, width, height, filter=filter, colour_space=colour_space)
except TypeError:
try:
idx = upload(pixels, width, height, filter=filter)
except TypeError:
idx = upload(pixels, width, height)
self._sizes[idx] = (width, height)
if self._retain_pixels:
self._pixels_by_id[idx] = pixels
return idx
[docs]
def release(self, idx: int, cache_key: str | None = None,
source_id: int | None = None) -> None:
"""Reclaim a texture slot + drop cache bookkeeping.
Called by the weakref.finalize attached in ``load_from_array`` when
the source ndarray is GC'd. Backend unregister is delegated to the
registrar when it exposes ``unregister_texture`` (desktop Engine);
web registrars may opt out.
"""
self._sizes.pop(idx, None)
self._pixels_by_id.pop(idx, None)
if cache_key is not None:
self._cache.pop(cache_key, None)
unreg = getattr(self._registrar, "unregister_texture", None)
if callable(unreg):
try:
unreg(idx)
except Exception:
log.exception("unregister_texture(%d) failed", idx)
self._finalizers.pop(idx, None)