"""Play Mode: Run/pause/stop lifecycle for in-editor game preview.
Manages scene serialization/restoration, camera switching, input routing,
and per-frame process/physics updates during play mode. The game runs in
a separate SceneTree with isolated input state so editor and game input
never interfere.
Usage:
play_mode = PlayMode(editor_state)
play_mode.start() # F5: serialize scene, begin processing
play_mode.toggle_pause() # F7: pause/resume processing
play_mode.stop() # F6: restore pre-play scene state
# Called each frame by the editor's main loop:
play_mode.update(dt)
"""
import logging
import math
import os
import subprocess
import time
import traceback as _tb
from collections.abc import Callable
from pathlib import Path
from simvx.core import (
Camera3D,
Input,
Key,
MouseButton,
Node,
OrbitCamera3D,
ScenePhases,
SceneTree,
key_to_name,
step_scene_logic,
)
from simvx.core._scene_internal import _deserialize_node, _serialize_node
from simvx.core.debug.profiler import FrameProfiler
from simvx.core.hot_reload import HotReloadManager
from .state import State
log = logging.getLogger("simvx.play")
_hot_reload_log = logging.getLogger("simvx.editor.play_mode")
# Directories to skip when discovering watchable .py files under the project root.
_HOT_RELOAD_SKIP_DIRS = frozenset({".venv", "venv", "__pycache__", "node_modules", "dist", "build", ".git"})
# Orbit camera input sensitivity
_ORBIT_SENSITIVITY = math.radians(0.35) # radians per pixel of mouse movement
_ZOOM_SENSITIVITY = 1.5 # distance change per scroll step
_PITCH_MIN = math.radians(-85.0) # upper half-sphere: can't go below floor
_PITCH_MAX = math.radians(-2.0) # prevent flipping to underside
# Border colours for the viewport overlay (RGBA, 0-1 range)
_COLOUR_PLAYING = (0.2, 0.8, 0.2, 1.0) # Green: game is running
_COLOUR_PAUSED = (1.0, 0.6, 0.0, 1.0) # Orange: game is paused
_COLOUR_STOPPED = None # No border when stopped
# ======================================================================
# PlayMode
# ======================================================================
[docs]
class PlayMode:
"""Manages the run/pause/stop lifecycle for previewing games in the editor.
The game scene runs in an isolated SceneTree with its own input state.
The editor's process loop calls ``update(dt)`` every frame to drive
the game cooperatively (no threading).
"""
def __init__(self, state: State):
self._state = state
# Serialized snapshot of the scene before play started
self._saved_scene_data: dict | None = None
# Separate SceneTree for the running game. Its ``_own_input`` (created by
# SceneTree(isolated_input=True)) IS the game's isolated input; see the
# ``game_input`` property. No separate global-swap input container.
self._game_tree: SceneTree | None = None
# The game's own camera (if the scene contains one)
self._game_camera: Camera3D | None = None
# Runtime metrics
self._elapsed_time: float = 0.0
self._frame_count: int = 0
# Whether ready() has been called on the game tree
self._ready_called: bool = False
# Orbit camera drag state
self._orbit_dragging: bool = False
self._last_mouse: tuple[float, float] = (0.0, 0.0)
# Detached game window subprocess
self._detached_proc: subprocess.Popen | None = None
# Scene transition tracking
self._last_root: Node | None = None
# GPU game viewport renderer (created when an App with graphics is available)
self._game_viewport = None # GameViewportRenderer | None
self._game_viewport_enabled: bool = False
# Live resize tracking. The viewport panel calls
# ``ensure_game_viewport_size`` each frame; we debounce so a
# mid-drag splitter doesn't thrash GPU memory.
self._game_viewport_size: tuple[int, int] = (0, 0)
self._game_viewport_pending_size: tuple[int, int] = (0, 0)
self._game_viewport_stable_frames: int = 0
# Buffered UI events for game tree dispatch (drained during update)
self._ui_event_queue: list[dict] = []
# Optional profiler -- when set, update() times phases and samples
# per-node costs into it. Assigned by Root at init time.
self._profiler: FrameProfiler | None = None
# Hot-reload manager (created on start, torn down on stop). Watches
# project script files and class-swaps live nodes when sources change.
self._hot_reload_mgr: HotReloadManager | None = None
# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
[docs]
def start(self, on_tree_created=None) -> None:
"""Begin play mode (triggered by F5).
Serializes the current scene so it can be restored on stop,
creates an isolated game tree with its own input state, and
calls on_ready() on all game nodes.
Args:
on_tree_created: Optional callback ``fn(game_root)`` invoked after
the game tree is constructed but before ``on_ready()`` is called.
Used by the editor to load scripts onto the cloned tree.
"""
if self._state.is_playing:
return
root = self._get_root()
if root is None:
return
# Deep-serialize the scene for later restoration
self._saved_scene_data = _serialize_node(root)
# Clone the scene into a separate game tree
game_root = _deserialize_node(self._saved_scene_data)
if game_root is None:
return
self._game_tree = SceneTree(isolated_input=True)
self._game_tree.root = game_root
game_root._enter_tree(self._game_tree)
# Locate the game's Camera3D (if any) for rendering
self._game_camera = self._find_game_camera(game_root)
# Reset runtime metrics
self._elapsed_time = 0.0
self._frame_count = 0
self._ready_called = False
# User game code gets recovery behaviour, not crashes
Node.strict_errors = False
# Flip the state flags
self._state.is_playing = True
self._state.is_paused = False
self._state.play_state_changed.emit()
# Hook for script loading (before ready)
if on_tree_created is not None:
on_tree_created(game_root)
# Initialize the game tree -- call ready() on all nodes
self._call_ready(game_root)
self._ready_called = True
self._last_root = game_root
# Wire hot reload: watch project scripts for live class-swap. The
# toolbar toggle stores its preference on State; honour it now
# so the user's last choice persists across PlayMode restarts.
self._hot_reload_mgr = HotReloadManager(self._game_tree)
self._hot_reload_mgr.enabled = self._state.hot_reload_enabled
for path in self._discover_watchable_scripts():
try:
self._hot_reload_mgr.watch(path)
except Exception:
# Justified: a single un-importable script must not abort play start.
_hot_reload_log.exception("hot_reload: failed to watch %s", path)
[docs]
def set_hot_reload_enabled(self, enabled: bool) -> None:
"""Enable or disable script hot-reload during play mode.
Updates :attr:`State.hot_reload_enabled` AND, when a play
session is active, the live :class:`HotReloadManager` so the change
takes effect immediately without restarting PlayMode.
"""
self._state.hot_reload_enabled = bool(enabled)
if self._hot_reload_mgr is not None:
self._hot_reload_mgr.enabled = bool(enabled)
[docs]
def toggle_pause(self) -> None:
"""Toggle pause during play mode (triggered by F7).
When paused, process/physics updates are skipped but the scene
continues to render so the user can inspect the frozen state.
"""
if not self._state.is_playing:
return
self._state.is_paused = not self._state.is_paused
self._state.play_state_changed.emit()
[docs]
def stop(self) -> None:
"""Stop play mode and restore the pre-play scene (triggered by F6).
Destroys the game tree and input state, deserializes the saved
snapshot back into the editor's scene tree.
"""
if not self._state.is_playing:
return
# Clear play flags first
self._state.is_playing = False
self._state.is_paused = False
# Restore strict errors for editor's own code
Node.strict_errors = True
# Tear down the hot-reload manager (unwatch all files).
if self._hot_reload_mgr is not None:
for path in list(self._hot_reload_mgr.watched_files):
self._hot_reload_mgr.unwatch(path)
self._hot_reload_mgr = None
# Tear down the game tree
self._game_tree = None
self._ui_event_queue.clear()
# Restore scene from the saved snapshot
if self._saved_scene_data is not None:
restored_root = _deserialize_node(self._saved_scene_data)
if restored_root is not None:
self._state.edited_scene.set_root(restored_root)
self._saved_scene_data = None
# Stop any detached process
if self._detached_proc is not None:
self._detached_proc.terminate()
self._detached_proc = None
# Destroy game viewport renderer
self.destroy_game_viewport()
# Discard runtime references
self._game_camera = None
self._elapsed_time = 0.0
self._frame_count = 0
self._ready_called = False
self._last_root = None
# Notify listeners
self._state.play_state_changed.emit()
self._state.scene_changed.emit()
[docs]
def start_detached(self) -> None:
"""Start the game in a separate OS window via subprocess."""
if self._state.is_playing:
return
project_dir = self._state.project_path
if project_dir is None:
log.warning("Cannot start detached: no project path set")
return
try:
self._detached_proc = subprocess.Popen(["simvx", "run"], cwd=str(project_dir))
except FileNotFoundError:
log.error("'simvx' command not found, is simvx-core installed?")
return
self._state.is_playing = True
self._state.play_state_changed.emit()
[docs]
def stop_detached(self) -> None:
"""Stop the detached game window if running."""
if self._detached_proc is not None:
self._detached_proc.terminate()
self._detached_proc = None
if self._state.is_playing and self._game_tree is None:
self._state.is_playing = False
self._state.play_state_changed.emit()
[docs]
@property
def is_detached(self) -> bool:
"""Whether a detached game process is running."""
return self._detached_proc is not None
def _check_scene_transition(self) -> None:
"""Detect if the game tree's root changed (scene transition)."""
if self._game_tree is None:
return
current_root = self._game_tree.root
if current_root is not self._last_root:
self._last_root = current_root
self._game_camera = self._find_game_camera(current_root) if current_root else None
self._state.scene_changed.emit()
[docs]
def attach_profiler(self, profiler: FrameProfiler | None) -> None:
"""Attach (or detach) a :class:`FrameProfiler` for play-mode metrics.
When attached, :meth:`update` times each phase (``physics``,
``process``, ``ui``, ``total``) and samples per-node timings into
the profiler. The profiler also drives the editor's profiler panel.
"""
self._profiler = profiler
[docs]
@property
def profiler(self) -> FrameProfiler | None:
return self._profiler
[docs]
def update(self, dt: float) -> None:
"""Per-frame update, called by the editor's process loop.
When playing and not paused, processes the game tree under its own
per-tree input (``game_tree.activate_input()``), so game scripts using
``Input.is_action_pressed()`` transparently see the game's isolated input
without swapping the global Input singleton.
"""
if not self._state.is_playing:
self._orbit_dragging = False
self._game_camera = None
return
# Check for scene transitions
self._check_scene_transition()
# Always tick metrics (even when paused, for display purposes)
self._elapsed_time += dt
self._frame_count += 1
# Lazily locate the game camera
if self._game_camera is None and self._game_tree is not None:
root = self._game_tree.root
if root is not None:
self._game_camera = self._find_game_camera(root)
# Poll hot-reload manager so script edits apply live (also during pause).
# Justified: reload errors during play must NOT crash the editor; scripts are
# frequently mid-edit and import/syntax failures are a normal recoverable state.
if self._hot_reload_mgr is not None:
try:
self._hot_reload_mgr.poll(dt)
except Exception:
_hot_reload_log.exception("hot_reload: poll failed")
if self._state.is_paused:
return
if self._game_tree is None or self._game_tree.root is None:
return
gi = self._game_tree._own_input
prof = self._profiler
# Select timing-aware or plain tree walkers based on profiler presence.
# These error-tolerant / profiled walks are editor policy (per-node
# script-error recovery, per-node timing) and are passed to the shared
# core sequence as the physics/process node_walk so the editor owns HOW
# the tree is walked while core owns the physics->process ORDER.
if prof is not None and prof.enabled:
prof.begin("total")
process_tree = self._process_tree_profiled
physics_tree = self._physics_process_tree_profiled
else:
process_tree = PlayMode._process_tree # type: ignore[assignment]
physics_tree = PlayMode._physics_process_tree # type: ignore[assignment]
def node_walk(node: Node | None, walk_dt: float, phase: str) -> None:
if node is None:
return
if phase == "physics":
physics_tree(node, walk_dt)
else:
process_tree(node, walk_dt)
on_phase: Callable[[str, str], None] | None = None
if prof is not None:
phase_prof = prof
def on_phase(name: str, edge: str) -> None:
(phase_prof.begin if edge == "begin" else phase_prof.end)(name)
phases = ScenePhases(node_walk=node_walk, on_phase=on_phase)
# Swap in game input, dispatch UI events, process, swap back. The editor's
# load-bearing order is UI -> physics -> process (queued viewport UI events
# must reach widgets BEFORE that frame's physics/logic run), so the UI drain
# stays here, before step_scene_logic, rather than in its ui_events slot
# (which the windowed loop uses for its between-physics-and-logic sync).
if gi is not None:
# Isolation is the game tree's OWN per-tree Input (activate_input),
# not a global-singleton swap: forwarded viewport events were fed into
# gi directly, and activate_input makes gi the active Input the game's
# scripts + ui_input see for the duration. The frame boundary
# (_new_frame before, _end_frame after) is preserved exactly.
gi._new_frame()
with self._game_tree.activate_input():
try:
# Dispatch queued UI events to the game tree's UIInputManager
if self._ui_event_queue:
if prof is not None:
prof.begin("ui")
for evt in self._ui_event_queue:
self._game_tree.ui_input(**evt)
self._ui_event_queue.clear()
if prof is not None:
prof.end("ui")
step_scene_logic(self._game_tree, dt, phases=phases)
finally:
gi._end_frame()
# Flush any nodes queued for deletion in the game tree
self._game_tree._flush_deletes()
else:
# Fallback: process without isolated input
step_scene_logic(self._game_tree, dt, phases=phases)
self._game_tree._flush_deletes()
if prof is not None and prof.enabled:
prof.count_nodes(self._game_tree)
prof.end("total")
prof.end_frame()
# ------------------------------------------------------------------
# Input forwarding
# ------------------------------------------------------------------
# ------------------------------------------------------------------
# Game tree access
# ------------------------------------------------------------------
[docs]
@property
def game_tree(self) -> SceneTree | None:
"""The game's isolated SceneTree, or None when not playing."""
return self._game_tree
# ------------------------------------------------------------------
# Game viewport (GPU render-to-texture)
# ------------------------------------------------------------------
[docs]
@property
def game_texture_id(self) -> int | None:
"""Bindless texture ID of the game viewport, or None if not available.
Returns a valid texture ID when the game viewport renderer is active
and ready. The viewport panel checks this to decide whether to display
a GPU-rendered game view or fall back to wireframe.
"""
gvp = self._game_viewport
if gvp is not None and gvp.ready:
return gvp.texture_id
return None
[docs]
@property
def game_viewport(self):
"""The GameViewportRenderer instance, or None."""
return self._game_viewport
[docs]
def create_game_viewport(self, engine, width: int, height: int) -> None:
"""Create the offscreen game viewport renderer.
Called by Root when play mode starts and a graphics engine
is available.
"""
try:
from simvx.graphics.renderer.game_viewport import GameViewportRenderer
self._game_viewport = GameViewportRenderer(engine)
self._game_viewport.create(width, height)
self._game_viewport_enabled = True
log.info("Game viewport renderer created (%dx%d)", width, height)
except Exception:
log.warning("Could not create game viewport renderer (no GPU?)", exc_info=True)
self._game_viewport = None
self._game_viewport_enabled = False
[docs]
def resize_game_viewport(self, width: int, height: int) -> None:
"""Resize the offscreen game viewport immediately (no debounce)."""
if self._game_viewport is not None:
self._game_viewport.resize(width, height)
self._game_viewport_size = (width, height)
self._game_viewport_pending_size = (width, height)
self._game_viewport_stable_frames = 0
game_tree = self.game_tree
if game_tree is not None:
game_tree.screen_size = (width, height)
[docs]
def ensure_game_viewport_size(
self,
width: int,
height: int,
*,
debounce_frames: int = 6,
) -> None:
"""Debounced live resize hook called from the viewport panel each frame.
During a splitter drag the panel size changes many frames in a row.
Only resize once the target size has been stable for
``debounce_frames`` consecutive frames so we don't churn GPU memory
every frame. The first call after viewport creation seeds the
baseline without triggering a resize.
"""
if self._game_viewport is None:
return
target = (max(int(width), 64), max(int(height), 64))
# Seed baseline so the freshly created viewport is treated as in-sync.
if self._game_viewport_size == (0, 0):
self._game_viewport_size = target
self._game_viewport_pending_size = target
return
if target != self._game_viewport_pending_size:
self._game_viewport_pending_size = target
self._game_viewport_stable_frames = 0
else:
self._game_viewport_stable_frames += 1
if target != self._game_viewport_size and self._game_viewport_stable_frames >= debounce_frames:
self.resize_game_viewport(*target)
[docs]
def destroy_game_viewport(self) -> None:
"""Destroy the offscreen game viewport renderer."""
if self._game_viewport is not None:
self._game_viewport.destroy()
self._game_viewport = None
self._game_viewport_enabled = False
self._game_viewport_size = (0, 0)
self._game_viewport_pending_size = (0, 0)
self._game_viewport_stable_frames = 0
# ------------------------------------------------------------------
# Camera management
# ------------------------------------------------------------------
[docs]
def get_active_camera(self) -> Camera3D | None:
"""Return the camera that should drive the viewport.
During play mode the game's own Camera3D is used. Outside of
play mode the editor's orbit camera is returned.
"""
if self._state.is_playing and self._game_camera is not None:
return self._game_camera
return self._state.editor_camera
def _find_game_camera(self, root: Node) -> Camera3D | None:
"""Traverse the scene tree and return the first Camera3D node.
Skips the editor's own camera instance (stored in State).
"""
editor_cam = self._state.editor_camera
for node in root.find_all(Camera3D):
if node is not editor_cam:
return node
return None
def _handle_orbit_input(self) -> None:
"""Handle mouse drag to orbit the game camera during play mode.
Works with OrbitCamera3D nodes: left-drag orbits, scroll zooms.
Pitch is clamped to the upper half-sphere so the camera can't go below the floor.
"""
cam = self._game_camera
if cam is None or not isinstance(cam, OrbitCamera3D):
return
mx, my = Input.mouse_position
lmb = Input.is_mouse_button_pressed(MouseButton.LEFT)
if lmb:
if self._orbit_dragging:
dx = mx - self._last_mouse[0]
dy = my - self._last_mouse[1]
if dx != 0 or dy != 0:
cam.yaw -= dx * _ORBIT_SENSITIVITY
cam.pitch = max(_PITCH_MIN, min(_PITCH_MAX, cam.pitch - dy * _ORBIT_SENSITIVITY))
cam.update_transform()
self._orbit_dragging = True
else:
self._orbit_dragging = False
self._last_mouse = (mx, my)
# Scroll to zoom
_sx, sy = Input.scroll_delta
if sy != 0:
cam.zoom(sy * _ZOOM_SENSITIVITY)
# ------------------------------------------------------------------
# Visual indicator
# ------------------------------------------------------------------
[docs]
def get_border_colour(self) -> tuple[float, float, float, float] | None:
"""Return a viewport border colour indicating the current play state.
* Green ``(0.2, 0.8, 0.2, 1.0)`` -- game is running.
* Orange ``(1.0, 0.6, 0.0, 1.0)`` -- game is paused.
* ``None`` -- editor is in normal (stopped) mode.
"""
if not self._state.is_playing:
return _COLOUR_STOPPED
if self._state.is_paused:
return _COLOUR_PAUSED
return _COLOUR_PLAYING
# ------------------------------------------------------------------
# Runtime metrics (read-only)
# ------------------------------------------------------------------
[docs]
@property
def elapsed_time(self) -> float:
"""Seconds elapsed since play mode started."""
return self._elapsed_time
[docs]
@property
def frame_count(self) -> int:
"""Number of frames processed since play mode started."""
return self._frame_count
[docs]
@property
def is_active(self) -> bool:
"""Convenience: True when the game is playing (paused or not)."""
return self._state.is_playing
# ------------------------------------------------------------------
# Internal helpers
# ------------------------------------------------------------------
def _get_root(self) -> Node | None:
"""Return the editor scene root, or None."""
tree = self._state.edited_scene
if tree is None:
return None
return tree.root
def _discover_watchable_scripts(self) -> list[str]:
"""Return absolute paths of project ``.py`` files for hot-reload to watch.
Walks ``state.project_path`` (when set) and collects every ``.py`` file
outside the standard skip dirs (``.venv``, ``__pycache__``, etc.).
Returns an empty list when no project path is configured.
"""
project_path = self._state.project_path
if project_path is None:
return []
root = Path(project_path)
if not root.is_dir():
return []
scripts: list[str] = []
for dirpath, dirnames, filenames in os.walk(root):
# Prune skip dirs in-place so os.walk doesn't descend into them.
dirnames[:] = [d for d in dirnames if d not in _HOT_RELOAD_SKIP_DIRS]
for fname in filenames:
if fname.endswith(".py"):
scripts.append(str(Path(dirpath) / fname))
return scripts
@staticmethod
def _call_ready(node: Node) -> None:
"""Recursively call ``on_ready()`` on a node and all its descendants.
Children are readied before their parent, matching the SceneTree
convention (depth-first, bottom-up). Errors are logged, not raised.
"""
for child in list(node.children):
PlayMode._call_ready(child)
try:
node.on_ready()
except Exception:
tb = _tb.format_exc()
log.exception("Script error in %s.ready -- node disabled", node.name)
node._script_error = True
try:
Node.script_error_raised.emit(node, "ready", tb)
except Exception:
# justified: signal-handler error during ready; primary script error already logged
pass
@staticmethod
def _process_tree(node: Node, dt: float) -> None:
"""Recursively call ``on_update(dt)`` and tick coroutines."""
if getattr(node, "_script_error", False):
return
try:
node.on_update(dt)
except AssertionError:
raise
except Exception:
node._script_error = True
tb = _tb.format_exc()
log.exception("Script error in %s.process -- node disabled", node.name)
try:
Node.script_error_raised.emit(node, "update", tb)
except Exception:
# justified: signal-handler error during process; primary script error already logged
pass
return
node._tick_coroutines(dt)
for child in list(node.children):
PlayMode._process_tree(child, dt)
@staticmethod
def _physics_process_tree(node: Node, dt: float) -> None:
"""Recursively call ``on_fixed_update(dt)``."""
if getattr(node, "_script_error", False):
return
try:
node.on_fixed_update(dt)
except AssertionError:
raise
except Exception:
node._script_error = True
tb = _tb.format_exc()
log.exception("Script error in %s.physics_process -- node disabled", node.name)
try:
Node.script_error_raised.emit(node, "fixed_update", tb)
except Exception:
# justified: signal-handler error during physics_process; primary script error already logged
pass
return
for child in list(node.children):
PlayMode._physics_process_tree(child, dt)
# ------------------------------------------------------------------
# Profiling variants -- per-node timed tree walks.
# ------------------------------------------------------------------
def _process_tree_profiled(self, node: Node, dt: float) -> None:
"""Timed :meth:`_process_tree` that feeds per-node samples to the profiler."""
if getattr(node, "_script_error", False):
return
prof = self._profiler
t0 = time.perf_counter()
try:
node.on_update(dt)
except AssertionError:
raise
except Exception:
node._script_error = True
tb = _tb.format_exc()
log.exception("Script error in %s.process -- node disabled", node.name)
try:
Node.script_error_raised.emit(node, "update", tb)
except Exception:
# justified: signal-handler error during profiled process; primary script error already logged
pass
return
finally:
if prof is not None:
prof.sample_node(node.path, "process", (time.perf_counter() - t0) * 1000.0)
node._tick_coroutines(dt)
for child in list(node.children):
self._process_tree_profiled(child, dt)
def _physics_process_tree_profiled(self, node: Node, dt: float) -> None:
"""Timed :meth:`_physics_process_tree` with per-node samples."""
if getattr(node, "_script_error", False):
return
prof = self._profiler
t0 = time.perf_counter()
try:
node.on_fixed_update(dt)
except AssertionError:
raise
except Exception:
node._script_error = True
tb = _tb.format_exc()
log.exception("Script error in %s.physics_process -- node disabled", node.name)
try:
Node.script_error_raised.emit(node, "fixed_update", tb)
except Exception:
# justified: signal-handler error during profiled physics_process; primary script error already logged
pass
return
finally:
if prof is not None:
prof.sample_node(node.path, "physics", (time.perf_counter() - t0) * 1000.0)
for child in list(node.children):
self._physics_process_tree_profiled(child, dt)