Source code for simvx.core.scene_io.loader

"""Runtime loading: import a scene `.py` file or folder and instantiate its primary Node.

`load_scene` is the canonical way to take a path on disk and return a live
:class:`~simvx.core.Node` tree. For *editing* the source, use
:class:`SceneFile` / :class:`SceneModule`; this module is the runtime side.

Module names
------------
A loaded scene is a real module, and its name is a pure function of its
resolved path: the same file always lands on the same :data:`sys.modules` key
and two different files never share one. The readable part of the name is the
path relative to the project the scene belongs to (the directory holding
``simvx.toml``) or, failing that, to the scene's own directory; a digest of the
full resolved path follows it, so two projects that both contain
``levels/forest.py`` stay apart. The name carries no dots: a dotted key whose
parent packages do not exist is one that :mod:`pickle`, and anything else
reaching for a class through ``__import__``, cannot resolve.

What sits beside a file scene
-----------------------------
``python scene.py`` can ``import helpers`` from beside itself only because the
interpreter launcher puts the script's directory on :data:`sys.path`, and no
import API reproduces that. Scenes are ordinary Python files that must load from
anywhere, so the loader supplies the same reach without the same cost. A
permanent :data:`sys.path` entry would make every neighbour of every scene
importable under its bare name for the rest of the process, which is how two
scenes in two directories end up sharing one ``helpers`` module; ``levels/a.py``
and ``levels/b.py`` each keeping a ``util.py`` beside them is the most ordinary
project layout there is, and it has to load.

So each scene directory gets a synthetic package of its own, named for the
directory the way a scene module is named for its file. While a scene's own
top-level code runs, a bare ``import helpers`` resolves to ``helpers.py`` in
that directory and the module is registered as a submodule of that package:
``a/util.py`` and ``b/util.py`` are two modules with two names, and a class
defined in either carries a ``__module__`` that still resolves. The bare name is
bound alongside it for the length of the exec, which is what makes a second
``import`` of the same neighbour, and any ``import helpers.thing``, find what the
first one loaded; it is unbound afterwards, so nothing outside the scene ever
sees it. Two scenes in the *same* directory share one set of neighbours, as they
would under ``python``.

The reach lasts as long as the scene's own top-level code, which is where a
scene states what it needs; a scene that defers ``import helpers`` to the inside
of a method is asking for it after the scene has run, and should import at
module level instead. Only Python sources and packages are found this way, not
extension modules.

Folder scenes need none of this: they are packages, their files import each
other relatively, and nothing about them goes on :data:`sys.path` either.

:data:`sys.meta_path`
---------------------
The three finders below are installed on the first load, not on import, so a
process that never loads a scene has an untouched import system. Once installed
they stay: a scene module and its neighbours remain in :data:`sys.modules` for
the life of the process, there is no unload, and a submodule of either, or a
:func:`importlib.reload` of a scene, still has to resolve afterwards. All three
are inert for every name outside the two scene prefixes, so what they cost a
host process after that is one dict lookup per import miss.

Freshness
---------
Loading the same unchanged file twice returns the same class, so a repeat load
keeps object identity; a load after an edit compiles the new text. Both come
from one rule: the compiled result is kept against a digest of the file's
bytes, and a hit means the source has not changed since it was compiled. The
digest is of the content itself, not of size and modification time, because
those cannot tell a rewrite apart from what it replaced when the write lands on
the same byte count and the timestamp does not move, which is one save on any
filesystem that keeps whole-second times. A scene is no fresher than the
neighbours it imported, so
those count towards its signature too, and an edited neighbour is dropped from
:data:`sys.modules` so that the reload runs it again. For a folder scene the
signature covers every ``.py`` under the folder.
"""

from __future__ import annotations

import hashlib
import importlib
import importlib.machinery
import importlib.util
import re
import sys
from collections.abc import Iterable, Iterator
from contextlib import contextmanager
from pathlib import Path
from types import ModuleType
from typing import TYPE_CHECKING, Any

from .detection import primary_node_class_from_source

if TYPE_CHECKING:
    from ..node import Node

__all__ = ["import_file", "load_scene"]

#: Prefix every scene module name carries, so scene modules are recognisable in
#: :data:`sys.modules` and never collide with a real importable package.
_NAME_PREFIX = "_simvx_scene_"

#: Prefix for the synthetic package that holds one scene directory's neighbours.
_BESIDE_PREFIX = "_simvx_beside_"

#: The two namespaces the scene finders answer for, and nothing else.
_SCENE_PREFIXES = (_NAME_PREFIX, _BESIDE_PREFIX)

#: Runs of characters that cannot appear in a module name.
_NOT_IN_A_NAME = re.compile(r"[^0-9A-Za-z_]+")

#: How much of the path digest goes into a module name.
_DIGEST_LENGTH = 12

#: Compiled scene classes, keyed by resolved path, each held against the source
#: signature it was compiled from. One entry per path: a recompile replaces it.
_compiled: dict[str, tuple[Any, type]] = {}

#: The files each scene imported from beside itself, by resolved scene path. A
#: scene is only as fresh as the neighbours its own code ran.
_scene_neighbours: dict[str, tuple[Path, ...]] = {}

#: The synthetic package holding one scene directory's neighbours, by directory.
_beside_packages: dict[str, ModuleType] = {}

#: Each loaded neighbour, by its namespaced module name, against the file and
#: source signature it was compiled from.
_beside_signatures: dict[str, tuple[Path, Any]] = {}

#: The file each scene module was loaded from, and the folder it is a package
#: of, by module name. A name derived from a path is on no search path, so this
#: is the only way anything can resolve it again.
_scene_files: dict[str, tuple[Path, Path | None]] = {}


[docs] def load_scene(path: str | Path) -> Node: """Load a scene from a ``.py`` file or scene-module folder. File path → runs the file as a module of its own and instantiates the primary :class:`Node` subclass. Modules beside the file are importable by bare name while it runs, as they would be under ``python scene.py``. Folder path → runs the folder as a package and instantiates the primary class declared in ``__init__.py`` or, as a fallback, in ``<folder>/<folder>.py``. The primary class is instantiated directly: a scene IS a ``Node`` subclass in a ``.py`` file, so importing the module is all the "script loading" there is. Raises: ValueError: If the path holds no usable ``Node`` subclass. """ p = Path(path) if p.is_dir(): return _load_folder(p) return _load_file(p)
[docs] def import_file(path: str | Path) -> ModuleType: """Import *path* as a module of its own, leaving :data:`sys.path` alone. The name is the same pure function of the path :func:`load_scene` uses, so a file reached either way is one module, and what sits beside it is importable by bare name while it runs. This is the import a tool wants when it needs the module rather than a node tree, such as watching a file for changes. A file already loaded is returned as it stands; use :func:`importlib.reload` on the result to run it again. """ resolved = Path(path).resolve() name = _module_name(resolved, is_folder=False) existing = sys.modules.get(name) if existing is not None: return existing _install_finders() return _run_file(name, resolved, resolved.read_text(encoding="utf-8"))[0]
# -- Module naming ------------------------------------------------------------ def _anchor_for(directory: Path) -> Path: """The directory a scene's module name is expressed relative to. The project root when the scene belongs to one, so that a name reads as the path a developer would say out loud; otherwise the scene's own directory, which every scene has. """ from ..project import find_project project_file = find_project(directory) if project_file is not None: return project_file.parent return directory def _readable_name(target: Path, anchor: Path) -> str: """*target* spelled as a module-name fragment, relative to *anchor*.""" try: parts = target.relative_to(anchor).parts except ValueError: parts = (target.name,) readable = "_".join(_NOT_IN_A_NAME.sub("_", part) for part in parts) return readable or _NOT_IN_A_NAME.sub("_", target.name) or "root" def _digest(target: Path) -> str: return hashlib.sha256(str(target).encode("utf-8")).hexdigest()[:_DIGEST_LENGTH] def _module_name(resolved: Path, *, is_folder: bool) -> str: """The :data:`sys.modules` key for the scene at *resolved*.""" target = resolved if is_folder else resolved.with_suffix("") return f"{_NAME_PREFIX}{_readable_name(target, _anchor_for(resolved.parent))}_{_digest(resolved)}" def _beside_package_name(directory: Path) -> str: """The :data:`sys.modules` key for *directory*'s package of neighbours.""" return f"{_BESIDE_PREFIX}{_readable_name(directory, _anchor_for(directory))}_{_digest(directory)}" # -- Freshness ---------------------------------------------------------------- def _source_signature(path: Path) -> Any: """What one file looks like to the cache: a digest of its bytes.""" return hashlib.sha256(path.read_bytes()).digest() def _file_signature(path: Path, neighbours: Iterable[Path]) -> Any: """What a file scene and everything it imported from beside it look like.""" entries = [] for source in sorted(neighbours): try: entries.append((str(source), _source_signature(source))) except OSError: entries.append((str(source), None)) return (_source_signature(path), tuple(entries)) def _folder_signature(folder: Path) -> Any: """What every ``.py`` file under *folder* looks like to the cache.""" entries = [] for source in folder.rglob("*.py"): entries.append((str(source.relative_to(folder)), _source_signature(source))) return tuple(sorted(entries)) def _cached_class(resolved: Path, signature: Any) -> type | None: """The class compiled from this exact source, if it is still the source.""" key = str(resolved) entry = _compiled.get(key) if entry is None: return None if entry[0] != signature: del _compiled[key] return None return entry[1] def _remember_class(resolved: Path, signature: Any, node_cls: type) -> None: """Hold *node_cls* against the source it was compiled from.""" _compiled[str(resolved)] = (signature, node_cls) def _forget_edited_neighbours(directory: Path) -> None: """Drop every neighbour of *directory* whose file has changed since it ran. A module left in :data:`sys.modules` is never executed again, so recompiling a scene whose helper was edited would still run the helper's old code. """ prefix = f"{_beside_package_name(directory)}." package = _beside_packages.get(str(directory)) for name in [n for n in sys.modules if n.startswith(prefix)]: recorded = _beside_signatures.get(name) if recorded is None: continue source, signature = recorded try: current: Any = _source_signature(source) except OSError: current = None if current == signature: continue del sys.modules[name] _beside_signatures.pop(name, None) if package is not None: package.__dict__.pop(name[len(prefix) :], None) # -- Modules beside a file scene ---------------------------------------------- def _sibling_source(directory: Path, name: str) -> Path | None: """The file in *directory* that a top-level ``import name`` would run.""" module = directory / f"{name}.py" if module.is_file(): return module package_init = directory / name / "__init__.py" if package_init.is_file(): return package_init return None def _beside_package(directory: Path) -> ModuleType: """The package *directory*'s neighbours are loaded as submodules of.""" key = str(directory) package = _beside_packages.get(key) if package is not None and sys.modules.get(package.__name__) is package: return package name = _beside_package_name(directory) spec = importlib.machinery.ModuleSpec(name, None, is_package=True) spec.submodule_search_locations = [str(directory)] package = importlib.util.module_from_spec(spec) sys.modules[name] = package _beside_packages[key] = package return package class _SiblingScope: """One scene directory, reachable by bare name for as long as a scene runs. A neighbour is registered under the directory's package name, which is where it stays. It is *also* bound to its bare name while this scope is the active one, because that binding is what an ``import`` inside the scene looks at first: without it a second scene in the same directory would run the same helper a second time, and ``import helpers.thing`` could not find its parent at all. Whatever the bare name displaced, an installed package or a standard library module, is put back when the scope ends. Displacing it in the first place is the point: a file beside the script wins under ``python`` too. """ def __init__(self, directory: Path) -> None: self.directory = directory self.package = _beside_package(directory) self._bare: dict[str, ModuleType] = {} self._displaced: dict[str, ModuleType | None] = {} self._stashed: dict[str, ModuleType] = {} prefix = f"{self.package.__name__}." for name, module in list(sys.modules.items()): if name.startswith(prefix) and "." not in name[len(prefix) :]: self.bind(name[len(prefix) :], module) @property def loaded(self) -> tuple[Path, ...]: """The neighbouring files this directory's scenes have run.""" files = (getattr(module, "__file__", None) for module in self._bare.values()) return tuple(Path(file) for file in files if file) def spec_for(self, name: str) -> Any: """The spec for a neighbour called *name*, or None if there is none.""" if _sibling_source(self.directory, name) is None: return None try: finder = _BESIDE_HOOK(str(self.directory)) except ImportError: return None spec = finder.find_spec(f"{self.package.__name__}.{name}", None) if spec is not None and isinstance(spec.loader, _SiblingLoader): spec.loader.claim(name, self) return spec def bind(self, name: str, module: ModuleType) -> None: self._displaced.setdefault(name, sys.modules.get(name)) self._bare[name] = module sys.modules[name] = module def unbind(self, name: str) -> None: self._bare.pop(name, None) displaced = self._displaced.pop(name, None) if displaced is None: sys.modules.pop(name, None) else: sys.modules[name] = displaced def record(self, module: ModuleType) -> None: """Note the source a neighbour was compiled from, for the next load.""" file = getattr(module, "__file__", None) if not file: return try: _beside_signatures[module.__name__] = (Path(file), _source_signature(Path(file))) except OSError: pass def deactivate(self) -> None: """Put :data:`sys.modules` back as it was, so an inner scene sees its own.""" for name, displaced in self._displaced.items(): for key in [k for k in sys.modules if k == name or k.startswith(f"{name}.")]: self._stashed[key] = sys.modules.pop(key) if displaced is not None: sys.modules[name] = displaced def activate(self) -> None: """Re-apply the bare names over whatever holds them now.""" for name in self._bare: self._displaced[name] = sys.modules.get(name) sys.modules.update(self._stashed) self._stashed.clear() def retire(self) -> None: """Give up the bare names for good. The namespaced modules stay.""" self.deactivate() self._stashed.clear() #: The scene directories currently running, innermost last. A scene may load #: another scene from its own top-level code, and the inner one's neighbours #: must win over the outer one's for as long as it runs. _scopes: list[_SiblingScope] = [] @contextmanager def _sibling_imports(directory: Path) -> Iterator[_SiblingScope]: """Make *directory*'s neighbours importable by bare name for the block.""" outer = _scopes[-1] if _scopes else None if outer is not None: outer.deactivate() scope = _SiblingScope(directory) _scopes.append(scope) try: yield scope finally: _scopes.pop() scope.retire() if outer is not None: outer.activate() # -- Running the source ------------------------------------------------------- class _SourceOnlyLoader(importlib.machinery.SourceFileLoader): """A loader that compiles the file, never a cached ``.pyc`` beside it.""" def get_code(self, fullname: str) -> Any: source = importlib.util.decode_source(self.get_data(self.path)) return compile(source, self.path, "exec", dont_inherit=True) class _SiblingLoader(_SourceOnlyLoader): """Runs a neighbour of a scene, under the scene directory's package name. The bare name is bound before the module's code runs, so that a neighbour importing a neighbour, or importing itself back, sees the module that is already under way rather than starting a second copy of it. """ def claim(self, bare_name: str, scope: _SiblingScope) -> None: self._bare_name = bare_name self._scope = scope def exec_module(self, module: ModuleType) -> None: scope = getattr(self, "_scope", None) if scope is None: super().exec_module(module) return name = self._bare_name scope.bind(name, module) try: super().exec_module(module) except BaseException: scope.unbind(name) raise setattr(scope.package, name, module) scope.record(module) _SOURCE_ONLY_HOOK = importlib.machinery.FileFinder.path_hook((_SourceOnlyLoader, importlib.machinery.SOURCE_SUFFIXES)) _BESIDE_HOOK = importlib.machinery.FileFinder.path_hook((_SiblingLoader, importlib.machinery.SOURCE_SUFFIXES)) class _SceneReloadLoader(_SourceOnlyLoader): """Re-runs a scene module, with its directory's neighbours reachable again.""" def exec_module(self, module: ModuleType) -> None: directory = Path(self.path).parent _forget_edited_neighbours(directory) with _sibling_imports(directory): super().exec_module(module) class _SceneModuleFinder: """Finds a scene module by the synthetic name it was loaded under. A name derived from a path is on no search path, so nothing could resolve it a second time. :func:`importlib.reload`, which is how a scene is hot reloaded, throws away ``__spec__`` and re-resolves the name, and would fail without this. The answer is always the file the module was loaded from. """ @staticmethod def find_spec(fullname: str, path: Any = None, target: Any = None) -> Any: entry = _scene_files.get(fullname) if path is not None or entry is None: return None file, package_path = entry loader = _SceneReloadLoader(fullname, str(file)) locations = [str(package_path)] if package_path is not None else None return importlib.util.spec_from_file_location( fullname, str(file), loader=loader, submodule_search_locations=locations ) class _SceneSubmoduleFinder: """Finds the files inside a scene package, and only those. Everything a scene package holds is loaded from its source for the same reason the scene's own entry file is: an edit saved and reloaded in the same second, at the same size, would otherwise come back as the code from before the save. This covers a folder scene's own files and the relative imports between a scene directory's neighbours. Names outside a scene package are left to the rest of :data:`sys.meta_path`. """ @staticmethod def find_spec(fullname: str, path: Any = None, target: Any = None) -> Any: if path is None or "." not in fullname or not fullname.startswith(_SCENE_PREFIXES): return None for entry in path: try: finder = _SOURCE_ONLY_HOOK(entry) except ImportError: continue spec = finder.find_spec(fullname, target) if spec is not None: return spec return None class _SiblingFinder: """Resolves a bare ``import name`` to the file beside the running scene. Inert unless a scene's own top-level code is running, which is the window in which a scene may name its neighbours. """ @staticmethod def find_spec(fullname: str, path: Any = None, target: Any = None) -> Any: if path is not None or "." in fullname or not _scopes: return None return _scopes[-1].spec_for(fullname) def _install_finders() -> None: """Put the scene finders on :data:`sys.meta_path`, on the first load only. Ahead of the path finder, which would otherwise serve a scene's files from their bytecode cache and a scene's neighbour from an installed package of the same name. Importing this module installs nothing. """ for finder in (_SiblingFinder, _SceneSubmoduleFinder, _SceneModuleFinder): if not any(installed is finder for installed in sys.meta_path): sys.meta_path.insert(0, finder) def _module_from_source(name: str, file: Path, source: str, package_path: Path | None = None) -> ModuleType: """Register a module named *name* and run *source* in it. The text passed in is the text that runs. The import machinery would otherwise be free to run a cached ``.pyc`` instead, and it decides whether one is stale from the source's modification time in whole seconds and its size in bytes: an editor that writes a scene and loads it back in the same second gets the code from before its own save whenever the rewrite happened to land on the same number of bytes. Compiling what was read closes that. *package_path* makes the module a package rooted at that folder, so the relative imports in a folder scene resolve. """ if package_path is not None: spec = importlib.util.spec_from_file_location(name, str(file), submodule_search_locations=[str(package_path)]) else: spec = importlib.util.spec_from_file_location(name, str(file)) if spec is None: raise ValueError(f"Cannot load scene from {file}") module = importlib.util.module_from_spec(spec) sys.modules[name] = module _scene_files[name] = (file, package_path) try: # A scene is code by design; this runs it, which is what loading one is. exec(compile(source, str(file), "exec"), module.__dict__) except BaseException: sys.modules.pop(name, None) _scene_files.pop(name, None) raise return module def _run_file(name: str, resolved: Path, source: str) -> tuple[ModuleType, tuple[Path, ...]]: """Run a single-file module with its directory's neighbours reachable.""" directory = resolved.parent _forget_edited_neighbours(directory) with _sibling_imports(directory) as scope: module = _module_from_source(name, resolved, source) return module, scope.loaded def _load_file(path: Path) -> Node: """Run the file and instantiate the class it names.""" from ..node import Node resolved = path.resolve() key = str(resolved) signature = _file_signature(resolved, _scene_neighbours.get(key, ())) cached = _cached_class(resolved, signature) if cached is not None: root: Node = cached() return root source = path.read_text(encoding="utf-8") class_name = primary_node_class_from_source(source, path=path) if class_name is None: raise ValueError(f"No Node subclass found in {path}") _install_finders() module, neighbours = _run_file(_module_name(resolved, is_folder=False), resolved, source) _scene_neighbours[key] = neighbours node_cls = getattr(module, class_name, None) if node_cls is None or not isinstance(node_cls, type) or not issubclass(node_cls, Node): raise ValueError(f"Class {class_name!r} in {path} is not a Node subclass") _remember_class(resolved, _file_signature(resolved, neighbours), node_cls) root = node_cls() return root def _drop_package(name: str) -> None: """Forget a scene package and its submodules from a previous load.""" for loaded in [n for n in sys.modules if n == name or n.startswith(f"{name}.")]: del sys.modules[loaded] _scene_files.pop(loaded, None) def _package_module(name: str, folder: Path, init: Path, init_source: str | None) -> ModuleType: """The package a folder scene runs as. A folder without ``__init__.py`` still loads: it becomes an empty package whose files are importable through it, which is what the import machinery makes of such a folder anyway. """ if init_source is not None: return _module_from_source(name, init, init_source, package_path=folder) spec = importlib.machinery.ModuleSpec(name, None, is_package=True) spec.submodule_search_locations = [str(folder)] module = importlib.util.module_from_spec(spec) sys.modules[name] = module return module def _load_folder(path: Path) -> Node: from ..node import Node # Imported here rather than at module level: SceneModule needs parso, and # this module sits on the eager `import simvx.core` chain (via hot_reload), # which must stay importable in the browser runtime where parso is not # shipped. Folder scenes pay for parso only when one is actually loaded. from .scene_module import SceneModule if not SceneModule.is_folder_scene(path): raise ValueError(f"{path} is not a scene module") resolved = path.resolve() signature = _folder_signature(resolved) cached = _cached_class(resolved, signature) if cached is not None: root: Node = cached() return root init = path / "__init__.py" namespaced = path / f"{path.name}.py" init_source = init.read_text(encoding="utf-8") if init.is_file() else None _install_finders() package_name = _module_name(resolved, is_folder=True) _drop_package(package_name) package = _package_module(package_name, path, init, init_source) class_name: str | None = None holder = package if init_source is not None: candidate = primary_node_class_from_source(init_source, path=init) if candidate is not None and hasattr(package, candidate): class_name = candidate if class_name is None and namespaced.is_file(): text = namespaced.read_text(encoding="utf-8") candidate = primary_node_class_from_source(text, path=namespaced) if candidate is not None: attribute = _NOT_IN_A_NAME.sub("_", path.name) sub = _module_from_source(f"{package_name}.{attribute}", namespaced, text) setattr(package, attribute, sub) if hasattr(sub, candidate): class_name = candidate holder = sub if class_name is None: raise ValueError(f"No Node subclass resolvable in scene module {path}") node_cls = getattr(holder, class_name) if not isinstance(node_cls, type) or not issubclass(node_cls, Node): raise ValueError(f"Class {class_name!r} in {path} is not a Node subclass") _remember_class(resolved, signature, node_cls) root = node_cls() return root