Source code for bascenev1._dependency

# Released under the MIT License. See LICENSE for details.
#
"""Functionality related to object/asset dependencies."""

from __future__ import annotations

import weakref
from typing import Generic, TypeVar, TYPE_CHECKING, override

import babase

import _bascenev1

if TYPE_CHECKING:
    from typing import Any

    import bascenev1

T = TypeVar('T', bound='DependencyComponent')


class Dependency(Generic[T]):
    """A dependency on a DependencyComponent (with an optional config).

    Category: **Dependency Classes**

    This class is used to request and access functionality provided
    by other DependencyComponent classes from a DependencyComponent class.
    The class functions as a descriptor, allowing dependencies to
    be added at a class level much the same as properties or methods
    and then used with class instances to access those dependencies.
    For instance, if you do 'floofcls = bascenev1.Dependency(FloofClass)'
    you would then be able to instantiate a FloofClass in your class's
    methods via self.floofcls().
    """

    def __init__(self, cls: type[T], config: Any = None):
        """Instantiate a Dependency given a bascenev1.DependencyComponent type.

        Optionally, an arbitrary object can be passed as 'config' to
        influence dependency calculation for the target class.
        """
        self.cls: type[T] = cls
        self.config = config
        self._hash: int | None = None

[docs] def get_hash(self) -> int: """Return the dependency's hash, calculating it if necessary.""" from efro.util import make_hash if self._hash is None: self._hash = make_hash((self.cls, self.config)) return self._hash
def __get__(self, obj: Any, cls: Any = None) -> T: if not isinstance(obj, DependencyComponent): if obj is None: raise TypeError( 'Dependency must be accessed through an instance.' ) raise TypeError( f'Dependency cannot be added to class of type {type(obj)}' ' (class must inherit from bascenev1.DependencyComponent).' ) # We expect to be instantiated from an already living # DependencyComponent with valid dep-data in place.. assert cls is not None # Get the DependencyEntry this instance is associated with and from # there get back to the DependencySet entry = getattr(obj, '_dep_entry') if entry is None: raise RuntimeError('Invalid dependency access.') entry = entry() assert isinstance(entry, DependencyEntry) depset = entry.depset() assert isinstance(depset, DependencySet) if not depset.resolved: raise RuntimeError( "Can't access data on an unresolved DependencySet." ) # Look up the data in the set based on the hash for this Dependency. assert self._hash in depset.entries entry = depset.entries[self._hash] assert isinstance(entry, DependencyEntry) retval = entry.get_component() assert isinstance(retval, self.cls) return retval class DependencyComponent: """Base class for all classes that can act as or use dependencies. Category: **Dependency Classes** """ _dep_entry: weakref.ref[DependencyEntry] def __init__(self) -> None: """Instantiate a DependencyComponent.""" # For now lets issue a warning if these are instantiated without # a dep-entry; we'll make this an error once we're no longer # seeing warnings. # entry = getattr(self, '_dep_entry', None) # if entry is None: # print(f'FIXME: INSTANTIATING DEP CLASS {type(self)} DIRECTLY.')
[docs] @classmethod def dep_is_present(cls, config: Any = None) -> bool: """Return whether this component/config is present on this device.""" del config # Unused here. return True
[docs] @classmethod def get_dynamic_deps(cls, config: Any = None) -> list[Dependency]: """Return any dynamically-calculated deps for this component/config. Deps declared statically as part of the class do not need to be included here; this is only for additional deps that may vary based on the dep config value. (for instance a map required by a game type) """ del config # Unused here. return []
class DependencyEntry: """Data associated with a dep/config pair in bascenev1.DependencySet.""" # def __del__(self) -> None: # print('~DepEntry()', self.cls) def __init__(self, depset: DependencySet, dep: Dependency[T]): # print("DepEntry()", dep.cls) self.cls = dep.cls self.config = dep.config # Arbitrary data for use by dependencies in the resolved set # (the static instance for static-deps, etc). self.component: DependencyComponent | None = None # Weakref to the depset that includes us (to avoid ref loop). self.depset = weakref.ref(depset) def get_component(self) -> DependencyComponent: """Return the component instance, creating it if necessary.""" if self.component is None: # We don't simply call our type to instantiate our instance; # instead we manually call __new__ and then __init__. # This allows us to inject its data properly before __init__(). print('creating', self.cls) instance = self.cls.__new__(self.cls) # pylint: disable=protected-access, unnecessary-dunder-call instance._dep_entry = weakref.ref(self) instance.__init__() # type: ignore depset = self.depset() assert depset is not None self.component = instance component = self.component assert isinstance(component, self.cls) if component is None: raise RuntimeError( f'Accessing DependencyComponent {self.cls} ' 'in an invalid state.' ) return component class DependencySet(Generic[T]): """Set of resolved dependencies and their associated data. Category: **Dependency Classes** To use DependencyComponents, a set must be created, resolved, and then loaded. The DependencyComponents are only valid while the set remains in existence. """ def __init__(self, root_dependency: Dependency[T]): # print('DepSet()') self._root_dependency = root_dependency self._resolved = False self._loaded = False # Dependency data indexed by hash. self.entries: dict[int, DependencyEntry] = {} # def __del__(self) -> None: # print("~DepSet()")
[docs] def resolve(self) -> None: """Resolve the complete set of required dependencies for this set. Raises a bascenev1.DependencyError if dependencies are missing (or other Exception types on other errors). """ if self._resolved: raise RuntimeError('DependencySet has already been resolved.') # print('RESOLVING DEP SET') # First, recursively expand out all dependencies. self._resolve(self._root_dependency, 0) # Now, if any dependencies are not present, raise an Exception # telling exactly which ones (so hopefully they'll be able to be # downloaded/etc. missing = [ Dependency(entry.cls, entry.config) for entry in self.entries.values() if not entry.cls.dep_is_present(entry.config) ] if missing: raise DependencyError(missing) self._resolved = True
# print('RESOLVE SUCCESS!') @property def resolved(self) -> bool: """Whether this set has been successfully resolved.""" return self._resolved
[docs] def get_asset_package_ids(self) -> set[str]: """Return the set of asset-package-ids required by this dep-set. Must be called on a resolved dep-set. """ ids: set[str] = set() if not self._resolved: raise RuntimeError('Must be called on a resolved dep-set.') for entry in self.entries.values(): if issubclass(entry.cls, AssetPackage): assert isinstance(entry.config, str) ids.add(entry.config) return ids
[docs] def load(self) -> None: """Instantiate all DependencyComponents in the set. Returns a wrapper which can be used to instantiate the root dep. """ # NOTE: stuff below here should probably go in a separate 'instantiate' # method or something. if not self._resolved: raise RuntimeError("Can't load an unresolved DependencySet") for entry in self.entries.values(): # Do a get on everything which will init all payloads # in the proper order recursively. entry.get_component() self._loaded = True
@property def root(self) -> T: """The instantiated root DependencyComponent instance for the set.""" if not self._loaded: raise RuntimeError('DependencySet is not loaded.') rootdata = self.entries[self._root_dependency.get_hash()].component assert isinstance(rootdata, self._root_dependency.cls) return rootdata def _resolve(self, dep: Dependency[T], recursion: int) -> None: # Watch for wacky infinite dep loops. if recursion > 10: raise RecursionError('Max recursion reached') hashval = dep.get_hash() if hashval in self.entries: # Found an already resolved one; we're done here. return # Add our entry before we recurse so we don't repeat add it if # there's a dependency loop. self.entries[hashval] = DependencyEntry(self, dep) # Grab all Dependency instances we find in the class. subdeps = [ cls for cls in dep.cls.__dict__.values() if isinstance(cls, Dependency) ] # ..and add in any dynamic ones it provides. subdeps += dep.cls.get_dynamic_deps(dep.config) for subdep in subdeps: self._resolve(subdep, recursion + 1) class AssetPackage(DependencyComponent): """bascenev1.DependencyComponent representing a bundled package of assets. Category: **Asset Classes** """ def __init__(self) -> None: super().__init__() # This is used internally by the get_package_xxx calls. self.context = babase.ContextRef() entry = self._dep_entry() assert entry is not None assert isinstance(entry.config, str) self.package_id = entry.config print(f'LOADING ASSET PACKAGE {self.package_id}')
[docs] @override @classmethod def dep_is_present(cls, config: Any = None) -> bool: assert isinstance(config, str) # Temp: hard-coding for a single asset-package at the moment. if config == 'stdassets@1': return True return False
[docs] def gettexture(self, name: str) -> bascenev1.Texture: """Load a named bascenev1.Texture from the AssetPackage. Behavior is similar to bascenev1.gettexture() """ return _bascenev1.get_package_texture(self, name)
[docs] def getmesh(self, name: str) -> bascenev1.Mesh: """Load a named bascenev1.Mesh from the AssetPackage. Behavior is similar to bascenev1.getmesh() """ return _bascenev1.get_package_mesh(self, name)
[docs] def getcollisionmesh(self, name: str) -> bascenev1.CollisionMesh: """Load a named bascenev1.CollisionMesh from the AssetPackage. Behavior is similar to bascenev1.getcollisionmesh() """ return _bascenev1.get_package_collision_mesh(self, name)
[docs] def getsound(self, name: str) -> bascenev1.Sound: """Load a named bascenev1.Sound from the AssetPackage. Behavior is similar to bascenev1.getsound() """ return _bascenev1.get_package_sound(self, name)
[docs] def getdata(self, name: str) -> bascenev1.Data: """Load a named bascenev1.Data from the AssetPackage. Behavior is similar to bascenev1.getdata() """ return _bascenev1.get_package_data(self, name)
class TestClassFactory(DependencyComponent): """Another test dep-obj.""" _assets = Dependency(AssetPackage, 'stdassets@1') def __init__(self) -> None: super().__init__() print('Instantiating TestClassFactory') self.tex = self._assets.gettexture('black') self.mesh = self._assets.getmesh('landMine') self.sound = self._assets.getsound('error') self.data = self._assets.getdata('langdata') class TestClassObj(DependencyComponent): """Another test dep-obj.""" class TestClass(DependencyComponent): """A test dep-obj.""" _testclass = Dependency(TestClassObj) _factoryclass = Dependency(TestClassFactory, 123) _factoryclass2 = Dependency(TestClassFactory, 123) def __del__(self) -> None: print('~TestClass()') def __init__(self) -> None: super().__init__() print('TestClass()') self._actor = self._testclass print('got actor', self._actor) print('have factory', self._factoryclass) print('have factory2', self._factoryclass2) def test_depset() -> None: """Test call to try this stuff out...""" if bool(False): print('running test_depset()...') def doit() -> None: depset = DependencySet(Dependency(TestClass)) try: depset.resolve() except DependencyError as exc: for dep in exc.deps: if dep.cls is AssetPackage: print('MISSING ASSET PACKAGE', dep.config) else: raise RuntimeError( f'Unknown dependency error for {dep.cls}' ) from exc except Exception as exc: print('DependencySet resolve failed with exc type:', type(exc)) if depset.resolved: depset.load() testobj = depset.root # instance = testclass(123) print('INSTANTIATED ROOT:', testobj) doit() # To test this, add prints on __del__ for stuff used above; # everything should be dead at this point if we have no cycles. print('everything should be cleaned up...') babase.quit() class DependencyError(Exception): """Exception raised when one or more bascenev1.Dependency items are missing. Category: **Exception Classes** (this will generally be missing assets). """ def __init__(self, deps: list[bascenev1.Dependency]): super().__init__() self._deps = deps @property def deps(self) -> list[bascenev1.Dependency]: """The list of missing dependencies causing this error.""" return self._deps