"""Live Python file operations mixin for State."""
import importlib
import inspect
import logging
import os
import sys
import traceback
from pathlib import Path
from typing import TYPE_CHECKING, Any
from simvx.core import Node, SceneTree, Vec2
from .scene_file_ops import record_source_baseline
from .workspace_tabs import SceneTabState
if TYPE_CHECKING:
from types import ModuleType
from .state import State
log = logging.getLogger(__name__)
[docs]
class LiveFileOps:
"""Mixin for opening Python scene files and working with live Node objects.
The editor imports Python files, identifies Node subclasses, instantiates
them, and manipulates the live objects directly. Properties are read from
instances via ``get_properties()``.
"""
if TYPE_CHECKING:
self: State
[docs]
def open_file(self, path: str | Path) -> None:
"""Open a Python file, import it, and instantiate its primary Node class."""
path = Path(path).resolve()
if not path.exists():
log.error("File not found: %s", path)
return
if path.suffix != ".py":
log.error("Not a Python file: %s", path)
return
module = self._import_module(path)
if module is None:
return
classification = self.classify_file(module, str(path))
classes = self.find_node_classes(module)
primary = self.get_primary_class(module, str(path))
if primary is None:
log.warning("No Node subclass found in %s", path)
return
instance = self.instantiate_class(primary)
if instance is None:
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)
tab = self.workspace.active_scene
if tab:
tab.scene_tree.set_root(instance)
tab.scene_path = path
tab.tab_name = path.stem
tab.source_file = str(path)
tab.source_class = primary
tab.source_module = module
tab.file_classification = classification
tab.selection.clear()
tab.undo_stack.clear()
tab.modified = False
else:
tab = self._build_live_tab(instance, path, primary, module, classification)
self.workspace.add_scene_tab(tab)
# This tab now stands for this file, and the tree it stands for is new:
# the watcher calls straight back in here on every external edit, so
# anything still keyed by the previous instance's nodes is stale.
record_source_baseline(tab, instance, path)
# Update state-level tracking
self.edited_file = str(path)
self.edited_module = module
self.edited_class = primary
self.file_classes = classes
self.file_classification = classification
self._add_recent(str(path))
self.scene_changed.emit()
[docs]
def instantiate_class(self, cls: type) -> Node | None:
"""Safely instantiate a Node subclass. Returns None on error."""
try:
return cls()
except Exception:
log.error("Failed to instantiate %s:\n%s", cls.__name__, traceback.format_exc())
# Create an error placeholder
placeholder = Node(name=f"{cls.__name__} (error)")
placeholder._script_error = True
return placeholder
[docs]
def classify_file(self, module: ModuleType, file_path: str) -> str:
"""Classify a Python file. Returns 'main', 'scene', or 'node'."""
classes = self.find_node_classes(module)
class_names = {name for name, _ in classes}
# 'main' -- has class named Main
if "Main" in class_names:
return "main"
# 'scene' -- class name matches filename (case-insensitive)
stem = Path(file_path).stem.lower()
for name, _ in classes:
if name.lower() == stem:
return "scene"
# 'node' -- everything else
return "node"
[docs]
def find_node_classes(self, module: ModuleType) -> list[tuple[str, type]]:
"""Find all Node subclasses defined in a module."""
result = []
for name, obj in inspect.getmembers(module, inspect.isclass):
if issubclass(obj, Node) and obj is not Node and obj.__module__ == module.__name__:
result.append((name, obj))
return result
[docs]
def get_primary_class(self, module: ModuleType, file_path: str) -> type | None:
"""Get the primary class to instantiate from a module.
Priority: Main > filename match > sole class > None
"""
classes = self.find_node_classes(module)
if not classes:
return None
class_dict = dict(classes)
# Priority 1: class named Main
if "Main" in class_dict:
return class_dict["Main"]
# Priority 2: class name matches filename (case-insensitive)
stem = Path(file_path).stem.lower()
for name, cls in classes:
if name.lower() == stem:
return cls
# Priority 3: sole class
if len(classes) == 1:
return classes[0][1]
return None
def _build_live_tab(self, instance: Node, path: Path, cls: type, module: Any, classification: str) -> SceneTabState:
"""Build a SceneTabState from an already-instantiated live node."""
import math
from simvx.core import Control as _Control
from simvx.core import Node2D, OrbitCamera3D
tree = SceneTree(screen_size=Vec2(800, 600))
tree.set_root(instance)
sub_mode = "2d" if isinstance(instance, (Node2D, _Control)) else "3d"
cam = OrbitCamera3D(name="EditorCamera")
cam.pitch = math.radians(-35.0)
cam.yaw = math.radians(30.0)
cam.distance = 8.0
cam.update_transform()
from .workspace_tabs import _SceneTabPlaceholder
return SceneTabState(
scene_tree=tree,
viewport_sub_mode=sub_mode,
editor_camera=cam,
placeholder=_SceneTabPlaceholder(name=f"Scene:{instance.name}"),
tab_name=path.stem,
scene_path=path,
source_file=str(path),
source_class=cls,
source_module=module,
file_classification=classification,
)
def _import_module(self, path: Path) -> ModuleType | None:
"""Import a Python file as a module. Returns None on error.
The scene loader owns the import: the module name is a pure function of
the resolved path, so two files never share one, and the module lands in
:data:`sys.modules` where hot reload and :mod:`pickle` can resolve it.
``import_file`` returns an already-loaded module as it stands, and
opening a file must serve what is on disk, so that case runs the file
again through :func:`importlib.reload`.
"""
from simvx.core.scene_io.loader import import_file
loaded_before = set(sys.modules)
try:
module = import_file(path)
if module.__name__ in loaded_before:
module = importlib.reload(module)
return module
except Exception:
log.error("Failed to import %s:\n%s", path, traceback.format_exc())
return None
[docs]
class FileWatcher:
"""Polls file mtimes to detect external modifications.
Usage::
watcher = FileWatcher()
watcher.watch("/path/to/player.py")
# Periodically:
changed = watcher.check() # returns list of changed paths
"""
def __init__(self):
self._watched: dict[str, float] = {} # path -> last known mtime
[docs]
def watch(self, path: str | Path) -> None:
"""Start watching a file. Records its current mtime."""
p = str(Path(path).resolve())
try:
self._watched[p] = os.stat(p).st_mtime
except OSError:
self._watched[p] = 0.0
[docs]
def unwatch(self, path: str | Path) -> None:
"""Stop watching a file."""
self._watched.pop(str(Path(path).resolve()), None)
[docs]
def watch_directory(self, directory: str | Path, suffix: str = ".py") -> None:
"""Watch all files with given suffix in a directory tree."""
d = Path(directory)
if d.is_dir():
for f in d.rglob(f"*{suffix}"):
self.watch(f)
[docs]
def check(self) -> list[str]:
"""Check all watched files for mtime changes. Returns list of changed paths."""
changed: list[str] = []
for path, last_mtime in list(self._watched.items()):
try:
current_mtime = os.stat(path).st_mtime
except OSError:
continue # file deleted or inaccessible
if current_mtime != last_mtime:
self._watched[path] = current_mtime
changed.append(path)
return changed
[docs]
def clear(self) -> None:
"""Stop watching all files."""
self._watched.clear()