EG/ssbo_component/ssbo_editor.py

3966 lines
160 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import sys
import os
import struct
import time
import types
from panda3d.core import (
Filename, loadPrcFileData, GeomVertexFormat,
GeomVertexWriter, InternalName, Shader, Texture, SamplerState,
Vec3, Vec4, Point2, Point3, LMatrix4f, ShaderBuffer, GeomEnums, OmniBoundingVolume,
TransparencyAttrib, BoundingSphere, NodePath,
GraphicsEngine, WindowProperties, FrameBufferProperties,
GraphicsPipe, GraphicsOutput, Camera, DisplayRegion, OrthographicLens,
BoundingBox, BitMask32, Material, MaterialAttrib, ColorAttrib, TextureAttrib, PNMImage,
KeyboardButton
)
# p3dimgui.backend first tries `from shaders import *`, which can be shadowed by
# project folders named `shaders/` and leave VERT_SHADER/FRAG_SHADER undefined.
# Seed a valid fallback module before importing p3dimgui.
_shaders_mod = sys.modules.get("shaders")
if not (_shaders_mod and hasattr(_shaders_mod, "VERT_SHADER") and hasattr(_shaders_mod, "FRAG_SHADER")):
_shaders_mod = types.ModuleType("shaders")
_shaders_mod.FRAG_SHADER = """
#version 120
varying vec2 texcoord;
varying vec4 color;
uniform sampler2D p3d_Texture0;
void main() {
gl_FragColor = color * texture2D(p3d_Texture0, texcoord);
}
"""
_shaders_mod.VERT_SHADER = """
#version 120
attribute vec4 p3d_Vertex;
attribute vec4 p3d_Color;
varying vec2 texcoord;
varying vec4 color;
uniform mat4 p3d_ModelViewProjectionMatrix;
void main() {
texcoord = p3d_Vertex.zw;
color = p3d_Color.bgra;
gl_Position = p3d_ModelViewProjectionMatrix * vec4(p3d_Vertex.x, 0.0, -p3d_Vertex.y, 1.0);
}
"""
sys.modules["shaders"] = _shaders_mod
from p3dimgui import ImGuiBackend
from imgui_bundle import imgui
from rpcore.effect import Effect
from .ssbo_controller import ObjectController
from core.selection_outline import SelectionOutlineManager
class SSBOEditor:
"""
SSBO Editor Component
====================
Encapsulates the SSBO rendering, ImGui editor, and interaction logic.
Can be integrated into any ShowBase application using RenderPipeline.
"""
def __init__(self, base_app, render_pipeline, model_path=None, font_path=None):
self.base = base_app
self.rp = render_pipeline
self.controller = None
self.model = None
self.source_model = None
self.source_model_root = None
self._source_child_base_mats = {}
self._runtime_owner_base_mats = {}
self._model_root_last_snapshot_mat = LMatrix4f.ident_mat()
self.last_import_tree_key = None
self.last_import_root_name = None
self.ssbo = None
self.font_path = font_path
self._transform_gizmo = None
self._ssbo_transform_active = False
self._ssbo_selected_local_indices = []
self._ssbo_gizmo_proxy = None
# Internal State
self.selected_name = None
self.selected_ids = []
self.transform_ids = []
self.search_text = ""
self.last_search_text = None
self.filtered_nodes = []
self.debug_mode = False
self.keys = {}
self.pick_mask = BitMask32.bit(29)
self.pick_buffer = None
self._empty_pick_scene = NodePath("ssbo_pick_empty")
# Avoid heavy per-frame sync for huge group selections.
self._pick_sync_bg_limit = 256
self._last_group_sync_mat = None
self._last_single_sync_gid = None
self._last_single_sync_mat = None
self._group_proxy_initial_mat = None
self._group_proxy_source_initial_net_mat = None
# Performance toggle: forcing shadow tasks every frame is expensive.
# Keep it off by default so frustum/content reduction has clearer FPS impact.
self.realtime_shadow_updates = False
self._scheduler_tasks_original = None
self._realtime_shadow_tasks_enabled = False
self._outline_manager = getattr(self.base, "_selection_outline_manager", None)
if self._outline_manager is None:
self._outline_manager = SelectionOutlineManager(self.base)
setattr(self.base, "_selection_outline_manager", self._outline_manager)
# Initialize ImGui Backend if not already present
if not hasattr(self.base, 'imgui_backend'):
print("[SSBOEditor] Initializing ImGui Backend...")
self.base.imgui_backend = ImGuiBackend()
self.load_font()
# Register Events
self.base.accept("imgui-new-frame", self.draw_imgui)
self.base.accept("f", self.focus_on_selected)
self.base.accept("d", self.toggle_debug)
self.base.accept("mouse1", self.on_mouse_click)
# Register Input Tasks
for key in ['arrow_up', 'arrow_down', 'arrow_left', 'arrow_right', 'z', 'x']:
self.base.accept(key, self.keys.__setitem__, [key, True])
self.base.accept(f"{key}-up", self.keys.__setitem__, [key, False])
# Add Tasks
self.base.taskMgr.add(self.update_task, "update_task")
# Load Model if provided
if model_path:
self.load_model(model_path)
def _get_model_file_size(self, model_path):
try:
return int(os.path.getsize(model_path))
except Exception:
return 0
def _should_skip_heavy_material_processing(self, *, model_path=None, geom_count=0, node_count=0):
file_size = self._get_model_file_size(model_path) if model_path else 0
if file_size >= 200 * 1024 * 1024:
return True
if geom_count >= 2048:
return True
if node_count >= 4096:
return True
return False
def _capture_scheduler_tasks_snapshot(self):
scheduler = getattr(self.rp, "task_scheduler", None)
if not scheduler or not hasattr(scheduler, "_tasks"):
return
if self._scheduler_tasks_original is None:
self._scheduler_tasks_original = [list(frame_tasks) for frame_tasks in scheduler._tasks]
def _enable_realtime_shadow_tasks(self):
"""
Force PSSM-related scheduled tasks to run every frame to avoid visible
shadow lag/ghosting while editing moving objects.
"""
scheduler = getattr(self.rp, "task_scheduler", None)
if not scheduler or not hasattr(scheduler, "_tasks"):
return
self._capture_scheduler_tasks_snapshot()
required = {
"pssm_scene_shadows",
"pssm_distant_shadows",
"pssm_convert_distant_to_esm",
"pssm_blur_distant_vert",
"pssm_blur_distant_horiz",
}
changed = False
for frame_tasks in scheduler._tasks:
for task_name in required:
if task_name not in frame_tasks:
frame_tasks.append(task_name)
changed = True
if changed:
print("[SSBOEditor] Realtime shadow tasks enabled (PSSM updates every frame).")
self._realtime_shadow_tasks_enabled = True
def _disable_realtime_shadow_tasks(self):
"""Restore scheduler layout captured before realtime shadow override."""
scheduler = getattr(self.rp, "task_scheduler", None)
if not scheduler or not hasattr(scheduler, "_tasks"):
return
if self._scheduler_tasks_original is None:
self._realtime_shadow_tasks_enabled = False
return
scheduler._tasks[:] = [list(frame_tasks) for frame_tasks in self._scheduler_tasks_original]
self._realtime_shadow_tasks_enabled = False
def set_realtime_shadow_updates(self, enabled):
"""Public toggle for aggressive per-frame shadow updates."""
self.realtime_shadow_updates = bool(enabled)
if self.realtime_shadow_updates:
self._enable_realtime_shadow_tasks()
else:
self._disable_realtime_shadow_tasks()
def load_font(self):
"""Load custom font for ImGui"""
io = imgui.get_io()
# Load Chinese Glyph Ranges
glyph_ranges = None
try:
if hasattr(io.fonts, 'get_glyph_ranges_chinese_full'):
glyph_ranges = io.fonts.get_glyph_ranges_chinese_full()
elif hasattr(io.fonts, 'get_glyph_ranges_chinese_simplified_common'):
glyph_ranges = io.fonts.get_glyph_ranges_chinese_simplified_common()
except Exception as e:
print(f"[SSBOEditor] Warning: Could not get Chinese glyph ranges: {e}")
try:
if self.font_path and os.path.exists(self.font_path):
io.fonts.clear()
# If glyph_ranges is None, it uses default (Basic Latin)
if glyph_ranges:
io.fonts.add_font_from_file_ttf(self.font_path, 18.0, glyph_ranges=glyph_ranges)
else:
io.fonts.add_font_from_file_ttf(self.font_path, 18.0)
else:
# Fallback to default or common font
default_font = os.path.join(os.path.dirname(os.path.dirname(__file__)), "font", "msyh.ttc")
if os.path.exists(default_font):
io.fonts.clear()
io.fonts.add_font_from_file_ttf(default_font, 18.0, glyph_ranges=glyph_ranges)
else:
io.fonts.clear()
io.fonts.add_font_default()
except Exception as e:
print(f"[SSBOEditor] Font load error: {e}")
io.fonts.clear()
io.fonts.add_font_default()
def _fixBlackMaterials(self, model):
try:
from panda3d.core import MaterialAttrib, Material, GeomNode
for geom_path in model.findAllMatches('**/+GeomNode'):
geom_node = geom_path.node()
node_state = geom_path.getState()
if node_state.hasAttrib(MaterialAttrib.getClassType()):
mat_attrib = node_state.getAttrib(MaterialAttrib.getClassType())
mat = mat_attrib.getMaterial()
if mat:
is_black = False
if mat.hasBaseColor():
c = mat.getBaseColor()
if c.x <= 0.05 and c.y <= 0.05 and c.z <= 0.05:
is_black = True
elif mat.hasDiffuse():
c = mat.getDiffuse()
if c.x <= 0.05 and c.y <= 0.05 and c.z <= 0.05:
is_black = True
if is_black or not (mat.hasBaseColor() or mat.hasDiffuse()):
new_mat = Material(mat)
new_mat.setBaseColor((0.8, 0.8, 0.8, 1.0))
new_mat.setDiffuse((0.8, 0.8, 0.8, 1.0))
geom_path.setState(node_state.setAttrib(MaterialAttrib.make(new_mat)))
for i in range(geom_node.getNumGeoms()):
geom_state = geom_node.getGeomState(i)
if geom_state.hasAttrib(MaterialAttrib.getClassType()):
mat_attrib = geom_state.getAttrib(MaterialAttrib.getClassType())
mat = mat_attrib.getMaterial()
if mat:
is_black = False
if mat.hasBaseColor():
c = mat.getBaseColor()
if c.x <= 0.05 and c.y <= 0.05 and c.z <= 0.05:
is_black = True
elif mat.hasDiffuse():
c = mat.getDiffuse()
if c.x <= 0.05 and c.y <= 0.05 and c.z <= 0.05:
is_black = True
if is_black or not (mat.hasBaseColor() or mat.hasDiffuse()):
new_mat = Material(mat)
new_mat.setBaseColor((0.8, 0.8, 0.8, 1.0))
new_mat.setDiffuse((0.8, 0.8, 0.8, 1.0))
geom_node.setGeomState(i, geom_state.setAttrib(MaterialAttrib.make(new_mat)))
else:
new_mat = Material()
new_mat.setBaseColor((0.8, 0.8, 0.8, 1.0))
new_mat.setDiffuse((0.8, 0.8, 0.8, 1.0))
new_mat.setSpecular((0.2, 0.2, 0.2, 1.0))
new_mat.setRoughness(0.8)
geom_node.setGeomState(i, geom_state.addAttrib(MaterialAttrib.make(new_mat)))
model.clearColor()
except Exception as e:
print(f'修复黑色模型材质时出错: {e}')
def _load_source_model_from_path(self, model_path, apply_black_fix=True, repair_textures=True):
"""Load a source model NodePath from disk without touching current runtime state."""
source_model = None
last_error = None
load_path = model_path
try:
from scene.gltf_support import ensure_gltf_visual_bam, probe_gltf_metadata
gltf_meta = probe_gltf_metadata(model_path)
if gltf_meta.get("is_gltf"):
has_anim = gltf_meta.get("has_animations", False)
if has_anim:
project_manager = getattr(self.base, "project_manager", None)
project_root = getattr(project_manager, "current_project_path", "") if project_manager else ""
cached_visual_path = ensure_gltf_visual_bam(
model_path,
project_root=project_root,
skip_animations=False, # 既然是为了动画,就不跳过
flatten_nodes=False,
)
if cached_visual_path and cached_visual_path != model_path:
load_path = cached_visual_path
print(f"[GLTF智能加载] SSBO检测到动画使用缓存: {cached_visual_path}")
else:
print(f"[GLTF智能加载] SSBO识别为静态模型跳过缓存以获得最大流畅度")
except Exception as e:
print(f"[GLTF可见缓存] SSBO回退原始模型加载: {e}")
loader_options = None
try:
from panda3d.core import LoaderOptions
loader_options = LoaderOptions()
if hasattr(loader_options, "LF_no_cache"):
loader_options.setFlags(loader_options.getFlags() | loader_options.LF_no_cache)
elif hasattr(loader_options, "LFNoCache"):
loader_options.setFlags(loader_options.getFlags() | loader_options.LFNoCache)
except Exception:
loader_options = None
for fn in self._build_filename_candidates(load_path):
try:
if loader_options is not None:
source_model = self.base.loader.loadModel(fn, loader_options)
else:
source_model = self.base.loader.loadModel(fn)
if source_model and not source_model.is_empty():
break
except Exception as e:
last_error = e
source_model = None
if not source_model or source_model.is_empty():
if last_error:
raise RuntimeError(f"Failed to load model '{model_path}': {last_error}")
raise RuntimeError(f"Failed to load model '{model_path}'")
if apply_black_fix:
self._fixBlackMaterials(source_model)
if repair_textures:
self._repair_missing_textures(source_model, model_path)
return source_model
def _detach_shared_materials_in_subtree(self, root_np):
"""Clone materials in-place so one imported model instance cannot affect another."""
try:
from panda3d.core import Material, MaterialAttrib
if not self._node_is_valid(root_np):
return
for np in root_np.find_all_matches("**"):
if not self._node_is_valid(np):
continue
try:
if np.hasMaterial():
material = np.getMaterial()
if material is not None:
np.setMaterial(Material(material), 1)
except Exception:
pass
try:
panda_node = np.node()
if panda_node is None or not hasattr(panda_node, "getNumGeoms"):
continue
for geom_index in range(panda_node.getNumGeoms()):
geom_state = panda_node.getGeomState(geom_index)
if not geom_state.hasAttrib(MaterialAttrib):
continue
material_attrib = geom_state.getAttrib(MaterialAttrib)
material = material_attrib.getMaterial() if material_attrib else None
if material is None:
continue
panda_node.setGeomState(
geom_index,
geom_state.setAttrib(MaterialAttrib.make(Material(material))),
)
except Exception:
continue
except Exception:
pass
def _materialize_explicit_geom_materials(self, root_np):
"""Bake each GeomNode's effective material into local GeomState and clear inherited node material."""
try:
from panda3d.core import Material, MaterialAttrib
if not self._node_is_valid(root_np):
return
for geom_np in root_np.find_all_matches("**/+GeomNode"):
if not self._node_is_valid(geom_np):
continue
try:
geom_node = geom_np.node()
except Exception:
continue
net_state = None
try:
net_state = geom_np.getNetState()
except Exception:
try:
net_state = geom_np.get_net_state()
except Exception:
net_state = None
for geom_index in range(geom_node.getNumGeoms()):
try:
geom_state = geom_node.getGeomState(geom_index)
except Exception:
continue
material = None
try:
if geom_state.hasAttrib(MaterialAttrib):
material_attrib = geom_state.getAttrib(MaterialAttrib)
material = material_attrib.getMaterial() if material_attrib else None
except Exception:
material = None
if material is None and net_state is not None:
try:
if net_state.hasAttrib(MaterialAttrib):
material_attrib = net_state.getAttrib(MaterialAttrib)
material = material_attrib.getMaterial() if material_attrib else None
except Exception:
material = None
if material is None:
try:
if geom_np.hasMaterial():
material = geom_np.getMaterial()
except Exception:
material = None
if material is None:
continue
try:
geom_node.setGeomState(
geom_index,
geom_state.setAttrib(MaterialAttrib.make(Material(material))),
)
except Exception:
continue
try:
if geom_np.hasMaterial():
geom_np.clearMaterial()
except Exception:
pass
try:
geom_np.clearAttrib(MaterialAttrib.getClassType())
except Exception:
pass
except Exception:
pass
def _set_node_name(self, node, name):
if not node:
return
try:
node.set_name(name)
return
except Exception:
pass
try:
node.setName(name)
except Exception:
pass
def _get_node_name(self, node, default_name=None):
if not node:
return default_name
try:
return node.get_name()
except Exception:
pass
try:
return node.getName()
except Exception:
return default_name
def _iter_children(self, node):
if not node:
return []
try:
return list(node.get_children())
except Exception:
try:
return list(node.getChildren())
except Exception:
return []
def _ensure_source_model_root(self):
root = self.source_model_root
if root:
try:
if not root.is_empty():
return root
except Exception:
try:
if not root.isEmpty():
return root
except Exception:
pass
self.source_model_root = NodePath("ssbo_source_scene_root")
return self.source_model_root
def _make_unique_source_child_name(self, desired_name):
root = self._ensure_source_model_root()
existing_names = set()
for child in self._iter_children(root):
try:
existing_names.add(child.get_name())
except Exception:
try:
existing_names.add(child.getName())
except Exception:
continue
base_name = desired_name or "imported_model"
if base_name not in existing_names:
return base_name
stem, ext = os.path.splitext(base_name)
index = 2
while True:
candidate = f"{stem}_{index}{ext}"
if candidate not in existing_names:
return candidate
index += 1
@staticmethod
def _tag_is_enabled(node, tag_name, default=False):
try:
if not node or not node.hasTag(tag_name):
return bool(default)
value = str(node.getTag(tag_name) or "").strip().lower()
if value in ("1", "true", "yes", "on"):
return True
if value in ("0", "false", "no", "off"):
return False
except Exception:
pass
return bool(default)
def _node_has_saved_animation_info(self, node):
if not self._node_is_valid(node):
return False
for tag_name in ("saved_has_animations", "has_animations"):
if self._tag_is_enabled(node, tag_name, default=False):
return True
try:
if node.hasTag("gltf_animation_count"):
return int(str(node.getTag("gltf_animation_count") or "0").strip() or "0") > 0
except Exception:
pass
return False
def _should_skip_scene_package_child_for_ssbo(self, node):
if not self._node_is_valid(node):
return False
if self._node_has_saved_animation_info(node):
return True
try:
if node.hasTag("ssbo_managed"):
return not self._tag_is_enabled(node, "ssbo_managed", default=True)
except Exception:
pass
return False
def _capture_source_child_base_mats(self):
"""Capture baseline local mats for each top-level source child."""
self._source_child_base_mats = {}
self._model_root_last_snapshot_mat = LMatrix4f.ident_mat()
root = self.source_model_root
if not root:
return
for child in self._iter_children(root):
if not self._node_is_valid(child):
continue
name = self._get_node_name(child, None)
if not name:
continue
try:
self._source_child_base_mats[name] = LMatrix4f(child.get_mat())
except Exception:
try:
self._source_child_base_mats[name] = LMatrix4f(child.getMat())
except Exception:
continue
def _get_top_level_group_keys(self):
if not self.controller or not getattr(self.controller, "tree_root_key", None):
return []
root_key = self.controller.tree_root_key
root_node = self.controller.tree_nodes.get(root_key)
if not root_node:
return []
return list(root_node.get("children", []))
def _invert_matrix_in_place(self, mat_value):
inv_mat = LMatrix4f(mat_value)
try:
inv_mat.invertInPlace()
return inv_mat
except Exception:
try:
inv_mat.invert_in_place()
return inv_mat
except Exception:
return None
def _capture_runtime_owner_base_mats(self):
"""Capture runtime model-space matrices keyed by hierarchy owner key."""
self._runtime_owner_base_mats = {}
controller = getattr(self, "controller", None)
if not controller or not self.model:
return
id_to_object_np = getattr(controller, "id_to_object_np", {}) or {}
id_to_name = getattr(controller, "id_to_name", {}) or {}
if not id_to_object_np or not id_to_name:
return
root_key = getattr(controller, "tree_root_key", None)
for global_id, obj_np in id_to_object_np.items():
if not self._node_is_valid(obj_np):
continue
owner_key = str(id_to_name.get(global_id, "") or "").strip()
if not owner_key or owner_key == root_key:
continue
if owner_key in self._runtime_owner_base_mats:
continue
try:
self._runtime_owner_base_mats[owner_key] = LMatrix4f(obj_np.get_mat(self.model))
except Exception:
try:
self._runtime_owner_base_mats[owner_key] = LMatrix4f(obj_np.getMat(self.model))
except Exception:
continue
def _snapshot_top_level_transforms_to_source_root(self):
"""Persist current runtime transforms back into the source scene tree."""
if not self.controller or not self.model or not self.source_model_root:
return
selected_key = self.selected_name
is_top_level_like_selection = False
if selected_key:
is_top_level_like_selection = bool(
self._selection_equivalent_top_level_key(selected_key)
or self._is_top_level_source_child_selection(selected_key)
or self._is_displayed_top_level_selection(selected_key)
or self._should_use_model_root_for_top_level_selection(selected_key)
)
selected_top_synced = self._snapshot_selected_top_level_transform_to_source()
# Top-level model transforms live on the aggregated runtime root/proxy.
# Batch-syncing descendant runtime objects in the same pass writes an
# opposite transform into the children, so the reopened scene visually
# cancels back to the original pose.
if not is_top_level_like_selection:
self._snapshot_runtime_transforms_to_source_root()
if not selected_top_synced:
self._snapshot_active_selection_transform_to_source()
self._snapshot_model_root_transform_to_source_root()
def _snapshot_selected_top_level_transform_to_source(self):
"""Explicitly persist currently selected top-level model transform, including wrapper selections."""
if not self.controller or not self.model or not self.source_model_root:
return False
selected_key = self.selected_name
root_key = getattr(self.controller, "tree_root_key", None)
if not selected_key or selected_key == root_key:
return False
top_key = self._selection_equivalent_top_level_key(selected_key)
if not top_key:
is_top_level_like_selection = bool(
self._is_top_level_source_child_selection(selected_key)
or self._is_displayed_top_level_selection(selected_key)
)
if is_top_level_like_selection:
top_key = selected_key
if not top_key:
return False
source_node = self._resolve_source_node_by_tree_key(top_key)
if not self._node_is_valid(source_node):
return False
scene_node = self.get_selection_scene_node()
if not self._node_is_valid(scene_node):
return False
# Single-model top-level selections are edited through self.model.
# In that case, use the dedicated model-root delta snapshot path.
if scene_node == self.model:
return self._snapshot_model_root_transform_to_source_root()
try:
current_net_mat = LMatrix4f(scene_node.get_mat(self.model))
except Exception:
try:
current_net_mat = LMatrix4f(scene_node.getMat(self.model))
except Exception:
return False
try:
reference_node = self._get_transform_snapshot_reference_node(source_node)
source_node.set_mat(reference_node, current_net_mat)
source_node.setTag("scene_transform_dirty", "true")
try:
self._cache_top_level_source_child_base_mat(source_node, source_node.get_mat())
except Exception:
try:
self._cache_top_level_source_child_base_mat(source_node, source_node.getMat())
except Exception:
pass
return True
except Exception:
try:
source_node.setMat(reference_node, current_net_mat)
source_node.setTag("scene_transform_dirty", "true")
try:
self._cache_top_level_source_child_base_mat(source_node, source_node.get_mat())
except Exception:
try:
self._cache_top_level_source_child_base_mat(source_node, source_node.getMat())
except Exception:
pass
return True
except Exception:
return False
def _snapshot_runtime_transforms_to_source_root(self):
"""Persist runtime object transforms for all owner keys back into source tree."""
controller = getattr(self, "controller", None)
if not controller or not self.model or not self.source_model_root:
return 0
id_to_object_np = getattr(controller, "id_to_object_np", {}) or {}
id_to_name = getattr(controller, "id_to_name", {}) or {}
if not id_to_object_np or not id_to_name:
return 0
owner_net_mats = {}
for global_id, obj_np in id_to_object_np.items():
if not self._node_is_valid(obj_np):
continue
owner_key = str(id_to_name.get(global_id, "") or "").strip()
if not owner_key:
continue
if owner_key == getattr(controller, "tree_root_key", None):
continue
if owner_key in owner_net_mats:
continue
try:
owner_net_mats[owner_key] = LMatrix4f(obj_np.get_mat(self.model))
except Exception:
try:
owner_net_mats[owner_key] = LMatrix4f(obj_np.getMat(self.model))
except Exception:
continue
if not owner_net_mats:
return 0
synced = 0
ordered_owner_items = sorted(
owner_net_mats.items(),
key=lambda item: str(item[0]).count("/"),
)
for owner_key, current_net_mat in ordered_owner_items:
source_node = self._resolve_source_node_by_tree_key(owner_key)
if not self._node_is_valid(source_node):
continue
try:
# Use robust Panda3D set_mat(ref, mat) to handle all coordinate conversions.
# This ensures the source node's net transform relative to the source root
# exactly matches the runtime node's net transform relative to self.model.
reference_node = self._get_transform_snapshot_reference_node(source_node)
source_node.set_mat(reference_node, current_net_mat)
source_node.setTag("scene_transform_dirty", "true")
# Update top-level child baseline cache for structural sync
try:
local_mat = source_node.get_mat()
except Exception:
local_mat = source_node.getMat()
self._cache_top_level_source_child_base_mat(source_node, local_mat)
# Mark as synced for the next incremental update
self._runtime_owner_base_mats[owner_key] = LMatrix4f(current_net_mat)
synced += 1
except Exception as e:
print(f"[SSBO] Snapshot failed for {owner_key}: {e}")
continue
return synced
def _snapshot_runtime_materials_to_source_root(self):
"""Persist current runtime material tags back into the source scene tree."""
controller = getattr(self, "controller", None)
if not controller or not self.model or not self.source_model_root:
return 0
id_to_object_np = getattr(controller, "id_to_object_np", {}) or {}
id_to_name = getattr(controller, "id_to_name", {}) or {}
if not id_to_object_np or not id_to_name:
return 0
synced = 0
material_tag_prefixes = ("material_", "scene_material_")
# We need to find all unique owner keys and their corresponding runtime nodes
# that actually hold the updated material tags.
owner_to_runtime = {}
for global_id, obj_np in id_to_object_np.items():
if not self._node_is_valid(obj_np):
continue
owner_key = str(id_to_name.get(global_id, "") or "").strip()
if not owner_key or owner_key == getattr(controller, "tree_root_key", None):
continue
# If multiple objects share an owner key, any of them having a dirty material tag
# should probably work, but we take the first one or one that is marked dirty.
if owner_key not in owner_to_runtime:
owner_to_runtime[owner_key] = obj_np
elif obj_np.hasTag("scene_material_dirty") and obj_np.getTag("scene_material_dirty").lower() == "true":
owner_to_runtime[owner_key] = obj_np
for owner_key, runtime_node in owner_to_runtime.items():
source_node = self._resolve_source_node_by_tree_key(owner_key)
if not self._node_is_valid(source_node):
continue
# Check if runtime node has any material tags to sync
tags_to_sync = {}
has_dirty = False
for tag_key in runtime_node.getTagKeys():
if any(tag_key.startswith(p) for p in material_tag_prefixes):
tags_to_sync[tag_key] = runtime_node.getTag(tag_key)
if tag_key == "scene_material_dirty" and tags_to_sync[tag_key].lower() == "true":
has_dirty = True
# Even if not explicitly marked dirty, if it has material tags, we sync them.
if tags_to_sync:
for t_key, t_val in tags_to_sync.items():
source_node.setTag(t_key, t_val)
synced += 1
return synced
def _snapshot_model_root_transform_to_source_root(self):
"""Apply only the aggregated runtime-root delta onto top-level source children."""
if not self.controller or not self.model or not self.source_model_root:
return False
# Persist whole-model transforms that live on the aggregated runtime root.
# Child/object sync above only captures transforms relative to self.model,
# so moving the whole imported model root would otherwise be lost on save.
model_root_mat = None
try:
model_root_mat = LMatrix4f(self.model.get_mat(self.base.render))
except Exception:
try:
model_root_mat = LMatrix4f(self.model.getMat(self.base.render))
except Exception:
model_root_mat = None
if model_root_mat is None:
return False
source_children = self._get_source_root_children()
if not source_children:
return False
# Single imported model: the editor manipulates self.model directly.
# Persist that absolute root transform straight onto the only top-level
# source child. Using delta composition here loses the saved root pose
# after reopen because there is no sibling model to preserve relative to.
if len(source_children) == 1 and self._node_is_valid(source_children[0]):
source_child = source_children[0]
try:
source_child.set_mat(model_root_mat)
source_child.setTag("scene_transform_dirty", "true")
self._cache_top_level_source_child_base_mat(source_child, model_root_mat)
self._model_root_last_snapshot_mat = LMatrix4f(model_root_mat)
return True
except Exception:
try:
source_child.setMat(model_root_mat)
source_child.setTag("scene_transform_dirty", "true")
self._cache_top_level_source_child_base_mat(source_child, model_root_mat)
self._model_root_last_snapshot_mat = LMatrix4f(model_root_mat)
return True
except Exception:
return False
previous_root_mat = getattr(self, "_model_root_last_snapshot_mat", None)
if previous_root_mat is None:
previous_root_mat = LMatrix4f.ident_mat()
model_root_delta = LMatrix4f(previous_root_mat)
try:
model_root_delta.invertInPlace()
except Exception:
try:
model_root_delta.invert_in_place()
except Exception:
return False
model_root_delta *= model_root_mat
applied = False
for source_child in source_children:
if not self._node_is_valid(source_child):
continue
try:
composed_mat = LMatrix4f(source_child.get_mat())
except Exception:
try:
composed_mat = LMatrix4f(source_child.getMat())
except Exception:
continue
composed_mat *= model_root_delta
try:
source_child.set_mat(composed_mat)
source_child.setTag("scene_transform_dirty", "true")
self._cache_top_level_source_child_base_mat(source_child, composed_mat)
applied = True
except Exception:
try:
source_child.setMat(composed_mat)
source_child.setTag("scene_transform_dirty", "true")
self._cache_top_level_source_child_base_mat(source_child, composed_mat)
applied = True
except Exception:
continue
if applied:
self._model_root_last_snapshot_mat = LMatrix4f(model_root_mat)
return applied
def _cache_top_level_source_child_base_mat(self, source_node, local_mat=None):
"""Keep top-level source-child baselines in sync with explicit subtree edits."""
if not self._node_is_valid(source_node) or not self._node_is_valid(self.source_model_root):
return False
try:
source_parent = source_node.get_parent()
except Exception:
try:
source_parent = source_node.getParent()
except Exception:
return False
if source_parent != self.source_model_root:
return False
child_name = self._get_node_name(source_node, None)
if not child_name:
return False
if local_mat is None:
try:
local_mat = LMatrix4f(source_node.get_mat())
except Exception:
try:
local_mat = LMatrix4f(source_node.getMat())
except Exception:
return False
else:
local_mat = LMatrix4f(local_mat)
self._source_child_base_mats[child_name] = local_mat
return True
def _snapshot_active_selection_transform_to_source(self):
"""Persist only the actively edited selection back into the source tree."""
if not self.controller or not self.model or not self.source_model_root:
return False
selected_key = self.selected_name
root_key = getattr(self.controller, "tree_root_key", None)
if not selected_key or selected_key == root_key:
return False
normalized_top_key = self._selection_equivalent_top_level_key(selected_key)
if normalized_top_key:
selected_key = normalized_top_key
scene_node = self.get_selection_scene_node()
source_node = self.get_selection_source_node()
if normalized_top_key:
normalized_source_node = self._resolve_source_node_by_tree_key(normalized_top_key)
if self._node_is_valid(normalized_source_node):
source_node = normalized_source_node
if not self._node_is_valid(scene_node) or not self._node_is_valid(source_node):
return False
# When a single imported model is selected via the aggregated runtime
# root, its whole-model transform lives on self.model rather than on the
# source child itself. Writing self.model's local identity back here
# would poison the top-level baseline and later structural edits
# (delete/attach/rebuild) would reapply a stale root transform.
if (
scene_node == self.model
and self._should_use_model_root_for_top_level_selection(selected_key)
):
return self._snapshot_model_root_transform_to_source_root()
is_top_level_like_selection = bool(
self._is_top_level_source_child_selection(selected_key)
or self._is_displayed_top_level_selection(selected_key)
)
# For regular non-root selections, runtime owner-key snapshot is the most
# reliable and fast path.
if not is_top_level_like_selection:
synced = self._snapshot_runtime_transforms_to_source_root()
if synced > 0:
return True
current_net_mat = None
proxy = getattr(self, "_group_proxy", None)
if (
proxy
and scene_node == proxy
and self._group_proxy_initial_mat is not None
and self._group_proxy_source_initial_net_mat is not None
):
try:
current_proxy_mat = LMatrix4f(proxy.get_mat(self.model))
except Exception:
try:
current_proxy_mat = LMatrix4f(proxy.getMat(self.model))
except Exception:
current_proxy_mat = None
if current_proxy_mat is not None:
proxy_delta_mat = LMatrix4f(self._group_proxy_initial_mat)
try:
proxy_delta_mat.invertInPlace()
except Exception:
try:
proxy_delta_mat.invert_in_place()
except Exception:
proxy_delta_mat = None
if proxy_delta_mat is not None:
proxy_delta_mat *= current_proxy_mat
current_net_mat = LMatrix4f(self._group_proxy_source_initial_net_mat)
current_net_mat *= proxy_delta_mat
# Make repeated snapshots idempotent until the selection is cleared.
self._group_proxy_initial_mat = LMatrix4f(current_proxy_mat)
self._group_proxy_source_initial_net_mat = LMatrix4f(current_net_mat)
if current_net_mat is None:
try:
current_net_mat = LMatrix4f(scene_node.get_mat(self.model))
except Exception:
try:
current_net_mat = LMatrix4f(scene_node.getMat(self.model))
except Exception:
current_net_mat = None
if current_net_mat is None:
return False
try:
# Corrected: use Panda3D native world-to-local projection relative to source root
reference_node = self._get_transform_snapshot_reference_node(source_node)
source_node.set_mat(reference_node, current_net_mat)
source_node.setTag("scene_transform_dirty", "true")
try:
self._cache_top_level_source_child_base_mat(source_node, source_node.get_mat())
except Exception:
try:
self._cache_top_level_source_child_base_mat(source_node, source_node.getMat())
except Exception:
pass
return True
except Exception as e:
print(f"[SSBO] Snapshot failed: {e}")
return False
return True
def _resolve_source_node_by_tree_key(self, tree_key):
"""Resolve controller tree key (e.g. 0/1/2) to source_model_root node."""
if not self.source_model_root or not tree_key:
return None
parts = str(tree_key).split("/")
if not parts or parts[0] != "0":
return None
node = self.source_model_root
for part in parts[1:]:
try:
child_index = int(part)
except Exception:
return None
try:
node = node.get_child(child_index)
except Exception:
try:
node = node.getChild(child_index)
except Exception:
return None
if not self._node_is_valid(node):
return None
return node
def _is_node_in_subtree(self, node, subtree_root):
if not self._node_is_valid(node) or not self._node_is_valid(subtree_root):
return False
current = node
while self._node_is_valid(current):
if current == subtree_root:
return True
try:
current = current.get_parent()
except Exception:
try:
current = current.getParent()
except Exception:
return False
return False
def _get_transform_snapshot_reference_node(self, source_node):
"""
Pick the correct reference node when snapshotting runtime transforms.
In the single-top-level-model path, the runtime moves that only top-level
child transform onto self.model. Descendant runtime matrices are then
relative to the sole source child, not to source_model_root. Saving those
descendants against source_model_root bakes the inverse root transform into
children, so reopening restores children but loses the whole-model pose.
"""
source_root = self.source_model_root
if not self._node_is_valid(source_node) or not self._node_is_valid(source_root):
return source_root
source_children = self._get_source_root_children()
if len(source_children) != 1 or not self._node_is_valid(source_children[0]):
return source_root
sole_source_child = source_children[0]
if source_node == sole_source_child:
return source_root
if self._is_node_in_subtree(source_node, sole_source_child):
return sole_source_child
return source_root
def _resolve_tree_key_for_source_node(self, node):
"""Resolve a source-tree NodePath back to the controller tree key."""
source_root = self._ensure_source_model_root()
if not self._node_is_valid(node) or not self._node_is_valid(source_root):
return None
if node == source_root:
return getattr(self.controller, "tree_root_key", "0") if self.controller else "0"
parts = []
current = node
while self._node_is_valid(current) and current != source_root:
try:
parent = current.get_parent()
except Exception:
try:
parent = current.getParent()
except Exception:
return None
if not self._node_is_valid(parent):
return None
child_index = None
children = self._iter_children(parent)
for index, child in enumerate(children):
if child == current:
child_index = index
break
if child_index is None:
return None
parts.append(str(child_index))
current = parent
if current != source_root:
return None
parts.reverse()
root_key = getattr(self.controller, "tree_root_key", "0") if self.controller else "0"
return "/".join([root_key] + parts) if parts else root_key
def sync_runtime_material_from_source_node(self, source_node):
"""Update only the runtime objects owned by the given source node."""
if not self.controller or not self._node_is_valid(source_node):
return False
owner_key = ""
try:
owner_key = str(source_node.getTag("imported_node_key") or "").strip()
except Exception:
owner_key = ""
if not owner_key:
owner_key = self._resolve_tree_key_for_source_node(source_node) or ""
if not owner_key:
return False
property_helpers = getattr(self.world, "property_helpers", None) if getattr(self, "world", None) else None
if not property_helpers:
return False
get_material_fn = getattr(property_helpers, "_get_renderable_node_material", None)
ensure_material_fn = getattr(property_helpers, "_ensure_material_for_node", None)
clone_material_fn = getattr(property_helpers, "_clone_material_for_node", None)
apply_geom_fn = getattr(property_helpers, "_apply_material_to_geom_states", None)
apply_surface_fn = getattr(property_helpers, "_apply_material_surface_state", None)
refresh_pipeline_fn = getattr(property_helpers, "_refresh_pipeline_material_mode", None)
material = get_material_fn(source_node) if callable(get_material_fn) else None
if material is None and callable(ensure_material_fn):
material = ensure_material_fn(source_node)
if material is None:
return False
target_ids = []
try:
current_source_node = self.get_selection_source_material_node()
except Exception:
current_source_node = None
if self._node_is_valid(current_source_node) and current_source_node == source_node:
target_ids = list(getattr(self, "selected_ids", []) or [])
if not target_ids:
target_ids = [
gid
for gid in getattr(self.controller, "id_to_object_np", {}).keys()
if self.controller.id_to_name.get(gid) == owner_key
]
updated = False
for gid in target_ids:
obj_np = getattr(self.controller, "id_to_object_np", {}).get(gid)
if not self._node_is_valid(obj_np):
continue
runtime_material = material
if callable(clone_material_fn):
try:
runtime_material = clone_material_fn(material, obj_np)
except Exception:
runtime_material = material
if callable(apply_geom_fn):
try:
apply_geom_fn(obj_np, runtime_material)
except Exception:
pass
if callable(apply_surface_fn):
try:
apply_surface_fn(obj_np, runtime_material)
except Exception:
pass
if callable(refresh_pipeline_fn):
try:
refresh_pipeline_fn(obj_np, runtime_material)
except Exception:
pass
updated = True
return updated
def _resolve_tree_key_for_source_node(self, node):
"""Resolve a source-tree NodePath back to its controller tree key."""
if not self._node_is_valid(node) or not self._node_is_valid(self.source_model_root):
return None
if node == self.source_model_root:
return getattr(self.controller, "tree_root_key", None) or "0"
indices = []
current = node
while self._node_is_valid(current) and current != self.source_model_root:
try:
parent = current.get_parent()
except Exception:
try:
parent = current.getParent()
except Exception:
return None
if not self._node_is_valid(parent):
return None
child_index = None
for idx, child in enumerate(self._iter_children(parent)):
if child == current:
child_index = idx
break
if child_index is None:
return None
indices.append(str(child_index))
current = parent
if current != self.source_model_root:
return None
indices.reverse()
return "0/" + "/".join(indices) if indices else "0"
def sync_runtime_material_from_source_node(self, source_node):
"""Push one source-node material change into matching runtime objects only."""
if not self.controller or not self._node_is_valid(source_node):
return False
if not self.is_source_tree_node(source_node):
return False
owner_key = self._resolve_tree_key_for_source_node(source_node)
if not owner_key:
return False
property_helpers = getattr(self.world, "property_helpers", None) if getattr(self, "world", None) else None
if property_helpers is None:
return False
get_material_fn = getattr(property_helpers, "_get_renderable_node_material", None)
ensure_material_fn = getattr(property_helpers, "_ensure_material_for_node", None)
clone_material_fn = getattr(property_helpers, "_clone_material_for_node", None)
apply_geom_fn = getattr(property_helpers, "_apply_material_to_geom_states", None)
apply_surface_fn = getattr(property_helpers, "_apply_material_surface_state", None)
refresh_pipeline_fn = getattr(property_helpers, "_refresh_pipeline_material_mode", None)
if not callable(apply_geom_fn) or not callable(apply_surface_fn):
return False
material = None
if callable(get_material_fn):
try:
material = get_material_fn(source_node)
except Exception:
material = None
if material is None and callable(ensure_material_fn):
try:
material = ensure_material_fn(source_node)
except Exception:
material = None
if material is None:
return False
matched = False
for gid, obj_np in (self.controller.id_to_object_np or {}).items():
try:
if self.controller.id_to_name.get(gid) != owner_key:
continue
except Exception:
continue
if not self._node_is_valid(obj_np):
continue
runtime_material = material
if callable(clone_material_fn):
try:
runtime_material = clone_material_fn(material, obj_np)
except Exception:
runtime_material = material
try:
apply_geom_fn(obj_np, runtime_material)
apply_surface_fn(obj_np, runtime_material)
if callable(refresh_pipeline_fn):
refresh_pipeline_fn(obj_np, runtime_material)
matched = True
except Exception:
continue
return matched
def _snapshot_runtime_materials_to_source_root(self):
"""
Persist runtime-edited material/geom render state back to source_model_root.
This keeps project save/load consistent for SSBO editing workflow.
"""
controller = self.controller
if not controller or not self.source_model_root:
return
synced = 0
effect_tags_synced = set()
root_effect_tags = {}
property_helpers = getattr(self.world, "property_helpers", None) if getattr(self, "world", None) else None
capture_snapshot_fn = getattr(property_helpers, "_capture_node_material_snapshot", None) if property_helpers else None
normalize_snapshot_fn = getattr(property_helpers, "_normalize_material_snapshot", None) if property_helpers else None
apply_snapshot_fn = getattr(property_helpers, "_apply_node_material_snapshot", None) if property_helpers else None
get_materials_fn = getattr(property_helpers, "_get_node_materials", None) if property_helpers else None
ensure_material_fn = getattr(property_helpers, "_ensure_material_for_node", None) if property_helpers else None
try:
model_root = self.model
if model_root and not model_root.is_empty():
for tag_name in (
"material_effect_metallic_enabled",
"material_effect_default_texture_enabled",
"material_effect_parallax_enabled",
"material_render_effect_signature",
):
if model_root.hasTag(tag_name):
root_effect_tags[tag_name] = model_root.getTag(tag_name)
except Exception:
root_effect_tags = {}
def _node_has_metallic_texture(node_np):
try:
stages = node_np.findAllTextureStages()
for i in range(stages.getNumTextureStages()):
stage = stages.getTextureStage(i)
if not stage:
continue
try:
sname = (stage.getName() or "").lower()
except Exception:
sname = ""
# convention in project helpers and RP metallic workflow
if "metallic" in sname or sname == "p3d_texture5":
return True
except Exception:
pass
return False
def _clone_snapshot_for_target(source_snapshot, target_node):
if not callable(normalize_snapshot_fn):
return None
snapshot = normalize_snapshot_fn(source_snapshot)
if snapshot is None:
return None
target_materials = []
try:
if callable(get_materials_fn):
target_materials = list(get_materials_fn(target_node) or [])
except Exception:
target_materials = []
if not target_materials and callable(ensure_material_fn):
try:
fallback = ensure_material_fn(target_node)
if fallback is not None:
target_materials = [fallback]
except Exception:
target_materials = []
source_entries = snapshot.get("materials", []) or []
cloned_entries = []
for idx, entry in enumerate(source_entries):
target_material = target_materials[idx] if idx < len(target_materials) else None
cloned_entries.append({
"material": target_material,
"base_color": entry.get("base_color"),
"roughness": entry.get("roughness"),
"metallic": entry.get("metallic"),
"ior": entry.get("ior"),
"emission": entry.get("emission"),
})
node_state = snapshot.get("node_state", {}) or {}
textures = dict(node_state.get("textures", {}) or {})
effect_tags = dict(node_state.get("effect_tags", {}) or {})
return {
"materials": cloned_entries,
"node_state": {
"textures": textures,
"effect_tags": effect_tags,
},
}
grouped_entries = {}
def _snapshot_score(snapshot, obj_np):
score = 0
try:
snapshot = normalize_snapshot_fn(snapshot) if callable(normalize_snapshot_fn) else snapshot
except Exception:
pass
if isinstance(snapshot, dict):
node_state = snapshot.get("node_state", {}) or {}
textures = node_state.get("textures", {}) or {}
effect_tags = node_state.get("effect_tags", {}) or {}
score += len([value for value in textures.values() if value]) * 100
score += len([value for value in effect_tags.values() if value]) * 50
for entry in snapshot.get("materials", []) or []:
if entry.get("base_color") is not None:
score += 5
for scalar_name in ("roughness", "metallic", "ior"):
if entry.get(scalar_name) is not None:
score += 3
if entry.get("emission") is not None:
score += 2
try:
if _node_has_metallic_texture(obj_np):
score += 25
except Exception:
pass
return score
for gid, obj_np in controller.id_to_object_np.items():
if not self._node_is_valid(obj_np):
continue
owner_key = controller.id_to_name.get(gid)
if not owner_key:
continue
source_node = self._resolve_source_node_by_tree_key(owner_key)
if not self._node_is_valid(source_node):
continue
runtime_snapshot = None
if callable(capture_snapshot_fn):
try:
runtime_snapshot = capture_snapshot_fn(obj_np)
except Exception:
runtime_snapshot = None
source_snapshot = None
if callable(capture_snapshot_fn):
try:
source_snapshot = capture_snapshot_fn(source_node)
except Exception:
source_snapshot = None
source_node_key = id(source_node)
entry = grouped_entries.get(source_node_key)
runtime_score = _snapshot_score(runtime_snapshot, obj_np)
source_score = _snapshot_score(source_snapshot, source_node)
# Saving should prefer the visible runtime state on ties; otherwise a
# stale source snapshot can overwrite a freshly edited child material.
if runtime_snapshot is not None and runtime_score >= source_score:
chosen_snapshot = runtime_snapshot
else:
chosen_snapshot = source_snapshot
candidate_score = max(runtime_score, source_score)
if entry is None or candidate_score > entry["score"]:
grouped_entries[source_node_key] = {
"gid": gid,
"obj_np": obj_np,
"source_node": source_node,
"snapshot": chosen_snapshot,
"score": candidate_score,
}
for source_node_key, entry in grouped_entries.items():
gid = entry["gid"]
obj_np = entry["obj_np"]
source_node = entry["source_node"]
source_snapshot = entry["snapshot"]
if source_node_key not in effect_tags_synced:
inferred_metallic = _node_has_metallic_texture(obj_np)
for tag_name in (
"material_effect_metallic_enabled",
"material_effect_default_texture_enabled",
"material_effect_parallax_enabled",
"material_render_effect_signature",
):
try:
if obj_np.hasTag(tag_name):
source_node.setTag(tag_name, obj_np.getTag(tag_name))
elif tag_name == "material_effect_metallic_enabled" and inferred_metallic:
source_node.setTag(tag_name, "1")
elif tag_name in root_effect_tags:
source_node.setTag(tag_name, root_effect_tags[tag_name])
except Exception:
pass
texture_slot_tags = (
"material_texture_diffuse",
"material_texture_normal",
"material_texture_ior",
"material_texture_roughness",
"material_texture_parallax",
"material_texture_metallic",
"material_texture_emission",
"material_texture_ao",
"material_texture_alpha",
"material_texture_detail",
"material_texture_gloss",
)
for tag_name in texture_slot_tags:
try:
if obj_np.hasTag(tag_name):
source_node.setTag(tag_name, obj_np.getTag(tag_name))
elif source_node.hasTag(tag_name):
source_node.clearTag(tag_name)
except Exception:
pass
effect_tags_synced.add(source_node_key)
if source_snapshot is not None and callable(apply_snapshot_fn):
try:
target_snapshot = _clone_snapshot_for_target(source_snapshot, source_node)
if target_snapshot is not None:
apply_snapshot_fn(source_node, target_snapshot)
source_node.setTag("scene_material_dirty", "true")
synced += 1
continue
except Exception:
pass
runtime_geom_state = None
try:
geom_paths = obj_np.findAllMatches("**/+GeomNode")
if geom_paths and geom_paths.getNumPaths() > 0:
runtime_geom_state = geom_paths.getPath(0).getNetState()
except Exception:
runtime_geom_state = None
if runtime_geom_state is None:
continue
try:
src_gnode = source_node.node()
set_src_state = getattr(src_gnode, "set_geom_state", None) or getattr(src_gnode, "setGeomState", None)
get_src_count = getattr(src_gnode, "get_num_geoms", None) or getattr(src_gnode, "getNumGeoms", None)
src_geom_count = int(get_src_count()) if callable(get_src_count) else 0
except Exception:
continue
src_geom_index = controller.id_to_geom_index.get(gid, 0)
if src_geom_index < 0 or src_geom_index >= src_geom_count:
continue
try:
if callable(set_src_state):
set_src_state(src_geom_index, runtime_geom_state)
source_node.setTag("scene_material_dirty", "true")
synced += 1
except Exception:
continue
if synced:
print(f"[SSBOEditor] Synced runtime material states back to source tree: {synced}")
def _restore_saved_material_bindings_from_tags(self, root_np):
"""Rebind saved texture/effect tags back onto loaded source nodes."""
if not self._node_is_valid(root_np):
return 0
property_helpers = getattr(self.base, "property_helpers", None)
capture_snapshot_fn = getattr(property_helpers, "_capture_node_material_snapshot", None) if property_helpers else None
apply_snapshot_fn = getattr(property_helpers, "_apply_node_material_snapshot", None) if property_helpers else None
texture_slots_fn = getattr(property_helpers, "_get_material_texture_slots", None) if property_helpers else None
if not callable(capture_snapshot_fn) or not callable(apply_snapshot_fn):
return 0
texture_types = []
if callable(texture_slots_fn):
try:
texture_types = list((texture_slots_fn() or {}).keys())
except Exception:
texture_types = []
if not texture_types:
texture_types = [
"diffuse", "normal", "ior", "roughness", "parallax",
"metallic", "emission", "ao", "alpha", "detail", "gloss",
]
effect_tag_names = (
"material_effect_metallic_enabled",
"material_effect_default_texture_enabled",
"material_effect_parallax_enabled",
"material_render_effect_signature",
)
restored = 0
try:
descendant_nodes = list(root_np.find_all_matches("**"))
except Exception:
try:
descendant_nodes = list(root_np.findAllMatches("**"))
except Exception:
descendant_nodes = []
for node in [root_np] + descendant_nodes:
if not self._node_is_valid(node):
continue
has_texture_tags = any(node.hasTag(f"material_texture_{texture_type}") for texture_type in texture_types)
has_effect_tags = any(node.hasTag(tag_name) for tag_name in effect_tag_names)
if not (has_texture_tags or has_effect_tags):
continue
try:
snapshot = capture_snapshot_fn(node)
except Exception:
snapshot = None
if not snapshot:
continue
try:
apply_snapshot_fn(node, snapshot)
restored += 1
except Exception:
continue
if restored:
print(f"[SSBOEditor] Restored saved material bindings from tags: {restored}")
return restored
def _clear_runtime_state(self, preserve_source_models=False, snapshot_transform=True):
"""Remove runtime SSBO controller/model state while optionally keeping source snapshots."""
self.clear_selection(snapshot_transform=snapshot_transform)
self._cleanup_group_proxy()
self._reset_pick_sync_cache()
self._teardown_gpu_picking()
controller = self.controller
pick_model = getattr(controller, "pick_model", None) if controller else None
model = self.model
removable_nodes = [pick_model, model]
if not preserve_source_models:
removable_nodes.extend([self.source_model, self.source_model_root])
for node in removable_nodes:
if not node:
continue
try:
if not node.is_empty():
node.remove_node()
except Exception:
try:
if not node.isEmpty():
node.removeNode()
except Exception:
pass
self.controller = None
self.model = None
self.selected_name = None
self.selected_ids = []
self.transform_ids = []
self.last_import_tree_key = None
self.last_import_root_name = None
self._model_root_last_snapshot_mat = LMatrix4f.ident_mat()
self._runtime_owner_base_mats = {}
if not preserve_source_models:
self.source_model = None
self.source_model_root = None
self._source_child_base_mats = {}
self._sync_pick_scene_binding()
def _get_source_root_children(self):
root = self.source_model_root
if not root:
return []
return [child for child in self._iter_children(root) if self._node_is_valid(child)]
def _rebuild_or_clear_runtime_from_current_source(self, highlight_root_name=None):
if self._get_source_root_children():
self.source_model = self.source_model_root
self._rebuild_runtime_from_source_root(highlight_root_name=highlight_root_name)
return self.model
self.last_import_tree_key = None
self.last_import_root_name = None
self.source_model = self.source_model_root
self._sync_pick_scene_binding()
return None
def find_source_child_by_name(self, child_name):
if not child_name:
return None
for child in self._get_source_root_children():
if self._get_node_name(child) == child_name:
return child
return None
def detach_source_child(self, child_name=None, child_np=None):
target = child_np if self._node_is_valid(child_np) else self.find_source_child_by_name(child_name)
if not self._node_is_valid(target):
return None
if self.controller and self.model:
self._snapshot_top_level_transforms_to_source_root()
self._clear_runtime_state(preserve_source_models=True, snapshot_transform=False)
try:
target.detach_node()
except Exception:
try:
target.detachNode()
except Exception:
return None
self._rebuild_or_clear_runtime_from_current_source()
return target
def delete_selected_source_node(self):
"""Delete the current SSBO selection from source_model_root and rebuild runtime."""
if not self.controller or self.selected_name is None:
return False
target = self._resolve_source_node_by_tree_key(self.selected_name)
if not self._node_is_valid(target):
return False
source_root = self._ensure_source_model_root()
if target == source_root:
children = self._get_source_root_children()
if not children:
return False
if self.controller and self.model:
self._snapshot_top_level_transforms_to_source_root()
self._clear_runtime_state(preserve_source_models=True, snapshot_transform=False)
for child in children:
try:
child.detach_node()
except Exception:
try:
child.detachNode()
except Exception:
continue
self.clear_selection(sync_world_selection=False, snapshot_transform=False)
self._rebuild_or_clear_runtime_from_current_source()
return True
if self.controller and self.model:
self._snapshot_top_level_transforms_to_source_root()
self._clear_runtime_state(preserve_source_models=True, snapshot_transform=False)
try:
target.detach_node()
except Exception:
try:
target.detachNode()
except Exception:
return False
self.clear_selection(sync_world_selection=False, snapshot_transform=False)
self._rebuild_or_clear_runtime_from_current_source()
return True
def attach_source_child(self, child_np, highlight_root_name=None):
if not self._node_is_valid(child_np):
return None
if self.controller and self.model:
self._snapshot_top_level_transforms_to_source_root()
self._clear_runtime_state(preserve_source_models=True, snapshot_transform=False)
source_root = self._ensure_source_model_root()
target_name = highlight_root_name or self._get_node_name(child_np, "imported_model")
self._set_node_name(child_np, target_name)
try:
child_np.reparent_to(source_root)
except Exception:
try:
child_np.reparentTo(source_root)
except Exception:
return None
self._rebuild_or_clear_runtime_from_current_source(highlight_root_name=target_name)
return self.model
def _rebuild_runtime_from_source_root(self, highlight_root_name=None):
root = self._ensure_source_model_root()
working_holder = NodePath("ssbo_source_scene_work")
working_root = root.copy_to(working_holder)
runtime_root_mat = None
source_children = self._get_source_root_children()
if len(source_children) == 1 and self._node_is_valid(source_children[0]):
try:
working_children = [child for child in self._iter_children(working_root) if self._node_is_valid(child)]
except Exception:
working_children = []
if len(working_children) == 1 and self._node_is_valid(working_children[0]):
working_child = working_children[0]
try:
runtime_root_mat = LMatrix4f(working_child.get_mat(working_root))
except Exception:
try:
runtime_root_mat = LMatrix4f(working_child.getMat(working_root))
except Exception:
runtime_root_mat = None
if runtime_root_mat is not None:
try:
working_child.set_mat(LMatrix4f.ident_mat())
except Exception:
try:
working_child.setMat(LMatrix4f.ident_mat())
except Exception:
runtime_root_mat = None
self.controller = ObjectController()
try:
geom_count = len(list(working_root.find_all_matches("**/+GeomNode")))
except Exception:
geom_count = 0
try:
node_count = len(list(working_root.find_all_matches("**")))
except Exception:
node_count = 0
skip_heavy_material_processing = self._should_skip_heavy_material_processing(
geom_count=geom_count,
node_count=node_count,
)
if skip_heavy_material_processing:
print(
f"[SSBOEditor] Skip heavy material processing for large scene "
f"(nodes={node_count}, geoms={geom_count})"
)
else:
self._detach_shared_materials_in_subtree(working_root)
self._materialize_explicit_geom_materials(working_root)
# Large scenes should prefer the old flat import path.
# Hybrid mode keeps per-child editing, but for vegetation/campus-like GLB
# scenes it explodes into tens of thousands of runtime objects/chunks and
# the editor becomes unusable or crashes.
# 提高使用 hybrid 模式的阈值。
# 对于类似 jyc.glb 的场景hybrid 模式的对象分块chunks比一个巨大的 flat 节点拥有更好的剔除性能和帧率。
# 只有在极端规模(如 > 20000 节点或 > 10000 几何体)的情况下才强制退回到 flat 模式以节省内存。
prefer_flat_mode = (
geom_count > 10000
or node_count > 20000
)
use_hybrid_mode = geom_count > 0 and not prefer_flat_mode
if prefer_flat_mode:
print(
f"[SSBOEditor] Large scene uses flat runtime path "
f"(nodes={node_count}, geoms={geom_count})"
)
if use_hybrid_mode:
count = self.controller.bake_ids_and_collect_hybrid(working_root)
else:
count = self.controller.bake_ids_and_collect(
working_root,
lightweight=prefer_flat_mode,
)
self.model = self.controller.model
self.model.reparent_to(self.base.render)
if runtime_root_mat is not None:
try:
self.model.set_mat(runtime_root_mat)
except Exception:
try:
self.model.setMat(runtime_root_mat)
except Exception:
pass
self._capture_runtime_owner_base_mats()
self.set_realtime_shadow_updates(self.realtime_shadow_updates)
self.setup_gpu_picking()
self._sync_pick_model_transform()
self.last_import_tree_key = None
self.last_import_root_name = highlight_root_name
if highlight_root_name:
root_key = getattr(self.controller, "tree_root_key", None)
root_node = self.controller.tree_nodes.get(root_key, {}) if root_key else {}
for child_key in root_node.get("children", []):
if self.controller.display_names.get(child_key) == highlight_root_name:
self.last_import_tree_key = child_key
break
try:
if not working_holder.is_empty():
working_holder.remove_node()
except Exception:
try:
if not working_holder.isEmpty():
working_holder.removeNode()
except Exception:
pass
mode_name = "hybrid" if use_hybrid_mode else "flat"
print(f"[SSBOEditor] Model loaded. Total objects: {count}, mode={mode_name}, geoms={geom_count}")
def load_model(
self,
model_path,
keep_source_model=False,
append=False,
scene_package_import=False,
rebuild_runtime=True,
):
"""Load and process one model into the aggregated SSBO scene."""
print(f"[SSBOEditor] Loading model: {model_path}")
skip_heavy_material_processing = self._should_skip_heavy_material_processing(model_path=model_path)
# scene_package_import: loading saved scene.bam from project.
# Keep stored material/texture states intact; repair heuristics can
# misclassify valid packed/relative texture refs and cause dark materials.
should_repair_textures = (not scene_package_import) and (not skip_heavy_material_processing)
should_fix_black_materials = (not scene_package_import) and (not skip_heavy_material_processing)
if skip_heavy_material_processing:
model_size_mb = self._get_model_file_size(model_path) / (1024.0 * 1024.0)
print(
f"[SSBOEditor] Large model fast path enabled for {os.path.basename(model_path)} "
f"({model_size_mb:.1f} MB)"
)
source_model = self._load_source_model_from_path(
model_path,
apply_black_fix=should_fix_black_materials,
repair_textures=should_repair_textures,
)
model_name = os.path.basename(model_path)
if model_name and not scene_package_import:
self._set_node_name(source_model, model_name)
if append and self.source_model_root:
if self.controller and self.model:
self._snapshot_top_level_transforms_to_source_root()
self._clear_runtime_state(preserve_source_models=True, snapshot_transform=False)
else:
self._clear_runtime_state(preserve_source_models=False)
source_root = self._ensure_source_model_root()
# 项目场景包导入(scene.bam)时,避免再包一层 "scene.bam" 根节点,
# 直接把其顶层子节点并入 source_root保持场景树与保存时一致。
if scene_package_import:
imported_roots = []
skipped_legacy_roots = 0
children = []
try:
children = [c for c in source_model.get_children() if c and not c.is_empty()]
except Exception:
try:
children = [c for c in source_model.getChildren() if c and not c.isEmpty()]
except Exception:
children = []
for child in children:
try:
child_name = self._get_node_name(child, "")
if child_name in {"render", "render2d", "aspect2d"}:
continue
if self._should_skip_scene_package_child_for_ssbo(child):
skipped_legacy_roots += 1
print(f"[SSBOSceneLoad] 跳过普通/动画模型: {child_name}")
continue
imported_child = child.copyTo(source_root)
if not skip_heavy_material_processing:
self._detach_shared_materials_in_subtree(imported_child)
imported_child.setTag("ssbo_managed", "true")
imported_roots.append(imported_child)
except Exception:
continue
if not imported_roots and skipped_legacy_roots == 0:
fallback_name = self._get_node_name(source_model, "scene_root")
unique_root_name = self._make_unique_source_child_name(fallback_name)
self._set_node_name(source_model, unique_root_name)
imported_root = source_model.copyTo(source_root)
self._set_node_name(imported_root, unique_root_name)
imported_root.setTag("ssbo_managed", "true")
if not skip_heavy_material_processing:
self._detach_shared_materials_in_subtree(imported_root)
imported_roots = [imported_root]
self.source_model = source_root
self._restore_saved_material_bindings_from_tags(source_root)
self._capture_source_child_base_mats()
if rebuild_runtime:
if self._get_source_root_children():
self._rebuild_runtime_from_source_root(highlight_root_name=None)
else:
self._rebuild_or_clear_runtime_from_current_source()
if len(imported_roots) == 1:
return imported_roots[0]
return source_root
unique_root_name = self._make_unique_source_child_name(model_name or "imported_model")
self._set_node_name(source_model, unique_root_name)
imported_root = source_model.copyTo(source_root)
self._set_node_name(imported_root, unique_root_name)
imported_root.setTag("ssbo_managed", "true")
if not skip_heavy_material_processing:
self._detach_shared_materials_in_subtree(imported_root)
if keep_source_model and not append:
self.source_model = imported_root
else:
self.source_model = source_root
self._capture_source_child_base_mats()
if rebuild_runtime:
self._rebuild_runtime_from_source_root(highlight_root_name=unique_root_name)
return imported_root
def _build_filename_candidates(self, path_text):
"""Build Filename candidates with wide-char first for Windows CJK paths."""
candidates = []
seen = set()
for ctor_name in ("fromOsSpecificW", "from_os_specific_w", "fromOsSpecific", "from_os_specific"):
ctor = getattr(Filename, ctor_name, None)
if not ctor:
continue
try:
fn = ctor(path_text)
key = fn.get_fullpath()
if key in seen:
continue
seen.add(key)
candidates.append(fn)
except Exception:
continue
if not candidates:
try:
candidates.append(Filename(path_text))
except Exception:
pass
return candidates
def _load_texture_from_path(self, texture_path):
"""Load texture with robust path constructors."""
for fn in self._build_filename_candidates(texture_path):
try:
tex = self.base.loader.loadTexture(fn)
if tex:
return tex
except Exception:
continue
return None
def _build_texture_search_dirs(self, model_path):
"""Build candidate directories for missing texture recovery."""
dirs = []
model_dir = os.path.dirname(os.path.abspath(model_path))
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
def add_dir(path):
if not path:
return
path = os.path.normpath(path)
if path in dirs:
return
if os.path.isdir(path):
dirs.append(path)
add_dir(model_dir)
try:
if os.path.isdir(model_dir):
for item in os.listdir(model_dir):
if item.lower().endswith('.fbm'):
fbm_dir = os.path.join(model_dir, item)
if os.path.isdir(fbm_dir):
add_dir(fbm_dir)
for sub in ("textures", "texture", "tex", "assets", "materials"):
add_dir(os.path.join(fbm_dir, sub))
except Exception:
pass
for sub in ("textures", "texture", "tex", "assets", "materials"):
add_dir(os.path.join(model_dir, sub))
parent = model_dir
for _ in range(2):
parent = os.path.dirname(parent)
if not parent:
break
for sub in ("textures", "texture", "tex", "assets", "materials"):
add_dir(os.path.join(parent, sub))
add_dir(os.path.join(project_root, "Resources"))
add_dir(os.path.join(project_root, "Resources", "textures"))
add_dir(os.path.join(project_root, "Resources", "materials"))
add_dir(os.path.join(project_root, "Resources", "models"))
return dirs
def _index_texture_files(self, dirs, limit=30000):
"""Index texture files by basename for fast lookup."""
texture_exts = {".png", ".jpg", ".jpeg", ".tga", ".bmp", ".dds", ".ktx", ".ktx2", ".webp"}
index = {}
scanned = 0
for root_dir in dirs:
try:
for root, _, files in os.walk(root_dir):
for filename in files:
ext = os.path.splitext(filename)[1].lower()
if ext not in texture_exts:
continue
key = filename.lower()
if key not in index:
index[key] = os.path.join(root, filename)
scanned += 1
if scanned >= limit:
return index
except Exception:
continue
return index
def _repair_missing_textures(self, model_np, model_path):
"""
Repair broken texture paths by basename search; if unresolved, clear that
missing texture binding to avoid black PBR sampling.
"""
if not model_np or model_np.is_empty():
return
search_dirs = self._build_texture_search_dirs(model_path)
texture_index = self._index_texture_files(search_dirs)
fixed = 0
cleared = 0
white_tex = self._get_white_fallback_texture()
nodes = [model_np] + list(model_np.find_all_matches("**"))
for node in nodes:
if not node or node.is_empty():
continue
try:
stages = node.find_all_texture_stages()
except Exception:
try:
stages = node.findAllTextureStages()
except Exception:
continue
try:
stage_count = stages.get_num_texture_stages()
stage_at = stages.get_texture_stage
except Exception:
try:
stage_count = stages.getNumTextureStages()
stage_at = stages.getTextureStage
except Exception:
continue
for i in range(stage_count):
stage = stage_at(i)
if not stage:
continue
try:
tex = node.get_texture(stage)
except Exception:
tex = node.getTexture(stage)
if not tex:
continue
if self._texture_is_valid(tex):
continue
basename = self._extract_texture_basename(tex)
if not basename:
continue
replacement = texture_index.get(basename.lower())
if replacement:
try:
new_tex = self._load_texture_from_path(replacement)
if new_tex:
try:
node.set_texture(stage, new_tex, 1)
except Exception:
node.setTexture(stage, new_tex, 1)
fixed += 1
continue
except Exception:
pass
# Missing texture with no replacement: CLEAR the texture binding.
# Do NOT use a white fallback texture, because if this is a Metallic or Roughness
# map, pure white will force Metallic=1.0 and Roughness=1.0, which turns models black!
try:
try:
node.clear_texture(stage)
except Exception:
node.clearTexture(stage)
cleared += 1
except Exception:
pass
# GeomState-level texture bindings (common in imported FBX/GLTF):
# inspect and apply the same fallback on this GeomNode path.
try:
gnode = node.node()
get_num_geoms = getattr(gnode, "get_num_geoms", None) or getattr(gnode, "getNumGeoms", None)
get_geom_state = getattr(gnode, "get_geom_state", None) or getattr(gnode, "getGeomState", None)
geom_count = int(get_num_geoms()) if callable(get_num_geoms) else 0
except Exception:
geom_count = 0
for gi in range(geom_count):
try:
state = get_geom_state(gi)
except Exception:
continue
if state is None:
continue
tattr = None
try:
if state.has_attrib(TextureAttrib.get_class_type()):
tattr = state.get_attrib(TextureAttrib.get_class_type())
except Exception:
try:
if state.hasAttrib(TextureAttrib.getClassType()):
tattr = state.getAttrib(TextureAttrib.getClassType())
except Exception:
tattr = None
if not tattr:
continue
try:
num_on = tattr.get_num_on_stages()
get_stage = tattr.get_on_stage
get_tex = tattr.get_on_texture
except Exception:
try:
num_on = tattr.getNumOnStages()
get_stage = tattr.getOnStage
get_tex = tattr.getOnTexture
except Exception:
continue
for si in range(int(num_on)):
try:
stage = get_stage(si)
tex = get_tex(stage)
except Exception:
continue
if not stage or not tex:
continue
if self._texture_is_valid(tex):
continue
basename = self._extract_texture_basename(tex)
replacement = texture_index.get(basename.lower()) if basename else None
if replacement:
try:
new_tex = self._load_texture_from_path(replacement)
if new_tex:
try:
node.set_texture(stage, new_tex, 100000)
except Exception:
node.setTexture(stage, new_tex, 100000)
fixed += 1
continue
except Exception:
pass
try:
try:
node.clear_texture(stage)
except Exception:
node.clearTexture(stage)
cleared += 1
except Exception:
pass
if fixed or cleared:
print(f"[SSBOEditor] Texture repair: fixed={fixed}, cleared_missing={cleared}")
self._apply_nonblack_material_fallback(model_np)
def _texture_is_valid(self, tex):
if not tex:
return False
tex_path = self._extract_texture_os_path(tex)
if tex_path and os.path.exists(tex_path):
return True
if tex_path:
# Some valid textures use Panda VFS virtual paths; keep them RAM-checkable.
path_norm = tex_path.replace("\\", "/")
if path_norm.startswith("/$$") or path_norm.startswith("$$"):
pass
else:
# File-backed texture with missing source path should be considered invalid,
# even when Panda keeps a tiny fallback image in RAM.
return False
try:
if tex.has_ram_image():
return tex.get_ram_image_size() > 0
except Exception:
try:
if tex.hasRamImage():
return tex.getRamImageSize() > 0
except Exception:
pass
return False
def _extract_texture_os_path(self, tex):
tex_path = ""
try:
if tex.has_fullpath():
fullpath = tex.get_fullpath()
try:
tex_path = fullpath.to_os_specific()
except Exception:
try:
tex_path = fullpath.toOsSpecific()
except Exception:
tex_path = str(fullpath)
except Exception:
try:
if tex.hasFullpath():
fullpath = tex.getFullpath()
try:
tex_path = fullpath.toOsSpecific()
except Exception:
tex_path = str(fullpath)
except Exception:
tex_path = ""
tex_path = str(tex_path or "").strip()
if not tex_path:
return ""
tex_path = os.path.normpath(tex_path)
if os.path.exists(tex_path):
return tex_path
# Convert Panda internal drive path (/d/foo/bar) to Windows path if needed.
if len(tex_path) >= 3 and tex_path[0] in ("/", "\\") and tex_path[1].isalpha() and tex_path[2] in ("/", "\\"):
drive_path = f"{tex_path[1]}:{tex_path[2:]}"
drive_path = os.path.normpath(drive_path)
if os.path.exists(drive_path):
return drive_path
return tex_path
def _extract_texture_basename(self, tex):
tex_path = self._extract_texture_os_path(tex)
if tex_path:
return os.path.basename(tex_path.replace("\\", "/"))
try:
name = tex.get_name()
except Exception:
try:
name = tex.getName()
except Exception:
name = ""
return os.path.basename(str(name).replace("\\", "/"))
def _get_white_fallback_texture(self):
tex = getattr(self, "_ssbo_white_fallback_tex", None)
if tex:
return tex
try:
img = PNMImage(2, 2, 4)
img.fill(1.0, 1.0, 1.0)
img.alpha_fill(1.0)
tex = Texture("ssbo_white_fallback")
tex.load(img)
tex.set_minfilter(Texture.FT_nearest)
tex.set_magfilter(Texture.FT_nearest)
self._ssbo_white_fallback_tex = tex
return tex
except Exception:
return None
def _node_has_valid_texture(self, node):
if not node or node.is_empty():
return False
try:
stages = node.find_all_texture_stages()
stage_count = stages.get_num_texture_stages()
stage_at = stages.get_texture_stage
except Exception:
try:
stages = node.findAllTextureStages()
stage_count = stages.getNumTextureStages()
stage_at = stages.getTextureStage
except Exception:
return False
for i in range(stage_count):
stage = stage_at(i)
if not stage:
continue
try:
tex = node.get_texture(stage)
except Exception:
tex = node.getTexture(stage)
if self._texture_is_valid(tex):
return True
# GeomState TextureAttrib bindings can hold the effective textures.
try:
gnode = node.node()
get_num_geoms = getattr(gnode, "get_num_geoms", None) or getattr(gnode, "getNumGeoms", None)
get_geom_state = getattr(gnode, "get_geom_state", None) or getattr(gnode, "getGeomState", None)
geom_count = int(get_num_geoms()) if callable(get_num_geoms) else 0
except Exception:
geom_count = 0
for gi in range(geom_count):
try:
state = get_geom_state(gi)
except Exception:
continue
if state is None:
continue
tattr = None
try:
if state.has_attrib(TextureAttrib.get_class_type()):
tattr = state.get_attrib(TextureAttrib.get_class_type())
except Exception:
try:
if state.hasAttrib(TextureAttrib.getClassType()):
tattr = state.getAttrib(TextureAttrib.getClassType())
except Exception:
tattr = None
if not tattr:
continue
try:
num_on = tattr.get_num_on_stages()
get_stage = tattr.get_on_stage
get_tex = tattr.get_on_texture
except Exception:
try:
num_on = tattr.getNumOnStages()
get_stage = tattr.getOnStage
get_tex = tattr.getOnTexture
except Exception:
continue
for si in range(int(num_on)):
try:
stage = get_stage(si)
tex = get_tex(stage)
except Exception:
continue
if self._texture_is_valid(tex):
return True
return False
def _is_node_material_dark(self, node):
"""Heuristic: detect near-black material/color state."""
if not node or node.is_empty():
return False
try:
state = node.get_state()
except Exception:
try:
state = node.getState()
except Exception:
return True
def _vec_dark(v):
try:
return max(float(v[0]), float(v[1]), float(v[2])) < 0.08
except Exception:
try:
return max(float(v.x), float(v.y), float(v.z)) < 0.08
except Exception:
return False
# Material color
mat = None
try:
if state.has_attrib(MaterialAttrib.get_class_type()):
mat_attr = state.get_attrib(MaterialAttrib.get_class_type())
mat = mat_attr.get_material()
except Exception:
try:
if state.hasAttrib(MaterialAttrib.getClassType()):
mat_attr = state.getAttrib(MaterialAttrib.getClassType())
mat = mat_attr.getMaterial()
except Exception:
mat = None
if mat:
for has_name, get_name in (
("has_base_color", "get_base_color"),
("has_diffuse", "get_diffuse"),
("hasBaseColor", "getBaseColor"),
("hasDiffuse", "getDiffuse"),
):
has_fn = getattr(mat, has_name, None)
get_fn = getattr(mat, get_name, None)
if callable(has_fn) and callable(get_fn):
try:
if has_fn():
return _vec_dark(get_fn())
except Exception:
continue
return False
# ColorAttrib fallback
try:
if state.has_attrib(ColorAttrib.get_class_type()):
color_attr = state.get_attrib(ColorAttrib.get_class_type())
if not color_attr.is_off():
return _vec_dark(color_attr.get_color())
except Exception:
try:
if state.hasAttrib(ColorAttrib.getClassType()):
color_attr = state.getAttrib(ColorAttrib.getClassType())
if not color_attr.isOff():
return _vec_dark(color_attr.getColor())
except Exception:
pass
return True
def _apply_nonblack_material_fallback(self, model_np):
"""If a geom has no texture and is effectively black, apply a neutral material."""
if not model_np or model_np.is_empty():
return
neutral = Material()
neutral.set_base_color((0.75, 0.75, 0.75, 1.0)) if hasattr(neutral, "set_base_color") else neutral.setBaseColor((0.75, 0.75, 0.75, 1.0))
neutral.set_diffuse((0.75, 0.75, 0.75, 1.0)) if hasattr(neutral, "set_diffuse") else neutral.setDiffuse((0.75, 0.75, 0.75, 1.0))
neutral.set_ambient((0.22, 0.22, 0.22, 1.0)) if hasattr(neutral, "set_ambient") else neutral.setAmbient((0.22, 0.22, 0.22, 1.0))
neutral.set_specular((0.1, 0.1, 0.1, 1.0)) if hasattr(neutral, "set_specular") else neutral.setSpecular((0.1, 0.1, 0.1, 1.0))
neutral.set_shininess(8.0) if hasattr(neutral, "set_shininess") else neutral.setShininess(8.0)
patched = 0
for geom_np in model_np.find_all_matches("**/+GeomNode"):
if self._node_has_valid_texture(geom_np):
continue
if not self._is_node_material_dark(geom_np):
continue
try:
try:
geom_np.set_material(neutral, 1)
except Exception:
geom_np.setMaterial(neutral, 1)
try:
geom_np.set_color_scale(1.0, 1.0, 1.0, 1.0)
except Exception:
geom_np.setColorScale(1.0, 1.0, 1.0, 1.0)
patched += 1
except Exception:
continue
if patched:
print(f"[SSBOEditor] Applied non-black fallback material to {patched} geom nodes.")
# No custom effect needed — RP default rendering for maximum FPS
def _inject_ssbo_into_shadow_state(self, effect_path):
"""Inject SSBO inputs into RP shadow tag state"""
try:
if not hasattr(self.rp.tag_mgr, 'containers'): return
shadow_container = self.rp.tag_mgr.containers.get("shadow")
if not shadow_container: return
tag_value = self.model.get_tag(shadow_container.tag_name)
if not tag_value: return
effect = Effect.load(effect_path, {})
if effect is None: return
shadow_shader = effect.get_shader_obj("shadow")
if shadow_shader is None: return
# Since inputs are now on Nodes (Chunks), we just need to ensure the shader is applied.
# extra_inputs is no longer needed if the inputs are on the nodes themselves?
# Wait, RP might override state.
# But specific shader inputs on NodePath have priority over State inputs usually?
# Let's try applying without extra inputs first.
self.rp.tag_mgr.apply_state(
"shadow", self.model, shadow_shader,
tag_value, 65)
print(f"[SSBO Shadow] Re-applied shadow state (tag='{tag_value}')")
except Exception as e:
print(f"[SSBO Shadow] Error injecting shadow state: {e}")
def setup_gpu_picking(self):
"""Setup GPU Picking (Basic implementation)"""
self._teardown_gpu_picking()
controller = getattr(self, "controller", None)
if controller and not getattr(controller, "supports_gpu_picking", True):
print("[GPU Picking] Disabled for lightweight large-scene runtime.")
return
win_props = WindowProperties()
win_props.set_size(1, 1)
fb_props = FrameBufferProperties()
fb_props.set_rgba_bits(8, 8, 8, 8)
fb_props.set_depth_bits(16)
self.pick_buffer = self.base.graphicsEngine.make_output(
self.base.pipe, "pick_buffer", -100,
fb_props, win_props,
GraphicsPipe.BF_refuse_window,
self.base.win.get_gsg(), self.base.win
)
if not self.pick_buffer:
print("[GPU Picking] Failed to create buffer!")
return
self.pick_texture = Texture()
self.pick_texture.set_minfilter(Texture.FT_nearest)
self.pick_texture.set_magfilter(Texture.FT_nearest)
self.pick_buffer.add_render_texture(self.pick_texture, GraphicsOutput.RTM_copy_ram)
self.pick_cam = Camera("pick_camera")
self.pick_cam.set_camera_mask(self.pick_mask)
self.pick_cam_np = self.base.cam.attach_new_node(self.pick_cam)
self.pick_lens = self.base.camLens.make_copy()
self.pick_cam.set_lens(self.pick_lens)
dr = self.pick_buffer.make_display_region()
dr.set_camera(self.pick_cam_np)
# Load pick shader
current_dir = os.path.dirname(os.path.abspath(__file__))
pick_vert_path = os.path.join(current_dir, "shaders", "pick_id.vert")
pick_frag_path = os.path.join(current_dir, "shaders", "pick_id.frag")
try:
# Read shader source directly from OS filesystem to avoid
# Panda3D VFS case-mismatch issues on Windows.
with open(pick_vert_path, 'r', encoding='utf-8') as f:
vert_src = f.read().replace('\r', '')
with open(pick_frag_path, 'r', encoding='utf-8') as f:
frag_src = f.read().replace('\r', '')
pick_shader = Shader.make(Shader.SL_GLSL, vert_src, frag_src)
pick_scene = getattr(self.controller, "pick_model", None) or self.model
if pick_scene and not pick_scene.is_empty():
pick_scene.show(self.pick_mask)
self.pick_cam.set_scene(pick_scene or self._empty_pick_scene)
initial_state = NodePath("initial")
initial_state.set_shader(pick_shader, 100)
# Remove global SSBO input, Chunks have their own inputs
# initial_state.set_shader_input("transforms", ssbo)
self.pick_cam.set_initial_state(initial_state.get_state())
except Exception as e:
print(f"[GPU Picking] Warning: pick shaders failed to load: {e}")
print("Picking disabled.")
return
self.pick_buffer.set_active(False)
self.pick_buffer.set_clear_color(Vec4(0, 0, 0, 0))
self.pick_buffer.set_clear_color_active(True)
def _teardown_gpu_picking(self):
"""Release old GPU picking resources before rebuilding them."""
pick_cam_np = getattr(self, "pick_cam_np", None)
if pick_cam_np is not None:
try:
if not pick_cam_np.is_empty():
pick_cam_np.remove_node()
except Exception:
try:
if not pick_cam_np.isEmpty():
pick_cam_np.removeNode()
except Exception:
pass
self.pick_cam_np = None
self.pick_cam = None
self.pick_lens = None
pick_buffer = getattr(self, "pick_buffer", None)
if pick_buffer is not None:
try:
self.base.graphicsEngine.remove_window(pick_buffer)
except Exception:
try:
self.base.graphicsEngine.removeWindow(pick_buffer)
except Exception:
pass
self.pick_buffer = None
self.pick_texture = None
def _sync_pick_model_transform(self):
"""Sync pick-scene clone to current source model world transform."""
if not self.controller or not self.model:
return
pick_model = getattr(self.controller, "pick_model", None)
if pick_model is None:
return
try:
if pick_model.isEmpty():
return
except Exception:
try:
if pick_model.is_empty():
return
except Exception:
pass
try:
if hasattr(self.controller, "get_model_world_mat"):
world_mat = self.controller.get_model_world_mat()
else:
world_mat = LMatrix4f(self.model.getNetTransform().getMat())
try:
pick_model.set_mat(world_mat)
except Exception:
pick_model.setMat(world_mat)
except Exception:
pass
def _refresh_ssbo_proxy_center(self):
"""Update proxy center when source model transform changes."""
if self._ssbo_transform_active:
return
if not self.controller or not self._ssbo_selected_local_indices:
return
if self._ssbo_gizmo_proxy is None:
return
try:
if self._ssbo_gizmo_proxy.isEmpty():
return
except Exception:
return
try:
center = self.controller.get_selection_center(self._ssbo_selected_local_indices)
self._ssbo_gizmo_proxy.set_pos(center)
except Exception:
pass
def _is_model_attached(self):
"""Whether the SSBO render root is still attached to scene graph."""
if not self.model or self.model.is_empty():
return False
parent = self.model.get_parent()
return bool(parent) and not parent.is_empty()
def _sync_pick_scene_binding(self):
"""Switch pick camera scene based on current model attachment state."""
if not hasattr(self, "pick_cam") or not self.pick_cam:
return
if self._is_model_attached() and self.controller:
target_scene = getattr(self.controller, "pick_model", None) or self.model
else:
target_scene = self._empty_pick_scene
if not target_scene:
target_scene = self._empty_pick_scene
if self.pick_cam.get_scene() != target_scene:
self.pick_cam.set_scene(target_scene)
def pick_object(self, mx, my):
if (not self.pick_buffer or not self.pick_texture or not self.pick_lens or
not self.controller or not self.model):
return False
self._sync_pick_model_transform()
def _resolve_pick_selection_key(node_key):
resolved_key = node_key
try:
mouse_watcher = getattr(self.base, "mouseWatcherNode", None)
if mouse_watcher and mouse_watcher.is_button_down(KeyboardButton.control()):
parts = str(node_key).split("/")
if len(parts) >= 2 and parts[0] == getattr(self.controller, "tree_root_key", "0"):
resolved_key = "/".join(parts[:2])
except Exception:
pass
return resolved_key
try:
self.pick_lens.setAspectRatio(self.base.camLens.getAspectRatio())
except Exception:
pass
self.pick_lens.set_fov(0.1)
self.pick_lens.set_film_offset(0, 0)
self.pick_cam.set_lens(self.pick_lens)
near_point = Point3()
far_point = Point3()
self.base.camLens.extrude(Point2(mx, my), near_point, far_point)
self.pick_cam_np.set_pos(0, 0, 0)
self.pick_cam_np.look_at(far_point)
# Ensure pick transforms are up-to-date before rendering the pick buffer.
# The per-frame sync task may not have run yet for this frame.
self._sync_pick_transforms()
self.pick_buffer.set_active(True)
self.base.graphicsEngine.render_frame()
self.pick_buffer.set_active(False)
ram_image = self.pick_texture.get_ram_image_as("RGBA")
if ram_image:
data = memoryview(ram_image)
if len(data) >= 4:
r, g, b, a = data[0], data[1], data[2], data[3]
if a > 0 and b == 0:
hit_id = r + (g << 8)
node_key = self.controller.id_to_name.get(hit_id)
if node_key:
selection_key = _resolve_pick_selection_key(node_key)
print(f"[Pick] Hit: ID={hit_id} -> {node_key} (select={selection_key})")
self.select_node(selection_key)
return True
return False
def on_mouse_click(self):
# Skip SSBO picking when user is interacting with the TransformGizmo,
# otherwise pick_object would clear the selection and detach the gizmo
# before the gizmo's own mouse handler fires.
if self._transform_gizmo and self._transform_gizmo.is_hovering:
return
if self.base.mouseWatcherNode.has_mouse():
try:
win_width, win_height = self.base.win.getSize()
mpos = self.base.mouseWatcherNode.get_mouse()
window_x = (float(mpos.x) + 1.0) * 0.5 * float(win_width)
window_y = (1.0 - float(mpos.y)) * 0.5 * float(win_height)
process_imgui_click = getattr(self.base, "processImGuiMouseClick", None)
if callable(process_imgui_click) and process_imgui_click(window_x, window_y):
return
except Exception:
pass
self._sync_pick_model_transform()
self._refresh_ssbo_proxy_center()
mpos = self.base.mouseWatcherNode.get_mouse()
if self.pick_object(mpos.x, mpos.y):
return
# In SSBO picking mode, a miss should clear the current SSBO selection.
# Falling back to legacy collision picking here tends to immediately
# re-hit broad helper/collision shells from the selected root model,
# which makes "click blank space to deselect" fail for top-level nodes.
if self.has_active_selection():
self.clear_selection()
return
try:
win_width, win_height = self.base.win.getSize()
window_x = (float(mpos.x) + 1.0) * 0.5 * float(win_width)
window_y = (1.0 - float(mpos.y)) * 0.5 * float(win_height)
event_handler = getattr(self.base, "event_handler", None)
if event_handler and hasattr(event_handler, "mousePressEventLeft"):
event_handler.mousePressEventLeft({
"x": window_x,
"y": window_y,
})
return
except Exception as e:
print(f"[SSBOEditor] Legacy pick fallback failed: {e}")
self.clear_selection()
def toggle_debug(self):
self.debug_mode = not self.debug_mode
def bind_transform_gizmo(self, gizmo):
"""Bind a TransformGizmo so it follows SSBO selection."""
self._transform_gizmo = gizmo
def _start_pick_sync_task(self):
"""Start a per-frame task that syncs pick transforms for selected objects."""
self.base.task_mgr.remove("ssbo_pick_sync")
self.base.task_mgr.add(self._pick_sync_task, "ssbo_pick_sync")
def _stop_pick_sync_task(self):
"""Stop the per-frame pick sync task."""
self.base.task_mgr.remove("ssbo_pick_sync")
def _pick_sync_task(self, task):
"""Per-frame: keep pick model transforms in sync with render model."""
self._sync_pick_transforms()
return task.cont
def _reset_pick_sync_cache(self):
self._last_group_sync_mat = None
self._last_single_sync_gid = None
self._last_single_sync_mat = None
def _matrices_close(self, a, b, eps=1e-5):
"""Small helper for robust matrix change detection."""
for r in range(4):
ra = a.get_row(r)
rb = b.get_row(r)
if (abs(ra[0] - rb[0]) > eps or
abs(ra[1] - rb[1]) > eps or
abs(ra[2] - rb[2]) > eps or
abs(ra[3] - rb[3]) > eps):
return False
return True
def _sync_pick_root_transform(self):
"""
Keep pick root aligned with the render root transform.
This covers transforms applied to the whole imported model
(for example, moving box.glb from scene hierarchy).
"""
if not self.controller or not self.model or not self._is_model_attached():
return
pick_root = getattr(self.controller, "pick_model", None)
if not pick_root:
return
if self.model.is_empty() or pick_root.is_empty():
return
pick_root.set_mat(self.base.render, self.model.get_mat(self.base.render))
def _sync_pick_transforms(self):
"""Sync pick model transforms to match render model transforms."""
if not self.controller:
return
self._sync_pick_root_transform()
transform_ids = self.get_selection_transform_ids()
if not transform_ids:
return
# Group selection can contain thousands of objects.
# Only resync when proxy transform has changed.
proxy = getattr(self, "_group_proxy", None)
if proxy and not proxy.is_empty() and len(transform_ids) > 1:
proxy_world = proxy.get_mat(self.base.render)
if self._last_group_sync_mat and self._matrices_close(proxy_world, self._last_group_sync_mat):
return
self._last_group_sync_mat = LMatrix4f(proxy_world)
else:
self._last_group_sync_mat = None
if len(transform_ids) == 1:
gid = transform_ids[0]
obj_np = self.controller.id_to_object_np.get(gid)
pick_np = self.controller.id_to_pick_np.get(gid)
if not (obj_np and pick_np and not obj_np.is_empty() and not pick_np.is_empty()):
return
obj_world_mat = obj_np.get_mat(self.base.render)
if (
self._last_single_sync_gid == gid and
self._last_single_sync_mat and
self._matrices_close(obj_world_mat, self._last_single_sync_mat)
):
return
self._last_single_sync_gid = gid
self._last_single_sync_mat = LMatrix4f(obj_world_mat)
pick_world_mat = pick_np.get_mat(self.base.render)
if not self._matrices_close(obj_world_mat, pick_world_mat):
pick_np.set_mat(self.base.render, obj_world_mat)
chunk_id = self.controller.id_to_chunk.get(gid)
if chunk_id is not None and chunk_id in self.controller.chunks:
self.controller.chunks[chunk_id]["dirty"] = True
return
self._last_single_sync_gid = None
self._last_single_sync_mat = None
for gid in transform_ids:
obj_np = self.controller.id_to_object_np.get(gid)
pick_np = self.controller.id_to_pick_np.get(gid)
if obj_np and pick_np and not obj_np.is_empty() and not pick_np.is_empty():
obj_world_mat = obj_np.get_mat(self.base.render)
pick_world_mat = pick_np.get_mat(self.base.render)
if not self._matrices_close(obj_world_mat, pick_world_mat):
# Sync by world transform so this stays correct even when
# the model root itself has been moved in scene hierarchy.
pick_np.set_mat(self.base.render, obj_world_mat)
chunk_id = self.controller.id_to_chunk.get(gid)
if chunk_id is not None and chunk_id in self.controller.chunks:
self.controller.chunks[chunk_id]["dirty"] = True
def sync_scene_nodes_to_pick(self, nodes):
"""Sync transformed scene nodes to both pick data and visible static chunks."""
if not self.controller:
return
self._sync_pick_scene_binding()
self._sync_pick_root_transform()
valid_nodes = []
for node in nodes or []:
if not self._node_is_valid(node):
continue
duplicate = False
for existing in valid_nodes:
try:
if existing == node:
duplicate = True
break
except Exception:
pass
if not duplicate:
valid_nodes.append(node)
if not valid_nodes:
self._reset_pick_sync_cache()
return
affected_chunks = set()
for gid, obj_np in self.controller.id_to_object_np.items():
if not self._node_is_valid(obj_np):
continue
matched = False
for target in valid_nodes:
try:
if obj_np == target:
matched = True
break
except Exception:
pass
if not matched:
continue
is_attached = False
try:
is_attached = bool(obj_np.has_parent())
except Exception:
try:
is_attached = bool(obj_np.hasParent())
except Exception:
is_attached = False
is_visible = False
if is_attached:
try:
is_visible = not obj_np.is_hidden()
except Exception:
try:
is_visible = not obj_np.isHidden()
except Exception:
is_visible = True
pick_np = self.controller.id_to_pick_np.get(gid)
if pick_np and not pick_np.is_empty():
if is_visible:
try:
pick_np.show()
except Exception:
pass
try:
pick_np.set_mat(self.base.render, obj_np.get_mat(self.base.render))
except Exception:
try:
pick_np.setMat(self.base.render, obj_np.getMat(self.base.render))
except Exception:
pass
else:
try:
pick_np.hide()
except Exception:
pass
chunk_id = self.controller.id_to_chunk.get(gid)
if chunk_id is not None and chunk_id in self.controller.chunks:
self.controller.chunks[chunk_id]["dirty"] = True
affected_chunks.add(chunk_id)
for chunk_id in affected_chunks:
chunk = self.controller.chunks.get(chunk_id)
if not chunk or chunk.get("dynamic_enabled"):
continue
try:
self.controller._rebuild_static_chunk(chunk_id)
except Exception:
pass
self._reset_pick_sync_cache()
def _update_outline_for_selection(self):
if not self._outline_manager:
return
transform_ids = self.get_selection_transform_ids()
if not self.controller or not transform_ids:
self._outline_manager.clear()
return
is_root_selection = (
self.controller and
self.selected_name == getattr(self.controller, "tree_root_key", None)
)
if is_root_selection:
self._outline_manager.clear()
return
targets = []
target_limit = max(1, int(getattr(self._outline_manager, "max_targets", 64)))
for gid in transform_ids:
obj_np = self.controller.id_to_object_np.get(gid)
if obj_np and not obj_np.is_empty():
targets.append(obj_np)
if len(targets) >= target_limit:
break
self._outline_manager.set_targets(targets)
def _node_is_valid(self, node):
if not node:
return False
try:
return not node.is_empty()
except Exception:
try:
return not node.isEmpty()
except Exception:
return False
def has_active_selection(self):
return bool(self.controller and self.selected_name is not None)
def is_flat_selection_mode(self):
if not self.controller:
return False
return not bool(getattr(self.controller, "id_to_object_np", {}) or {})
def get_selection_transform_ids(self):
if not self.controller or self.selected_name is None:
return []
return list(getattr(self, "transform_ids", []) or [])
def _is_root_selection(self):
return bool(
self.controller and
self.selected_name == getattr(self.controller, "tree_root_key", None)
)
def _is_top_level_source_child_selection(self, key=None):
if not self.controller:
return False
selected_key = key if key is not None else self.selected_name
if not selected_key or selected_key == getattr(self.controller, "tree_root_key", None):
return False
root_key = getattr(self.controller, "tree_root_key", None)
root_node = self.controller.tree_nodes.get(root_key, {}) if root_key else {}
top_level_children = root_node.get("children", []) or []
return selected_key in top_level_children
def _is_displayed_top_level_selection(self, key=None):
if not self.controller:
return False
selected_key = key if key is not None else self.selected_name
root_key = getattr(self.controller, "tree_root_key", None)
if not selected_key or not root_key or selected_key == root_key:
return False
current_key = selected_key
visited = set()
while current_key and current_key not in visited:
visited.add(current_key)
node = self.controller.tree_nodes.get(current_key)
if not node:
return False
parent_key = node.get("parent")
if parent_key == root_key:
return True
if not parent_key:
return False
if not self.controller.should_hide_tree_node(parent_key):
return False
current_key = parent_key
return False
def _resolve_top_level_source_child_key(self, key=None):
"""Return the direct child key under tree_root that owns the selection key."""
if not self.controller:
return None
selected_key = key if key is not None else self.selected_name
root_key = getattr(self.controller, "tree_root_key", None)
if not selected_key or not root_key or selected_key == root_key:
return None
current_key = selected_key
visited = set()
while current_key and current_key not in visited:
visited.add(current_key)
node = self.controller.tree_nodes.get(current_key)
if not node:
return None
parent_key = node.get("parent")
if parent_key == root_key:
return current_key
current_key = parent_key
return None
def _selection_equivalent_top_level_key(self, key=None):
"""Normalize wrapper selections to their equivalent top-level model key."""
if not self.controller:
return None
selected_key = key if key is not None else self.selected_name
if not selected_key:
return None
top_key = self._resolve_top_level_source_child_key(selected_key)
if not top_key:
return None
if selected_key == top_key:
return top_key
try:
selected_ids = set(self.controller.name_to_ids.get(selected_key, []) or [])
top_ids = set(self.controller.name_to_ids.get(top_key, []) or [])
except Exception:
return None
if not selected_ids or not top_ids:
return None
if selected_ids == top_ids:
return top_key
return None
def _should_use_model_root_for_top_level_selection(self, key=None):
if not self.controller:
return False
selected_key = key if key is not None else self.selected_name
root_key = getattr(self.controller, "tree_root_key", None)
if not selected_key or not root_key or selected_key == root_key:
return False
if not (
self._is_top_level_source_child_selection(selected_key)
or self._is_displayed_top_level_selection(selected_key)
):
return False
try:
root_node = self.controller.tree_nodes.get(root_key, {}) if root_key else {}
top_level_children = list(root_node.get("children", []) or [])
return len(top_level_children) <= 1
except Exception:
return False
def get_selection_scene_node(self):
"""Return a stable scene node for editor features that need one."""
if not self.controller or self.selected_name is None:
return None
has_dynamic_objects = bool(getattr(self.controller, "id_to_object_np", {}) or {})
transform_ids = self.get_selection_transform_ids()
if self._is_root_selection():
return self.model if self._node_is_valid(self.model) else None
if len(transform_ids) > 1:
proxy = getattr(self, "_group_proxy", None)
if proxy and self._node_is_valid(proxy):
return proxy
if self._should_use_model_root_for_top_level_selection():
return self.model if self._node_is_valid(self.model) else None
fallback_node = None
try:
fallback_node = getattr(self.controller, "key_to_node", {}).get(self.selected_name)
except Exception:
fallback_node = None
if self._node_is_valid(fallback_node):
return fallback_node
if not has_dynamic_objects:
return self.model if self._node_is_valid(self.model) else None
if len(self.selected_ids) >= 1:
obj_np = self.controller.id_to_object_np.get(self.selected_ids[0])
if self._node_is_valid(obj_np):
return obj_np
return None
def get_selection_runtime_material_node(self):
"""Return the runtime node that material editing should target."""
if self.controller and self.selected_name is not None:
fallback_node = getattr(self.controller, "key_to_node", {}).get(self.selected_name)
if self._node_is_valid(fallback_node):
return fallback_node
if not self.controller or not self.selected_ids:
return None
obj_np = self.controller.id_to_object_np.get(self.selected_ids[0])
if self._node_is_valid(obj_np):
return obj_np
return None
def get_selection_source_material_node(self):
"""Return the exact source-tree node for the currently edited material target."""
if not self.controller or not self.selected_ids:
return None
owner_key = self.controller.id_to_name.get(self.selected_ids[0])
if not owner_key:
return None
source_node = self._resolve_source_node_by_tree_key(owner_key)
if self._node_is_valid(source_node):
return source_node
return None
def get_selection_summary(self):
if not self.controller or self.selected_name is None:
return None
return {
"key": self.selected_name,
"display_name": self.controller.display_names.get(self.selected_name, self.selected_name),
"object_count": len(self.get_selection_transform_ids()),
"is_root": self._is_root_selection(),
"is_group": len(self.get_selection_transform_ids()) > 1 and not self._is_root_selection(),
}
def estimate_selection_cost(self, key=None):
"""Estimate how expensive a selection would be in hybrid SSBO mode."""
if not self.controller:
return {
"key": key,
"object_count": 0,
"chunk_count": 0,
"is_root": False,
"is_top_level_like": False,
}
selected_key = key if key is not None else self.selected_name
if selected_key is None:
return {
"key": None,
"object_count": 0,
"chunk_count": 0,
"is_root": False,
"is_top_level_like": False,
}
transform_ids = list(getattr(self.controller, "name_to_ids", {}).get(selected_key, []) or [])
chunk_ids = set()
for transform_id in transform_ids:
chunk_id = getattr(self.controller, "id_to_chunk", {}).get(transform_id)
if isinstance(chunk_id, (tuple, list)):
chunk_id = chunk_id[0] if chunk_id else None
if chunk_id is not None:
chunk_ids.add(chunk_id)
return {
"key": selected_key,
"object_count": len(transform_ids),
"chunk_count": len(chunk_ids),
"is_root": bool(selected_key == getattr(self.controller, "tree_root_key", None)),
"is_top_level_like": bool(
self._is_top_level_source_child_selection(selected_key)
or self._is_displayed_top_level_selection(selected_key)
),
}
def get_selection_key(self):
if not self.controller or self.selected_name is None:
return None
return self.selected_name
def get_selection_source_node(self):
if not self.controller or self.selected_name is None:
return None
if self._should_use_model_root_for_top_level_selection():
source_children = self._get_source_root_children()
if len(source_children) == 1 and self._node_is_valid(source_children[0]):
return source_children[0]
source_node = self._resolve_source_node_by_tree_key(self.selected_name)
if self._node_is_valid(source_node):
return source_node
fallback_scene_node = self.get_selection_scene_node()
if self.is_source_tree_node(fallback_scene_node):
return fallback_scene_node
return None
def is_source_tree_node(self, node):
if not self._node_is_valid(node) or not self._node_is_valid(self.source_model_root):
return False
current = node
while self._node_is_valid(current):
if current == self.source_model_root:
return True
try:
current = current.get_parent()
except Exception:
try:
current = current.getParent()
except Exception:
return False
return False
def refresh_runtime_from_source(self, preserve_selection=True):
if not self._get_source_root_children():
return False
selected_key = self.selected_name if preserve_selection else None
try:
# Loading/rebuilding from an already authoritative source tree must not
# snapshot the transient runtime state back into source again.
# Otherwise scene-open can overwrite freshly restored transforms with
# stale/identity runtime values before the rebuild happens.
self._clear_runtime_state(preserve_source_models=True, snapshot_transform=False)
self._rebuild_runtime_from_source_root(highlight_root_name=None)
self.force_static_chunk_idle_state()
if preserve_selection and selected_key and self.controller:
try:
if (
selected_key == getattr(self.controller, "tree_root_key", None)
or selected_key in getattr(self.controller, "tree_nodes", {})
or selected_key in getattr(self.controller, "name_to_ids", {})
):
self.select_node(selected_key)
except Exception:
pass
return True
except Exception as e:
print(f"[SSBOEditor] 从 source 刷新 runtime 失败: {e}")
return False
def get_source_tree_stats(self):
root = getattr(self, "source_model_root", None)
stats = {
"valid": False,
"has_parent": False,
"hidden": False,
"stashed": False,
"descendants": 0,
"geom_nodes": 0,
"top_children": 0,
"parent_name": "",
}
if not self._node_is_valid(root):
return stats
stats["valid"] = True
try:
stats["has_parent"] = bool(root.has_parent())
except Exception:
pass
try:
stats["hidden"] = bool(root.is_hidden())
except Exception:
pass
try:
stats["stashed"] = bool(root.is_stashed())
except Exception:
pass
try:
parent = root.get_parent()
if parent and not parent.is_empty():
stats["parent_name"] = parent.get_name() or ""
except Exception:
pass
try:
stats["descendants"] = root.find_all_matches("**").get_num_paths()
except Exception:
pass
try:
stats["geom_nodes"] = len(list(root.find_all_matches("**/+GeomNode")))
except Exception:
pass
try:
stats["top_children"] = len([child for child in root.get_children() if not child.is_empty()])
except Exception:
pass
return stats
def force_static_chunk_idle_state(self):
"""Force hybrid chunk runtime back to fully static idle mode."""
controller = getattr(self, "controller", None)
chunks = getattr(controller, "chunks", None) if controller else None
if not controller or not isinstance(chunks, dict) or not chunks:
return False
rebuilt = 0
for chunk_id in sorted(chunks):
try:
controller._rebuild_static_chunk(chunk_id)
controller._set_chunk_dynamic(chunk_id, False)
rebuilt += 1
except Exception:
continue
if rebuilt:
print(f"[SSBOEditor] Forced static idle chunks: {rebuilt}")
return rebuilt > 0
def _sync_editor_selection_reference(self, node):
selection = getattr(self.base, "selection", None)
if not selection:
return
selection.selectedNode = node
selection.selectedObject = node
def _clear_editor_selection_visuals(self):
selection = getattr(self.base, "selection", None)
if not selection:
return
try:
selection.clearSelectionBox()
except Exception:
pass
try:
selection._updateSelectionOutline(None)
except Exception:
pass
try:
selection.clearGizmo()
except Exception:
pass
def _find_tree_key_for_scene_node(self, node):
if not self.controller or not self._node_is_valid(node):
return None
if self.is_source_tree_node(node):
source_key = self._resolve_tree_key_for_source_node(node)
if source_key:
return source_key
if self.model and node == self.model:
return getattr(self.controller, "tree_root_key", None)
try:
for key, mapped_node in getattr(self.controller, "key_to_node", {}).items():
if mapped_node == node:
return key
except Exception:
pass
for gid, obj_np in self.controller.id_to_object_np.items():
if obj_np == node:
return self.controller.id_to_name.get(gid)
return None
def sync_scene_selection(self, node):
"""Mirror scene-tree selection back into SSBO state, or clear stale SSBO state."""
if not self.controller:
return
target_key = self._find_tree_key_for_scene_node(node)
if target_key:
if target_key == self.selected_name:
return
self.select_node(target_key, sync_world_selection=False)
return
if self.has_active_selection():
self.clear_selection(sync_world_selection=False)
def clear_selection(self, sync_world_selection=True, snapshot_transform=True):
if snapshot_transform:
selected_top_synced = self._snapshot_selected_top_level_transform_to_source()
if not selected_top_synced:
self._snapshot_active_selection_transform_to_source()
self._stop_pick_sync_task()
self._reset_pick_sync_cache()
self._cleanup_group_proxy()
self.selected_name = None
self.selected_ids = []
self.transform_ids = []
if self._outline_manager:
self._outline_manager.clear()
if self.controller:
self.controller.set_active_ids([])
if self._transform_gizmo:
self._transform_gizmo.detach()
if sync_world_selection:
self._clear_editor_selection_visuals()
self._sync_editor_selection_reference(None)
def on_model_deleted(self, deleted_node):
"""Called by app deletion flow when SSBO root model is deleted."""
if not deleted_node or deleted_node.is_empty() or not self.model:
return
if deleted_node != self.model:
return
self.clear_selection()
self._sync_pick_scene_binding()
def reset_scene_state(self):
"""Remove the current SSBO model/controller state before loading another scene."""
self._clear_runtime_state(preserve_source_models=False)
def _cleanup_group_proxy(self):
"""Reparent objects back to their chunk and remove the group proxy."""
proxy = getattr(self, '_group_proxy', None)
if not proxy:
return
originals = getattr(self, '_group_original_parents', {})
# Sync pick transforms and mark chunks dirty before reparenting
for gid in originals:
obj_np = self.controller.id_to_object_np.get(gid)
pick_np = self.controller.id_to_pick_np.get(gid)
if obj_np and pick_np and not obj_np.is_empty() and not pick_np.is_empty():
pick_np.set_mat(self.base.render, obj_np.get_mat(self.base.render))
chunk_id = self.controller.id_to_chunk.get(gid)
if chunk_id is not None and chunk_id in self.controller.chunks:
self.controller.chunks[chunk_id]["dirty"] = True
# Reparent objects back to their original chunk parents
for gid, parent_np in originals.items():
obj_np = self.controller.id_to_object_np.get(gid)
if obj_np and not obj_np.is_empty() and parent_np and not parent_np.is_empty():
obj_np.wrt_reparent_to(parent_np)
if not proxy.is_empty():
proxy.remove_node()
self._group_proxy = None
self._group_original_parents = {}
self._group_proxy_initial_mat = None
self._group_proxy_source_initial_net_mat = None
def update_selection_mask(self):
pass # No selection mask texture needed without custom shader
def select_node(self, key, sync_world_selection=True):
selected_top_synced = self._snapshot_selected_top_level_transform_to_source()
if not selected_top_synced:
self._snapshot_active_selection_transform_to_source()
# Clean up previous group proxy before changing selection
self._cleanup_group_proxy()
self._reset_pick_sync_cache()
self.selected_name = key
preferred_ids_fn = getattr(self.controller, "get_preferred_selection_ids", None)
if callable(preferred_ids_fn):
self.selected_ids = preferred_ids_fn(key)
else:
self.selected_ids = self.controller.name_to_ids.get(key, [])
self.transform_ids = list(self.controller.name_to_ids.get(key, []) or [])
has_dynamic_objects = bool(getattr(self.controller, "id_to_object_np", {}) or {})
is_root_selection = (
self.controller and
key == getattr(self.controller, "tree_root_key", None)
)
is_displayed_top_level_selection = self._is_displayed_top_level_selection(key)
use_model_root_for_top_level_selection = self._should_use_model_root_for_top_level_selection(key)
if sync_world_selection:
self._clear_editor_selection_visuals()
# Root selection should stay lightweight, but top-level imported model
# groups still need a stable proxy so their local transform can be
# snapshotted back into source_root correctly.
if (
is_root_selection
or use_model_root_for_top_level_selection
or not has_dynamic_objects
):
self.controller.set_active_ids([])
if self._outline_manager and not (not has_dynamic_objects and not is_root_selection):
self._outline_manager.clear()
fallback_scene_node = self.get_selection_scene_node()
if sync_world_selection:
self._sync_editor_selection_reference(fallback_scene_node)
if self._outline_manager and not is_root_selection and self._node_is_valid(fallback_scene_node):
self._outline_manager.set_targets([fallback_scene_node])
if sync_world_selection and not has_dynamic_objects:
selection = getattr(self.base, "selection", None)
if selection and self._node_is_valid(fallback_scene_node):
try:
selection.updateSelection(fallback_scene_node)
except Exception:
pass
if self._transform_gizmo:
if is_root_selection and self.model and not self.model.is_empty():
self._transform_gizmo.attach(self.model)
elif self._node_is_valid(fallback_scene_node):
self._transform_gizmo.attach(fallback_scene_node)
else:
self._transform_gizmo.detach()
self._stop_pick_sync_task()
return
self.controller.set_active_ids(self.transform_ids)
self._update_outline_for_selection()
if not self._transform_gizmo or not self.transform_ids:
if self._transform_gizmo:
self._transform_gizmo.detach()
return
if len(self.transform_ids) == 1:
# Single object: attach gizmo directly
obj_np = self.controller.id_to_object_np.get(self.transform_ids[0])
if obj_np and not obj_np.is_empty():
if sync_world_selection:
self._sync_editor_selection_reference(obj_np)
self._transform_gizmo.attach(obj_np)
self._start_pick_sync_task()
return
# Multiple objects (parent node): create a group proxy so all children
# follow the gizmo transform together.
from panda3d.core import Vec3
proxy = self.base.render.attach_new_node("ssbo_group_proxy")
try:
proxy.set_name(self.controller.display_names.get(key, "ssbo_group_proxy"))
proxy.setTag("is_ssbo_proxy", "1")
proxy.setTag("ssbo_selection_key", str(key))
except Exception:
pass
center = Vec3(0, 0, 0)
valid = []
for gid in self.transform_ids:
obj_np = self.controller.id_to_object_np.get(gid)
if obj_np and not obj_np.is_empty():
center += obj_np.get_pos(self.base.render)
valid.append(gid)
if not valid:
proxy.remove_node()
return
center /= len(valid)
proxy.set_pos(self.base.render, center)
self._group_proxy = proxy
self._group_original_parents = {}
self._group_proxy_initial_mat = None
self._group_proxy_source_initial_net_mat = None
for gid in valid:
obj_np = self.controller.id_to_object_np[gid]
self._group_original_parents[gid] = obj_np.get_parent()
obj_np.wrt_reparent_to(proxy)
source_node = self.get_selection_source_node()
if self._node_is_valid(source_node):
try:
self._group_proxy_initial_mat = LMatrix4f(proxy.get_mat(self.model))
except Exception:
try:
self._group_proxy_initial_mat = LMatrix4f(proxy.getMat(self.model))
except Exception:
self._group_proxy_initial_mat = None
try:
self._group_proxy_source_initial_net_mat = LMatrix4f(
source_node.get_mat(self.source_model_root)
)
except Exception:
try:
self._group_proxy_source_initial_net_mat = LMatrix4f(
source_node.getMat(self.source_model_root)
)
except Exception:
self._group_proxy_source_initial_net_mat = None
if sync_world_selection:
self._sync_editor_selection_reference(proxy)
self._transform_gizmo.attach(proxy)
# For huge groups, avoid per-frame full sync; we still sync on demand
# right before picking via pick_object().
if len(valid) <= self._pick_sync_bg_limit:
self._start_pick_sync_task()
else:
self._stop_pick_sync_task()
def _rebuild_filtered_tree_rows(self):
"""
Build a flattened tree-row list with depth info for rendering in ImGui,
while preserving source-model parent/child hierarchy.
"""
self.filtered_nodes = []
if not self.controller or not self.controller.tree_root_key:
return
search_lower = self.search_text.strip().lower()
def walk(key, depth):
node = self.controller.tree_nodes.get(key)
if not node:
return False, []
# Skip redundant wrapper nodes (e.g. ROOT under model file root),
# while preserving child hierarchy and selection mapping.
if self.controller.should_hide_tree_node(key):
merged_rows = []
merged_visible = False
for child_key in node["children"]:
visible, rows = walk(child_key, depth)
if visible:
merged_visible = True
merged_rows.extend(rows)
return merged_visible, merged_rows
display = self.controller.display_names.get(key, key)
obj_count = len(self.controller.name_to_ids.get(key, []))
name_match = (not search_lower) or (search_lower in display.lower())
child_rows = []
child_match = False
for child_key in node["children"]:
visible, rows = walk(child_key, depth + 1)
if visible:
child_match = True
child_rows.extend(rows)
visible = (not search_lower) or name_match or child_match
if not visible:
return False, []
row = (key, depth, display, obj_count)
return True, [row] + child_rows
_, rows = walk(self.controller.tree_root_key, 0)
self.filtered_nodes = rows
def focus_on_selected(self):
transform_ids = self.get_selection_transform_ids()
if self.selected_name and transform_ids:
first_id = transform_ids[0]
pos = self.controller.get_world_pos(first_id)
dist = 100
self.base.camera.set_pos(pos.x, pos.y - dist, pos.z + dist * 0.5)
self.base.camera.look_at(pos)
def draw_imgui(self):
if not self.controller: return
imgui.set_next_window_pos((10, 10), imgui.Cond_.first_use_ever)
imgui.set_next_window_size((350, 600), imgui.Cond_.first_use_ever)
expanded, opened = imgui.begin("Scene Tree (Component)")
if expanded:
imgui.text(f"FPS: {globalClock.getAverageFrameRate():.1f}")
imgui.separator()
changed, self.search_text = imgui.input_text("Search", self.search_text, 256)
if imgui.begin_child("ObjectList", (0, 380), child_flags=imgui.ChildFlags_.borders):
if self.search_text != self.last_search_text or not self.filtered_nodes:
self.last_search_text = self.search_text
self._rebuild_filtered_tree_rows()
count = len(self.filtered_nodes)
clipper = imgui.ListClipper()
clipper.begin(count)
while clipper.step():
for i in range(clipper.display_start, clipper.display_end):
key, depth, display, geom_count = self.filtered_nodes[i]
indent = " " * depth
label = f"{indent}{display} ({geom_count})##{key}"
is_selected = (key == self.selected_name)
if imgui.selectable(label, is_selected)[0]:
self.select_node(key)
imgui.end_child()
imgui.separator()
if self.selected_name:
selected_display = self.controller.display_names.get(self.selected_name, self.selected_name)
imgui.text_colored((1, 0.8, 0.2, 1), f"Selected: {selected_display}")
if imgui.button("Focus (F)"): self.focus_on_selected()
imgui.end()
# swap_transforms_task removed - motion blur disabled for performance
def update_task(self, task):
dt = globalClock.getDt()
io = imgui.get_io()
self._sync_pick_model_transform()
self._refresh_ssbo_proxy_center()
self._sync_pick_scene_binding()
# Scene-hierarchy transforms may move the whole SSBO model root; keep pick root in sync.
self._sync_pick_root_transform()
if io.want_capture_keyboard: return task.cont
transform_ids = self.get_selection_transform_ids()
if transform_ids and self.controller:
speed = 50 * dt
acc = Vec3(0, 0, 0)
if self.keys.get('arrow_up'): acc.z += speed
if self.keys.get('arrow_down'): acc.z -= speed
if self.keys.get('arrow_left'): acc.x -= speed
if self.keys.get('arrow_right'): acc.x += speed
if self.keys.get('z'): acc.y += speed
if self.keys.get('x'): acc.y -= speed
if acc.length_squared() > 0:
is_root_selection = (
self.selected_name == getattr(self.controller, "tree_root_key", None)
)
if is_root_selection and self.model and not self.model.is_empty():
next_pos = self.model.get_pos() + acc
if hasattr(self.model, "set_fluid_pos"):
self.model.set_fluid_pos(next_pos)
else:
self.model.set_pos(next_pos)
self._sync_pick_root_transform()
else:
for idx in transform_ids:
self.controller.move_object(idx, acc)
return task.cont