Source code for simvx.editor.scene_file_ops

"""Scene file operations mixin for State."""

import hashlib
import logging
from pathlib import Path
from typing import TYPE_CHECKING

from simvx.core import Node
from simvx.core.scene_io import SceneFile, SceneModule, load_scene, parse_source

from .scene_diff import (
    INFORMATIONAL,
    ReportEntry,
    _canonical_var_names,
    _source_children,
    apply_runtime_diff,
    file_baseline,
)
from .workspace_tabs import SceneTabState, UntitledTabState

if TYPE_CHECKING:
    from .state import State

log = logging.getLogger(__name__)


[docs] class ScenePlan: """A save worked out in full, with nothing yet written. A scene file is not always a thing the editor can write back exactly as the author left it. Children built in a loop, in a conditional or in a helper method have no statement the save can match against the scene, so it writes them out again beside the ones the file already builds; removing a child takes the author's own statements that stood on it; and a value with no source form never reaches the file at all. All of that is known before anything is written -- the reconciliation works on a parsed copy in memory -- so it is offered here rather than reported afterwards over a file that has already been overwritten. :meth:`SceneFileOps.plan_save` produces one and writes nothing; :meth:`SceneFileOps.commit_save` writes it. The two together are :meth:`SceneFileOps.save_scene`, which commits a clean plan itself and puts a plan with a report in front of the user first -- or commits that too, when there is no dialog wired to ask with and so nobody to ask. A plan is only good for the file and the tree it was made from; see :meth:`overtaken`, which the commit consults before it writes anything. """ __slots__ = ("path", "report", "_document", "_root", "_tab", "_baseline", "_fresh", "_kept", "_target") def __init__( self, path: Path, report: list[ReportEntry], *, document: SceneFile | SceneModule, root: Node, tab: SceneTabState | None, baseline: dict | None, fresh: dict | None, kept: dict[Node, set[str]], target: str | None, ) -> None: #: Where the save would write. self.path = path #: One :class:`~simvx.editor.scene_diff.ReportEntry` per thing this save #: would do beyond carrying the user's edits across: a value the file #: will not end up with, a statement the reconciliation would take out #: of ``__init__``, a child it could not account for and wrote a second #: construction for. Empty for a save that writes the scene and nothing #: else. Each entry says which class it belongs to; see #: :attr:`destructive`. self.report = report # The parsed file with the reconciliation already applied to it, which # the commit does no more than write out. self._document = document self._root = root self._tab = tab self._baseline = baseline self._fresh = fresh self._kept = kept # What the destination held when the plan was worked out, so the commit # can tell whether it is still writing over the copy it read. self._target = target
[docs] @property def clean(self) -> bool: """Would this save carry the whole scene and change nothing else?""" return not self.report
[docs] @property def destructive(self) -> bool: """Would this save change the author's own file beyond their edits? True when any entry of the report is :data:`~simvx.editor.scene_diff.DESTRUCTIVE`: statements swept out of ``__init__`` with a removed child, an attribute left unbound, a child written a second time because the file builds it where the reconciliation cannot see it. That is the class worth stopping a user for, and it is what :meth:`SceneFileOps.save_scene` puts a prompt in front of. A report of nothing but the other class is still a report -- the save did not carry everything, the tab stays modified and every entry reaches the log -- but it is not a question: the file keeps the line it had, and asking on every save of a scene holding one such slot would teach the user to click through the prompt without reading it. """ return any(entry.destructive for entry in self.report)
[docs] def overtaken(self) -> str | None: """What has moved since this plan was worked out, if anything. A plan holds a document parsed from one copy of a file and bookkeeping keyed by the nodes of one tree, and it is written some time after both were read: the user is asked about it first, and the question invites them to go and edit those lines by hand. Either can move in the meantime. Writing a document parsed from text nobody has any more would drop whatever the author has since written; recording a baseline keyed by nodes the tab no longer holds -- the file watcher reloads a tab by replacing its root outright -- would leave every later save reporting every kept slot, and the tab could never go clean again. So the answer is a sentence naming what moved, or ``None`` when the plan still describes the world it was made in. """ if _target_fingerprint(self.path) != self._target: return f"{self.path.name} has changed on disk since this save was worked out" if self._tab is not None and self._tab.scene_tree.root is not self._root: return f"the scene {self.path.name} was opened into has been reloaded since this save was worked out" return None
def _write(self) -> None: """Put the reconciled document on disk. :meth:`SceneFileOps.commit_save` only.""" if isinstance(self._document, SceneModule): self._document.save() else: self._document.save(self.path)
[docs] class SceneFileOps: """Mixin providing scene lifecycle and file dialog operations. Methods in this class are designed to be mixed into State, which provides the workspace, signals, and delegating properties they depend on. """ # Typed self for IDE support: at runtime this is always State. if TYPE_CHECKING: self: State
[docs] def new_scene(self, root_type: type = Node, *, populate: bool = False): """Create a new scene tab with the given root type.""" name = root_type.__name__ if root_type is not Node else "Root" tab = SceneTabState.create(root_type=root_type, name=name) if populate: from .default_scenes import populate_default_scene populate_default_scene(tab.scene_tree.root) self.workspace.add_scene_tab(tab) self.scene_changed.emit()
[docs] def open_scene(self, path: str | Path): """Load a scene from disk into the active tab, or a new tab.""" path = Path(path) if not path.exists(): return root = load_scene(str(path)) if not root: return # Check if already open, reload in place existing = self.workspace.find_scene_tab(path) if existing is not None: self.workspace.set_active(existing) # Load into the active scene tab (overwrite its state) tab = self.workspace.active_scene if tab: tab.scene_tree.set_root(root) tab.scene_path = path tab.tab_name = path.stem tab.selection.clear() tab.undo_stack.clear() tab.modified = False else: new_tab = SceneTabState.create(root_type=type(root), name=root.name) new_tab.scene_tree.set_root(root) new_tab.scene_path = path new_tab.tab_name = path.stem self.workspace.add_scene_tab(new_tab) tab = new_tab record_source_baseline(tab, root, path) self._add_recent(str(path)) self.scene_changed.emit()
[docs] def save_scene(self, path: str | Path | None = None, *, force: bool = False): """Save the current scene. Dispatches based on the active workspace tab: * **Untitled scratch buffer**: opens a file-save dialog. On accept the buffer's text is written to disk and the tab is promoted to a regular file-backed script tab. * **Script tab**: forwards to ``workspace.save_current_script``, which writes the editor text back to the file the tab edits. * **Scene tab** (or no active tab): falls through to the scene-save path. Scene save: The save is worked out first and written second (:meth:`plan_save`, :meth:`commit_save`). A plan that would change the author's own file beyond carrying the scene across -- take their statements out of ``__init__``, leave an attribute unbound, write a second construction for a child it could not account for -- is put in front of them first, and the file on disk is untouched until they accept (:attr:`ScenePlan.destructive`). Everything else is written straight away, including a save that could not carry a value into a line it will not overwrite: nothing of the author's is lost there, the file keeps what it said, and the message reaches the log and the tab's modified mark rather than a modal. The prompt needs a dialog wired to the editor (``_save_report_dialog``, installed by :class:`~simvx.editor.root.Root`); a caller with none, which means a headless or scripted save, commits either way and the report goes to the log as it always did. Returns ``True`` only for a save that has been written and carried the whole scene. A save waiting on the user answers ``False``, as does one that wrote a file the scene is still ahead of. ``force=True`` skips the unresolved-reference warning dialog (used by the dialog's "Save anyway" callback to avoid re-prompting). """ # Dispatch on active tab kind when no explicit path is given. if path is None: ws = self.workspace idx = ws.active_index if 0 <= idx < ws.tab_count: if ws.is_untitled_tab(idx): self._show_save_untitled_dialog(idx) return False if ws.is_script_tab(idx): ws.save_current_script() return True save_path = Path(path) if path else self.current_scene_path if not save_path: self._show_save_as_dialog() return False root = self.edited_scene.root if self.edited_scene else None if not root: return False # Unresolved-reference warning: scene references class names that only # exist in unsaved Untitled buffers (or otherwise have no on-disk file # yet). Allow saving anyway via the dialog, the .py file is still # syntactically valid; the import will fail at runtime until the user # saves the buffer to a real path that the scene file can import. if not force: unresolved = self._unresolved_class_names(root) if unresolved: self._show_unsaved_class_warning(unresolved, save_path) return False plan = self.plan_save(save_path) if plan is None: return False return self._ask_or_commit(plan)
def _ask_or_commit(self, plan: ScenePlan) -> bool: """Commit ``plan`` outright, or put it in front of the user first. The question is :attr:`ScenePlan.destructive`, not whether there is a report at all: a save that would rewrite the author's ``__init__`` is worth stopping for, and one that merely cannot carry a value into a line it will not touch is not. The second is written and warned about, which is what Godot, Unity and Unreal all do with a value their serialiser cannot express, and the warning still reaches the log and still leaves the tab modified. """ dialog = getattr(self, "_save_report_dialog", None) if not plan.destructive or dialog is None: return self.commit_save(plan) dialog.show_for(plan, on_confirm=lambda: self._commit_reviewed_plan(plan)) return False def _commit_reviewed_plan(self, plan: ScenePlan) -> bool: """Write a plan the user has approved, or ask again over a fresh one. The prompt tells the user which lines the save cannot carry, so the obvious thing for them to do is open the file and fix those lines, and the plan they were shown then describes text that is no longer on disk. The file watcher can also reload the tab underneath them. Either way the approval stands -- they asked for this scene to be saved to this path -- but the plan behind it does not, so it is worked out again and put back in front of them, since a fresh plan may have a different report to approve. A plan whose scene is no longer the one being edited is dropped instead: re-planning would take whatever scene is now in front of the user and write it to the other tab's file. Which scene that is, is :meth:`_scene_tab_for_save`'s answer, not the workspace's active tab: clicking into a script to fix one of the lines the prompt names is the obvious thing to do while it is open, and it leaves no scene tab active at all. That is not another scene, so it is not a reason to throw the save away. """ overtaken = plan.overtaken() if overtaken is None: return self.commit_save(plan) if plan._tab is not None and plan._tab is not self._scene_tab_for_save(): log.error("Nothing written to %s: %s", plan.path.name, overtaken) return False log.error("Working out the save of %s again: %s", plan.path.name, overtaken) fresh = self.plan_save(plan.path) if fresh is None: return False return self._ask_or_commit(fresh) def _scene_tab_for_save(self): """The scene tab a save is about, or ``None`` when no scene is open. The tab holding the tree :attr:`State.edited_scene` hands back, which is the active scene tab where there is one and the last one otherwise. A save is about a scene, and a script tab holding the keyboard focus does not make the scene in the editor stop being the one the user is working on: reading the workspace's active tab instead would plan a save with no tab to record its baseline on, and leave the tab's record of the file stale enough that every later save reported every slot it kept. """ return self._active_or_last_scene()
[docs] def plan_save(self, path: str | Path | None = None) -> ScenePlan | None: """Work out the save to ``path`` in full, writing nothing. For an existing on-disk source: parse it and reconcile the runtime tree against the parsed source via :func:`apply_runtime_diff`, which preserves comments, blank lines, hand-written code, and import ordering. For a brand-new scene (the path does not exist): emit greenfield via :meth:`SceneFile.from_runtime`. Folder scenes (when ``path`` is a directory) route through :class:`SceneModule`. In every case the result is a parsed document held in memory and a :attr:`ScenePlan.report` of everything the save would do beyond carrying the user's edits across; the file itself is not touched until :meth:`commit_save`. A value the file would end up without is one entry of that report. Both routes can produce one. Greenfield produces one when the emitter cannot write a value at all -- a texture built from pixels in memory, say. Preserve mode produces one for that same case, since a value with no source form has none here either, and additionally when the file spells a slot as an expression this layer will not overwrite; the test is then whether the file yields what the scene holds, so the message stops once the author either edits that expression or makes it yield the value the scene has -- repointing the binding it names counts, and leaves the line itself untouched. Answering that question costs a read of the file being written, which for a scene means *running* it: scenes are code, so there is no way to learn what an expression yields without executing the module. It is paid only when the reconciliation actually declined something, and it is paid here, while the copy on disk is still the one the author last saw. ``None`` when there is nothing to plan: no destination and none recorded on the tab, or no scene open. Neither opens a dialog -- that is :meth:`save_scene`'s business, not the planning's. """ save_path = Path(path) if path else self.current_scene_path if not save_path: return None root = self.edited_scene.root if self.edited_scene else None if not root: return None # Taken before the first read, so that anything landing on the file from # here on -- including during the reads below -- is caught by the commit. target = _target_fingerprint(save_path) active_tab = self._scene_tab_for_save() hints = active_tab.identity_hints if active_tab is not None else None # What the file about to be written already says. The tab has it when # this is a save back to the file it was read from. A Save As over some # *other* existing file is a diff against source this session never # read, so read it now: falling back to what that file's own text can # prove would drop every edit under a spelling this layer cannot parse # an answer out of, which is the whole reason the baseline exists. baseline = None if active_tab is not None and _same_file(active_tab.scene_path, save_path): baseline = active_tab.file_baseline or None if baseline is None and save_path.exists(): baseline = _baseline_of_file(save_path, root) # A slot the diff declines can stop diverging without its line changing: # the author repoints the binding it names. So what the file *yields* is # asked for again, from the copy still on disk, and only once the diff # has found something to ask about -- reading a scene runs it, and no # save that had nothing to declare should pay for that. fresh: dict | None = None def read_the_file_again() -> dict | None: nonlocal fresh fresh = _baseline_of_file(save_path, root) return fresh refusals: list[ReportEntry] = [] kept: dict[Node, set[str]] = {} document: SceneFile | SceneModule if save_path.is_dir() or SceneModule.is_folder_scene(save_path): document = SceneModule.load(save_path) kept = apply_runtime_diff( document.root.scene_class(), root, identity_hints=hints, baseline=baseline, refresh_baseline=read_the_file_again, report=refusals, ) elif save_path.exists(): document = SceneFile.load(save_path) kept = apply_runtime_diff( document.scene_class(), root, identity_hints=hints, baseline=baseline, refresh_baseline=read_the_file_again, report=refusals, ) else: # A file written from scratch has no author's text to destroy, so # every refusal the emitter makes is a value that did not reach the # file and nothing more. emitted: list[str] = [] document = SceneFile.from_runtime(root, report=emitted) refusals.extend(ReportEntry(message, INFORMATIONAL) for message in emitted) return ScenePlan( save_path, refusals, document=document, root=root, tab=active_tab, baseline=baseline, fresh=fresh, kept=kept, target=target, )
[docs] def commit_save(self, plan: ScenePlan) -> bool: """Write what ``plan`` described, and bring the tab into line with it. Returns ``True`` for a save that carried the whole scene. A plan with a report answers ``False`` and leaves the tab marked modified: the file on disk no longer holds everything the scene has, and calling that clean would tell the user their work is safe when part of it is not. Each entry goes to the log as an error, which is how it reaches the console panel. A plan the world has moved under (:meth:`ScenePlan.overtaken`) is not written at all: it answers ``False`` and says why in the log. Work the plan out again to save that scene -- which is what :meth:`save_scene` does when the user approves a plan that has gone stale while they were reading it. """ overtaken = plan.overtaken() if overtaken is not None: log.error("Nothing written to %s: %s", plan.path.name, overtaken) return False plan._write() # Marked modified rather than merely left alone, because a Save As of an # already clean tab starts from clean and would otherwise stay there. # All of it goes to the tab the plan was made for, which need not be the # one in front of the user by now: the state properties would follow the # active tab and point some other scene's tab at this file. if plan._tab is not None: # The freshest reading wins as the record to carry declined slots # forward from: it is what the file said moments ago, where the one # this save started with may predate an edit the author made in # their own editor. record_source_baseline( plan._tab, plan._root, plan.path, previous=plan._fresh or plan._baseline, kept=plan._kept ) plan._tab.scene_path = plan.path plan._tab.modified = not plan.clean else: self.current_scene_path = plan.path self._modified = not plan.clean for entry in plan.report: log.error("%s: %s", plan.path.name, entry) self._add_recent(str(plan.path)) self.scene_modified.emit() return plan.clean
# ------------------------------------------------------------------ # Unresolved-reference detection # ------------------------------------------------------------------ def _unresolved_class_names(self, root: Node) -> list[str]: """Return class names referenced in *root*'s tree that lack a saved file. A class is considered "unsaved" when its name appears in an open :class:`UntitledTabState`'s top-level ``class`` statements. Built-in engine classes and project classes already on disk are always considered resolved: they are saved by definition. The check is intentionally name-based rather than module-based: an Untitled buffer has no real ``__module__`` we could reach via ``importlib``, so the only stable signal we have is the source text. """ unsaved = self._unsaved_class_names() if not unsaved: return [] seen: set[str] = set() result: list[str] = [] for node in [root, *root.find_all(Node)]: cls_name = type(node).__name__ if cls_name in unsaved and cls_name not in seen: seen.add(cls_name) result.append(cls_name) return result def _unsaved_class_names(self) -> set[str]: """Top-level class names defined in any open Untitled scratch buffer.""" names: set[str] = set() for tab in self.workspace._tabs: if not isinstance(tab, UntitledTabState): continue try: tree = parse_source(tab.editor.text) except Exception: continue if tree.errors: continue for cls in tree.iter_classes(): names.add(cls.name.value) return names def _show_unsaved_class_warning(self, names: list[str], save_path: Path) -> None: """Open the modal warning, hooked to retry the save on confirm.""" dialog = getattr(self, "_unsaved_class_warning_dialog", None) if dialog is None: return # no UI wired (e.g. headless test that didn't install one) dialog.show_for( class_names=names, on_confirm=lambda: self.save_scene(save_path, force=True), ) def _show_save_untitled_dialog(self, index: int) -> None: """Prompt the user for a destination, then promote the Untitled tab.""" dlg = self._file_dialog if dlg is None: return dlg.file_selected.clear() def _on_chosen(p): self.workspace.promote_untitled_to_file(index, p) dlg.file_selected.connect(_on_chosen) start = str(self.project_path / "src") if self.project_path is not None else None dlg.show(mode="save", path=start, filter="*.py") def _show_open_dialog(self): if not self._file_dialog: return self._file_dialog.file_selected.clear() self._file_dialog.file_selected.connect(self.open_scene) start = str(self.current_scene_path.parent) if self.current_scene_path else None self._file_dialog.show(mode="open", path=start, filter="*.py") def _show_save_as_dialog(self): if not self._file_dialog: return self._file_dialog.file_selected.clear() self._file_dialog.file_selected.connect(lambda p: self.save_scene(p)) start = str(self.current_scene_path) if self.current_scene_path else None self._file_dialog.show(mode="save", path=start, filter="*.py") def _add_recent(self, path: str): if path in self.recent_files: self.recent_files.remove(path) self.recent_files.insert(0, path) self.recent_files = self.recent_files[:10]
def _target_fingerprint(path: Path) -> str | None: """What the destination holds right now, in one comparable value. A digest of the bytes rather than a size and a timestamp: the question is whether the text a plan was parsed from is still the text on disk, and the bytes answer it exactly, where a stamp only says a write happened. Scene files are small enough that reading one again costs nothing next to the parse and the import a save already pays for. ``None`` for a path with nothing at it, which is an answer in its own right: a greenfield plan that finds a file where it left empty space has been overtaken as surely as a preserving one whose file has been edited. A folder scene is fingerprinted over every module in the folder, since that is the unit :meth:`SceneModule.save` writes. """ digest = hashlib.sha256() try: if path.is_dir(): for module in sorted(path.glob("*.py")): digest.update(module.name.encode("utf-8")) digest.update(module.read_bytes()) else: digest.update(path.read_bytes()) except OSError: return None return digest.hexdigest() def _same_file(left: Path | None, right: Path | None) -> bool: """Do two paths name the same file, symlinks and ``..`` aside?""" if left is None or right is None: return False return Path(left).resolve() == Path(right).resolve()
[docs] def record_source_baseline( tab, root: Node, path: Path | None = None, *, previous: dict | None = None, kept: dict[Node, set[str]] | None = None, ) -> None: """Record what the file ``tab`` now describes says about the scene in it. Every route by which a tab comes to stand for a file goes through here -- :meth:`SceneFileOps.open_scene`, the live-file import and the reload the file watcher triggers (:meth:`LiveFileOps.open_file`), the project session's own open, and each save. Two things are kept, both keyed by runtime object so they survive renames and property edits: which source var each child came from, and what the file said about every constructor kwarg. Miss a route and the keys are objects from a tree that no longer exists, at which point every save reports every kept slot and the tab can never go clean again. After a save, ``previous`` and ``kept`` correct the one place the file and the tree disagree. The tree is otherwise the right answer -- the save just wrote it out -- but a slot the diff declined to write still says whatever it said before, so its old value is carried forward. Recording the live one there would have the file claiming to hold the very edit it dropped, and the next save would go clean over it. """ scene_class = _parse_scene_class(path) baseline = file_baseline(root, scene_class=scene_class) for node, kwargs in (kept or {}).items(): for kwarg in kwargs: was = (previous or {}).get(node, {}).get(kwarg) slot = baseline.get(node, {}).get(kwarg) if was is not None and slot is not None: baseline[node][kwarg] = slot._replace(value=was.value) tab.identity_hints = _build_initial_identity_hints(root, scene_class) tab.file_baseline = baseline
def _baseline_of_file(path: Path, runtime_root: Node) -> dict | None: """What the file at ``path`` says, keyed by the nodes of ``runtime_root``. Asked when a save is about to write over a file whose current answer matters: one this session never opened, or one holding a slot the diff has just declined. The tree the file yields is used only to be read, and its nodes are matched to the runtime's by canonical var name. Reading it means **running it**. A scene is a Python module and the values this needs are whatever its expressions evaluate to, so there is no parsing-only version of this question -- ``load_scene`` imports and executes the file, including any module-level code the author put there. Callers are expected to have a reason. ``None`` when the file will not load, in which case the save falls back to whatever record it already had, or failing that to the little the file's own text can prove. """ try: source_root = load_scene(str(path)) except Exception: log.debug("Could not load %s to see what it already says", path, exc_info=True) return None if source_root is None: return None return file_baseline(runtime_root, source_root, _parse_scene_class(path)) def _parse_scene_class(path: Path | None): """The parsed scene class at ``path``, for the text each kwarg is spelled as. ``None`` for a path that does not exist or will not parse: the baseline then records values without their spellings, which costs the ability to notice that the author has since rewritten a line by hand, and nothing else. """ if path is None or not Path(path).exists(): return None try: if Path(path).is_dir() or SceneModule.is_folder_scene(path): return SceneModule.load(path).root.scene_class() return SceneFile.load(path).scene_class() except Exception: log.debug("Could not parse %s for a save baseline", path, exc_info=True) return None def _build_initial_identity_hints(root: Node, scene_class=None) -> dict[Node, str]: """Map each top-level runtime child to the source var it is bound to. At scene-load time the runtime tree mirrors the on-disk source, so the ``n``-th ``add_child`` statement is what built the ``n``-th child and the var it binds is that child's name in the file. That is what the save has to be told, because it is the name it will look the child up by: an author is free to write ``hero = Panel()`` for a node the emitter would have called ``panel``, and a hint of ``panel`` matches nothing, leaving the save to remove the author's lines and write the child back -- without the children it had of its own, which only those lines built. The returned dict survives subsequent renames (``Node.name`` may change but the dict is keyed by Node identity) and feeds :func:`simvx.editor.scene_diff.apply_runtime_diff` so save can issue an in-place rename instead of remove + add. Falls back to the canonical names when the file cannot be read, or when it does not add exactly as many children as the tree has: a file that builds children in a loop or a helper method has no statement per child to line them up against, and a wrong pairing here is worse than none. """ children = list(root.children) if not children: return {} var_names: list[str | None] = list(_canonical_var_names(children)) if scene_class is not None: try: slots = _source_children(scene_class) except Exception: log.debug("Could not read the source's own child statements for identity hints", exc_info=True) slots = [] if len(slots) == len(children): var_names = [slot.var_name for slot in slots] return {child: name for child, name in zip(children, var_names, strict=True) if name is not None}