"""Texture -- an image the engine owns, with an identity, a version and an update path."""
from __future__ import annotations
import copy
import io
import os
import weakref
from collections.abc import Callable
from typing import Any
import numpy as np
from ._slots import slot_names
__all__ = ["Texture", "is_live_target"]
_UNSET = object()
#: Nodes that publish a live offscreen image: the structural marker each one
#: carries, the type that marker belongs to, and the assignments that sample it.
#: Duck-typed, the way every consumer that ROUTES a live feed reads them -- a
#: material's albedo, a sprite's texture, the backend's lazy-load skip -- and
#: they all read them through :func:`is_live_target`, so both kinds are accepted
#: wherever one is. Size is the exception: ``_offscreen_target_size`` measures
#: one off ``texture_size`` instead, which is why a wrapped target still
#: answered a real ``size``.
_LIVE_TARGETS = (
("_is_subviewport", "SubViewport", "sprite.texture = subviewport, or Material(albedo_map=subviewport)"),
("_is_renderview", "RenderView", "sprite.texture = render_view, or Material(albedo_map=render_view)"),
)
[docs]
def is_live_target(source: Any) -> bool:
"""Whether *source* is a node that publishes a live offscreen image.
The one question every consumer that ROUTES a live feed asks: a sprite
resolving its bindless slot each draw, a material binding an albedo, a
backend skipping the lazy path/bytes load. Both kinds answer it -- a
:class:`~simvx.core.SubViewport`, which renders its own subtree, and a
:class:`~simvx.core.RenderView`, which renders the main scene through a
second camera -- so no consumer has to name one kind and mean both.
"""
for marker, _kind, _spelling in _LIVE_TARGETS:
if getattr(source, marker, False):
return True
return False
def _refuse_live_target(source: Any) -> None:
"""Raise when *source* is a node that publishes a live offscreen image.
A live target is a texture source already: the backend owns its image and
re-renders it, so there are no pixels here for a resource to own or to
update. Wrapped, it would also slip past every duck-typed check downstream
(each reads the marker off the value the property holds, which would now be
a :class:`Texture`) and reach the file loader, which asks ``Path()`` for a
node. Refused where it is written instead, naming the assignment that works.
"""
for marker, kind, spelling in _LIVE_TARGETS:
if getattr(source, marker, False):
named = kind if type(source).__name__ == kind else f"{type(source).__name__} (a {kind})"
raise TypeError(
f"A {named} is a live texture already and cannot be wrapped in a Texture: the backend owns "
f"that image and re-renders it, so there are no pixels here to own or to update(). Assign "
f"the node itself, which is a supported spelling: {spelling}."
)
def _decoder_size(source: Any) -> tuple[int, int]:
"""Dimensions an imaging library reads from *source*'s header, or ``(0, 0)``.
The last resort for the containers this engine's own header reader does not
parse -- TGA, TIFF and the rest of the decoder's tail. Costs a header parse,
not a decode: opening an image exposes its dimensions without touching pixel
data. Answers ``(0, 0)`` when no imaging library is installed, which is the
web runtime's position and the reason a backend can still hand a size back
through :meth:`Texture._adopt_size`.
"""
if not isinstance(source, str | bytes | bytearray | memoryview | os.PathLike):
return (0, 0)
try:
from PIL import Image
except ImportError:
return (0, 0)
try:
handle = io.BytesIO(bytes(source)) if isinstance(source, bytes | bytearray | memoryview) else str(source)
with Image.open(handle) as img:
return (int(img.width), int(img.height))
except (OSError, ValueError):
return (0, 0)
def _measured_size(source: Any) -> tuple[int, int]:
"""Native ``(width, height)`` of a texture source, or ``(0, 0)`` when nothing can say.
The engine's own header reader answers first: it costs a short read and needs
no third-party library. Only the containers it does not parse fall through to
the decoder's header parse.
"""
from ..animation.sprite import _native_size_of
size = _native_size_of(source)
return size if size is not None else _decoder_size(source)
[docs]
class Texture:
"""An image resource: an identity, a version, and an explicit way to change it.
A raw texture source (a path, encoded bytes, an RGBA ndarray) answers
neither "is this the same image as before" nor "has it changed". A
``Texture`` answers both: its identity is stable across reassignment, and
:meth:`update` bumps :attr:`version`, which is what drives a re-upload.
Wrapping is optional. Every node property that takes a texture still
accepts a plain path, bytes or ndarray, and always will -- that is the
short spelling and it is not going away. Reach for ``Texture`` when the
image CHANGES after it is first shown::
heightmap = Texture(pixels) # RGBA uint8 (H, W, 4)
sprite.texture = heightmap
...
pixels[:] = erode(pixels)
heightmap.update() # re-uploads in place, same slot
icon = Texture("ui/icon.png", filter="nearest")
sprite.texture = icon
:meth:`update` may also swap the pixels outright
(``heightmap.update(new_pixels)``), which is the cheaper route when the
replacement is a fresh array rather than an in-place edit. What it cannot do
is change the SIZE: that is fixed when the texture is constructed, and a
differently sized image is a different texture. Build one and assign it over
the old, which re-resolves every consumer.
The sampling settings are fixed at construction: they select which backend
slot the image lands in, so changing one is a different texture, not an
update to this one. Each one you leave out stays unset, and an unset setting
defers to whatever the consumer asks for -- the sprite's ``filter``, the
material map role's colour space and mip chain. Only a setting you state
here overrides that, and it overrides it everywhere the texture is used::
sprite.texture = Texture(pixels) # follows sprite.filter
sprite.texture = Texture(pixels, filter="nearest") # crisp, wherever it is drawn
One resource owns one backend slot, so it carries one sampling. Two uses that
need different ones cannot both be served: the first wins and the second is
reported. Give each its own ``Texture`` over the same source.
A scene file can carry a texture whose source is a FILE: it is written as
the constructor call that rebuilds it, with exactly the sampling settings
that were stated, and one held by several nodes is written once into a
local so the reloaded scene shares it the same way. Pixels in memory have
no source form -- a scene is code, not an image container -- so a texture
over an array or over encoded bytes stops the save and is named in the
report. Build that one in ``on_ready()`` and assign it there, which is
where code belongs anyway.
A texture owns its image on the backend outright. Two resources over the
same file do not share it, nor does one share with the same file assigned
as a plain path, so :meth:`update` can never change an image something else
asked for; the price is one extra copy on the GPU when a file is used both
ways. Ownership runs the other way too: the backend slot is handed back when
the resource is collected, so building a texture per level or per entity
does not exhaust the texture table.
"""
_next_uid: int = 0
__slots__ = (
"_uid",
"_source",
"_version",
"_size",
"_filter",
"_colour_space",
"_premultiply_alpha",
"_mipmaps",
"_managers",
"__weakref__",
)
def __init__(
self,
source: Any,
*,
filter: str | None = None,
colour_space: str | None = None,
premultiply_alpha: bool | None = None,
mipmaps: bool | None = None,
) -> None:
"""Wrap *source* as an owned texture.
Every sampling setting is optional, and omitting one is not the same as
choosing its usual default: an omitted setting follows the consumer's
own request, so the same texture is sampled the way a plain path or
array would have been in that position. State one only to override that
everywhere the texture is used.
Args:
source: File path, encoded image bytes, or an RGBA ``uint8``
ndarray of shape ``(H, W, 4)`` -- the same set every texture
property accepts. A live offscreen target (a ``SubViewport``, a
``RenderView``) is refused: it is a texture source already, so
assign the node itself where a live feed is sampled -- either
kind to ``sprite.texture`` or to an ``albedo_map``.
filter: Sampler filter, ``"linear"`` or ``"nearest"``. Unset follows
the consumer (a sprite's ``filter`` property).
colour_space: ``"srgb"`` (perceptual colour) or ``"linear"`` (data
maps: normals, metallic-roughness, AO). Unset follows the
consumer, which is what keeps a texture used as a ``normal_map``
linear without being told twice.
premultiply_alpha: Multiply RGB by alpha before upload. Unset
follows the consumer.
mipmaps: Generate a runtime mip chain. Unset follows the consumer,
which is what keeps an ``albedo_map``'s chain.
"""
if source is None:
raise ValueError("Texture needs a source: a file path, encoded image bytes, or an RGBA uint8 ndarray")
_refuse_live_target(source)
if filter is not None and filter not in ("linear", "nearest"):
raise ValueError(f"filter must be 'linear' or 'nearest'; got {filter!r}")
if colour_space is not None and colour_space not in ("srgb", "linear"):
raise ValueError(f"colour_space must be 'srgb' or 'linear'; got {colour_space!r}")
Texture._next_uid += 1
self._uid: int = Texture._next_uid
self._source: Any = source
self._version: int = 0
self._size: tuple[int, int] | None = None
self._filter = filter
self._colour_space = colour_space
self._premultiply_alpha = None if premultiply_alpha is None else bool(premultiply_alpha)
self._mipmaps = None if mipmaps is None else bool(mipmaps)
# Backend texture managers that hold a slot for this resource. Weak, so
# a manager (and its renderer) can still be collected; each is told to
# re-upload when ``update`` bumps the version.
self._managers: weakref.WeakSet = weakref.WeakSet()
# --- Identity and content ------------------------------------------------
[docs]
@property
def source(self) -> Any:
"""The underlying path / bytes / ndarray."""
return self._source
[docs]
@property
def version(self) -> int:
"""Increments on every :meth:`update`. Backends re-upload when it moves."""
return self._version
[docs]
@property
def filter(self) -> str | None:
"""Sampler filter, or ``None`` when it follows the consumer's request."""
return self._filter
[docs]
@property
def colour_space(self) -> str | None:
"""Colour space, or ``None`` when it follows the consumer's request."""
return self._colour_space
[docs]
@property
def premultiply_alpha(self) -> bool | None:
"""Premultiplication, or ``None`` when it follows the consumer's request."""
return self._premultiply_alpha
[docs]
@property
def mipmaps(self) -> bool | None:
"""Mip-chain request, or ``None`` when it follows the consumer's request."""
return self._mipmaps
[docs]
@property
def size(self) -> tuple[int, int]:
"""Native pixel ``(width, height)``, or ``(0, 0)`` when it cannot be read.
Once known, the size never changes: :meth:`update` refuses pixels of a
different size, so anything that measures a texture once -- a sprite
drawing at native size, a nine-patch normalising its border UVs --
stays correct without watching for changes.
Resolved without the GPU (an ndarray's shape, or the image's header), so
it answers identically on every backend and before the first upload. Only
a source no header reader here can measure answers ``(0, 0)``, and only
until a backend decodes it and reports the dimensions back -- which on a
runtime with no imaging library means the first upload.
"""
if self._size is None:
self._size = _measured_size(self._source)
return self._size
# --- Mutation ------------------------------------------------------------
[docs]
def update(self, pixels: Any = None) -> None:
"""Replace this texture's pixels, keeping its size, its slot and its identity.
This is the change-the-image path and only that. Call it after mutating
the pixel array in place, or pass *pixels* to swap the array outright::
pixels[:] = erode(pixels)
heightmap.update() # or: heightmap.update(new_pixels)
Either way :attr:`version` moves and every backend holding a slot for
this resource writes the new pixels into that SAME slot, so anything
already drawing the texture keeps its handle and shows the new image on
the next frame. Only consumers of this resource are affected: the slot
is the resource's own, never one shared with a plain path or array.
Resizing is a different operation, and this is not it. Consumers measure
a texture when they resolve it -- a sprite sizes its quad to the native
pixels, a :class:`~simvx.core.NinePatchRect` normalises every border UV
by them -- so a slot that changed shape underneath them would draw the
new image at the old scale, or sample the wrong region, with nothing to
say so. A differently sized image is a different texture::
sprite.texture = Texture(bigger_pixels)
Reassignment re-resolves every consumer, which is what makes it the
right answer rather than a workaround.
Args:
pixels: Replacement source of the same size and pixel format, or
``None`` to re-upload the current one after an in-place edit.
Raises:
ValueError: If the pixels differ in size or format from this
texture's. Nothing is mutated and no backend is told, so a
refused update leaves every slot exactly as it was.
TypeError: If *pixels* is a live offscreen target, refused here for
the same reason the constructor refuses one -- and equally
without mutating anything. A live target measures as a real
size, so a size check cannot be trusted to catch one: when the
sizes match it passes straight through, and when they do not it
is reported as a resize.
"""
if pixels is not None:
_refuse_live_target(pixels)
self._refuse_reshape(self._source if pixels is None else pixels)
if pixels is not None:
self._source = pixels
self._version += 1
for manager in list(self._managers):
manager.refresh(self)
def _refuse_reshape(self, source: Any) -> None:
"""Raise unless *source* carries this texture's size and pixel format.
Runs before anything is mutated, so a refused update cannot half-apply.
A texture that has never had a measurable size has no invariant to
protect and nothing to compare against, so it passes. One that HAS a size
is held to it, and a replacement nothing here can measure fails that
check rather than passing on the benefit of the doubt: the whole point is
that a wrong size is silent, so an unverifiable one cannot be waved
through. It would not have survived the upload anyway -- a runtime with
no reader for the format has no decoder for it either.
"""
if isinstance(source, np.ndarray) and (source.ndim != 3 or source.shape[2] != 4 or source.dtype != np.uint8):
raise ValueError(
f"Texture.update() needs RGBA uint8 pixels of shape (H, W, 4); got {source.shape} of "
f"{source.dtype}. The slot was uploaded in one format and writes into it in that format."
)
was = self.size
if not (was[0] and was[1]):
return
now = _measured_size(source)
if now == was:
return
measured = f"the new pixels are {now[0]}x{now[1]}" if now[0] and now[1] else "the new source cannot be measured"
raise ValueError(
f"Texture.update() cannot change size: this texture is {was[0]}x{was[1]}, {measured}. Everything "
f"drawing it measured that size once. Create a new Texture and assign it over the old one "
f"(sprite.texture = Texture(pixels)); reassignment re-resolves everything."
)
# --- Backend plumbing ----------------------------------------------------
def _attach_manager(self, manager: Any) -> None:
"""Record a texture manager holding a slot for this resource."""
self._managers.add(manager)
def _adopt_size(self, width: int, height: int) -> None:
"""Record dimensions a backend measured, for a source this engine cannot read.
The header reader covers the common containers; the decoder accepts a
longer tail (TGA, TIFF and the rest of the PIL set) whose dimensions
exist nowhere until something decodes them. A manager that has just
decoded one hands them back here, which is what lets :meth:`update`
refuse a resize of those sources too, rather than having no evidence to
refuse it with.
Only ever fills a gap: a size the header answered is never overwritten,
so the two never disagree about a format both can measure.
"""
if self.size != (0, 0):
return
measured = (int(width), int(height))
if measured[0] > 0 and measured[1] > 0:
self._size = measured
# --- Identity across copies ----------------------------------------------
def _clone_into(self, clone: Texture, copy_value: Callable[[Any], Any]) -> Texture:
"""Fill *clone* from this texture, giving it an identity of its own.
``_uid`` is this texture's identity, and a backend keys its bindless
slot and that slot's reclamation on it. A copy that carried the
original's uid would be a second live resource claiming one slot:
whichever died first would hand back a slot the other still holds, and
an :meth:`update` on either would re-upload for both. Every copy gets a
fresh identity, and with it a fresh empty ``_managers``: no backend
holds a slot for the clone until something resolves it.
A subclass that declares no ``__slots__`` of its own keeps its state in
a ``__dict__`` instead, so both stores are carried across; pickling
already restores both, and the two paths must agree.
"""
for name in slot_names(type(self)):
if name in ("_uid", "_managers"):
continue
value = getattr(self, name, _UNSET)
if value is _UNSET: # a subclass slot that was never assigned
continue
setattr(clone, name, copy_value(value))
state = getattr(self, "__dict__", None)
if state:
clone.__dict__.update({name: copy_value(value) for name, value in state.items()})
Texture._next_uid += 1
clone._uid = Texture._next_uid
clone._managers = weakref.WeakSet()
return clone
[docs]
def __copy__(self) -> Texture:
return self._clone_into(object.__new__(type(self)), lambda value: value)
[docs]
def __deepcopy__(self, memo: dict) -> Texture:
clone = object.__new__(type(self))
memo[id(self)] = clone # one copy per texture, however many times a structure holds it
return self._clone_into(clone, lambda value: copy.deepcopy(value, memo))
[docs]
def __getstate__(self) -> tuple[dict | None, dict]:
"""Pickle everything but the process-local state.
``_managers`` holds weak references to live backend managers, which no
pickle can carry; ``_uid`` is only meaningful in the process that
minted it. Both are rebuilt by :meth:`__setstate__`.
"""
attrs = getattr(self, "__dict__", None)
slots = {
name: value
for name in slot_names(type(self))
if name not in ("_uid", "_managers") and (value := getattr(self, name, _UNSET)) is not _UNSET
}
return (dict(attrs) if attrs else None, slots)
[docs]
def __setstate__(self, state: Any) -> None:
"""Restore a pickled texture, minting a fresh identity.
The restored texture takes a new ``_uid`` rather than a number that may
already belong to a live texture here, and an empty ``_managers``: no
backend holds a slot for it until something resolves it.
"""
attrs, slots = state if isinstance(state, tuple) else (state, None)
if attrs:
self.__dict__.update(attrs) # a subclass that also carries a __dict__
for name, value in (slots or {}).items():
if name != "_uid":
setattr(self, name, value)
Texture._next_uid += 1
self._uid = Texture._next_uid
self._managers = weakref.WeakSet()
[docs]
def __repr__(self) -> str:
src = self._source
name = src if isinstance(src, str) else type(src).__name__
stated = [
f"{key}={value!r}"
for key, value in (
("filter", self._filter),
("colour_space", self._colour_space),
("premultiply_alpha", self._premultiply_alpha),
("mipmaps", self._mipmaps),
)
if value is not None
]
return f"Texture({', '.join([repr(name), *stated])}, version={self._version})"