#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ SUW Implementation - Python翻译版本 原文件: SUWImpl.rb (2019行) 用途: 核心实现类，SUWood的主要功能 翻译进度: Phase 1 - 几何类和基础框架 内存管理优化: 应用 Blender Python API 最佳实践 """ import re import math import logging import time import gc import weakref from typing import Optional, Any, Dict, List, Tuple, Union from contextlib import contextmanager # 设置日志 logger = logging.getLogger(__name__) # 尝试相对导入，失败则使用绝对导入 try: from .suw_constants import SUWood except ImportError: try: from suw_constants import SUWood except ImportError: # 如果都找不到，创建一个基本的存根 class SUWood: @staticmethod def suwood_path(version): return "." try: import bpy import mathutils import bmesh BLENDER_AVAILABLE = True except ImportError: BLENDER_AVAILABLE = False print("⚠️ Blender API 不可用，使用基础几何类") # 创建存根mathutils模块 class MockMathutils: class Vector: def __init__(self, vec): self.x, self.y, self.z = vec[:3] if len( vec) >= 3 else (vec + [0, 0])[:3] def normalized(self): return self def dot(self, other): return 0 class Matrix: @staticmethod def Scale(scale, size, axis): return MockMathutils.Matrix() @staticmethod def Translation(vec): return MockMathutils.Matrix() @staticmethod def Rotation(angle, size): return MockMathutils.Matrix() def __matmul__(self, other): return MockMathutils.Matrix() mathutils = MockMathutils() # ==================== 内存管理核心类 ==================== class BlenderMemoryManager: """Blender 内存管理器 - 应用最佳实践""" def __init__(self): self.object_registry = weakref.WeakSet() self.mesh_registry = weakref.WeakSet() self.material_registry = weakref.WeakSet() self.creation_stats = { 'objects_created': 0, 'meshes_created': 0, 'materials_created': 0, 'cleanup_calls': 0, 'failed_cleanups': 0 } self.last_cleanup_time = time.time() self.cleanup_interval = 300 # 5分钟 def register_object(self, obj): """注册对象到弱引用集合""" if obj: self.object_registry.add(obj) self.creation_stats['objects_created'] += 1 def register_mesh(self, mesh): """注册网格到弱引用集合""" if mesh: self.mesh_registry.add(mesh) self.creation_stats['meshes_created'] += 1 def should_cleanup(self): """检查是否应该执行清理""" return time.time() - self.last_cleanup_time > self.cleanup_interval def cleanup_orphaned_data(self): """清理孤立数据""" if not BLENDER_AVAILABLE: return 0 cleanup_count = 0 try: # 清理孤立网格 for mesh in list(bpy.data.meshes): if mesh.users == 0: try: bpy.data.meshes.remove(mesh, do_unlink=True) cleanup_count += 1 except: pass # 清理孤立材质 for material in list(bpy.data.materials): if material.users == 0: try: bpy.data.materials.remove(material, do_unlink=True) cleanup_count += 1 except: pass # 清理孤立纹理 for texture in list(bpy.data.textures): if texture.users == 0: try: bpy.data.textures.remove(texture, do_unlink=True) cleanup_count += 1 except: pass # 更新清理时间 self.last_cleanup_time = time.time() self.creation_stats['cleanup_calls'] += 1 logger.info(f"清理了 {cleanup_count} 个孤立数据块") return cleanup_count except Exception as e: logger.error(f"清理孤立数据失败: {e}") self.creation_stats['failed_cleanups'] += 1 return 0 # 全局内存管理器实例 memory_manager = BlenderMemoryManager() @contextmanager def safe_blender_operation(operation_name: str): """ 安全的 Blender 操作上下文管理器 使用方法： with safe_blender_operation("创建网格"): # 执行 Blender 操作 pass """ logger.info(f"开始操作: {operation_name}") start_time = time.time() # 记录初始状态 initial_objects = len(bpy.data.objects) if BLENDER_AVAILABLE else 0 initial_meshes = len(bpy.data.meshes) if BLENDER_AVAILABLE else 0 try: yield except Exception as e: logger.error(f"操作失败 {operation_name}: {e}") # 尝试清理 try: if memory_manager.should_cleanup(): memory_manager.cleanup_orphaned_data() gc.collect() except: pass raise finally: # 记录最终状态 final_objects = len(bpy.data.objects) if BLENDER_AVAILABLE else 0 final_meshes = len(bpy.data.meshes) if BLENDER_AVAILABLE else 0 elapsed_time = time.time() - start_time logger.info(f"操作完成: {operation_name}") logger.info(f"耗时: {elapsed_time:.2f}秒") logger.info(f"对象变化: {final_objects - initial_objects}") logger.info(f"网格变化: {final_meshes - initial_meshes}") # ==================== 几何类扩展 ==================== class Point3d: """3D点类 - 对应Ruby的Geom::Point3d""" def __init__(self, x: float = 0.0, y: float = 0.0, z: float = 0.0): self.x = x self.y = y self.z = z @classmethod def parse(cls, value: str): """从字符串解析3D点""" if not value or value.strip() == "": return None # 解析格式: "(x,y,z)" 或 "x,y,z" clean_value = re.sub(r'[()]*', '', value) xyz = [float(axis.strip()) for axis in clean_value.split(',')] # 转换mm为内部单位（假设输入是mm） return cls(xyz[0] * 0.001, xyz[1] * 0.001, xyz[2] * 0.001) def to_s(self, unit: str = "mm", digits: int = -1) -> str: """转换为字符串""" if unit == "cm": x_val = self.x * 100 # 内部单位转换为cm y_val = self.y * 100 z_val = self.z * 100 return f"({x_val:.3f}, {y_val:.3f}, {z_val:.3f})" else: # mm x_val = self.x * 1000 # 内部单位转换为mm y_val = self.y * 1000 z_val = self.z * 1000 if digits == -1: return f"({x_val}, {y_val}, {z_val})" else: return f"({x_val:.{digits}f}, {y_val:.{digits}f}, {z_val:.{digits}f})" def __str__(self): return self.to_s() def __repr__(self): return f"Point3d({self.x}, {self.y}, {self.z})" class Vector3d: """3D向量类 - 对应Ruby的Geom::Vector3d""" def __init__(self, x: float = 0.0, y: float = 0.0, z: float = 0.0): self.x = x self.y = y self.z = z @classmethod def parse(cls, value: str): """从字符串解析3D向量""" if not value or value.strip() == "": return None clean_value = re.sub(r'[()]*', '', value) xyz = [float(axis.strip()) for axis in clean_value.split(',')] return cls(xyz[0] * 0.001, xyz[1] * 0.001, xyz[2] * 0.001) def to_s(self, unit: str = "mm") -> str: """转换为字符串""" if unit == "cm": x_val = self.x * 100 # 内部单位转换为cm y_val = self.y * 100 z_val = self.z * 100 return f"({x_val:.3f}, {y_val:.3f}, {z_val:.3f})" elif unit == "in": return f"({self.x}, {self.y}, {self.z})" else: # mm x_val = self.x * 1000 # 内部单位转换为mm y_val = self.y * 1000 z_val = self.z * 1000 return f"({x_val}, {y_val}, {z_val})" def normalize(self): """归一化向量""" length = math.sqrt(self.x**2 + self.y**2 + self.z**2) if length > 0: return Vector3d(self.x/length, self.y/length, self.z/length) return Vector3d(0, 0, 0) def __str__(self): return self.to_s() class Transformation: """变换矩阵类 - 对应Ruby的Geom::Transformation""" def __init__(self, origin: Point3d = None, x_axis: Vector3d = None, y_axis: Vector3d = None, z_axis: Vector3d = None): self.origin = origin or Point3d(0, 0, 0) self.x_axis = x_axis or Vector3d(1, 0, 0) self.y_axis = y_axis or Vector3d(0, 1, 0) self.z_axis = z_axis or Vector3d(0, 0, 1) @classmethod def parse(cls, data: Dict[str, str]): """从字典解析变换""" origin = Point3d.parse(data.get("o")) x_axis = Vector3d.parse(data.get("x")) y_axis = Vector3d.parse(data.get("y")) z_axis = Vector3d.parse(data.get("z")) return cls(origin, x_axis, y_axis, z_axis) def store(self, data: Dict[str, str]): """存储变换到字典""" data["o"] = self.origin.to_s("mm") data["x"] = self.x_axis.to_s("in") data["y"] = self.y_axis.to_s("in") data["z"] = self.z_axis.to_s("in") # ==================== SUWood 材质类型常量 ==================== MAT_TYPE_NORMAL = 0 MAT_TYPE_OBVERSE = 1 MAT_TYPE_NATURE = 2 # ==================== SUWImpl 核心实现类 ==================== class SUWImpl: """SUWood核心实现类 - 完整翻译版本，应用内存管理最佳实践""" _instance = None _selected_uid = None _selected_obj = None _selected_zone = None _selected_part = None _scaled_zone = None _server_path = None _default_zone = None _creation_lock = False _mesh_creation_count = 0 _batch_operation_active = False def __init__(self): """初始化SUWImpl实例""" # 基础属性 self.parts = {} self.zones = {} self.hardwares = {} self.materials = {} self.textures = {} # 内存管理相关 self.object_references = {} # 存储对象名称而非引用 self.mesh_cache = {} self.material_cache = {} # 批量操作优化 self.deferred_updates = [] self.batch_size = 50 logger.info("SUWImpl 初始化完成，启用内存管理优化") @classmethod def get_instance(cls): """获取单例实例""" if cls._instance is None: cls._instance = cls() return cls._instance def startup(self): """启动SUWood系统""" logger.info("启动SUWood系统...") if BLENDER_AVAILABLE: try: self._create_layers() self._init_materials() self._init_default_zone() logger.info("✅ SUWood系统启动成功") except Exception as e: logger.error(f"❌ SUWood系统启动失败: {e}") raise else: logger.warning("⚠️ Blender不可用，使用基础模式") def _create_layers(self): """创建图层""" if not BLENDER_AVAILABLE: return try: # 创建必要的集合 collections = ["SUWood_Parts", "SUWood_Hardware", "SUWood_Zones"] for collection_name in collections: if collection_name not in bpy.data.collections: collection = bpy.data.collections.new(collection_name) bpy.context.scene.collection.children.link(collection) logger.info(f"创建集合: {collection_name}") except Exception as e: logger.error(f"创建图层失败: {e}") def _init_materials(self): """初始化材质""" if not BLENDER_AVAILABLE: return try: # 创建默认材质 default_materials = [ ("SUWood_Default", (0.8, 0.8, 0.8, 1.0)), ("SUWood_Wood", (0.6, 0.4, 0.2, 1.0)), ("SUWood_Metal", (0.7, 0.7, 0.8, 1.0)), ] for mat_name, color in default_materials: if mat_name not in bpy.data.materials: material = bpy.data.materials.new(mat_name) material.use_nodes = True if material.node_tree: principled = material.node_tree.nodes.get( "Principled BSDF") if principled: principled.inputs[0].default_value = color self.material_cache[mat_name] = material.name # 存储名称而非引用 memory_manager.register_object(material) logger.info(f"创建材质: {mat_name}") except Exception as e: logger.error(f"初始化材质失败: {e}") def _init_default_zone(self): """初始化默认区域""" try: self._default_zone = { "name": "Default Zone", "id": "default_zone_001", "visible": True, "parts": {}, "hardwares": {} } self.zones["default"] = self._default_zone logger.info("初始化默认区域") except Exception as e: logger.error(f"初始化默认区域失败: {e}") # ==================== 内存管理优化的核心方法 ==================== def validate_mesh_data(self, vertices: List, faces: List) -> bool: """验证网格数据有效性""" try: # 检查顶点数量 if len(vertices) < 3: logger.warning("顶点数量不足") return False # 检查面数据 for face in faces: if len(face) < 3: logger.warning("面顶点数量不足") return False # 检查顶点索引有效性 for vertex_index in face: if vertex_index >= len(vertices) or vertex_index < 0: logger.warning(f"无效的顶点索引: {vertex_index}") return False return True except Exception as e: logger.error(f"网格数据验证失败: {e}") return False def validate_coplanar_points(self, points: List[Tuple[float, float, float]]) -> bool: """验证点是否共面""" if len(points) < 4: return True # 3个点总是共面的 try: # 使用前三个点定义平面 p1 = mathutils.Vector(points[0]) p2 = mathutils.Vector(points[1]) p3 = mathutils.Vector(points[2]) # 计算平面法向量 normal = (p2 - p1).cross(p3 - p1).normalized() # 检查其他点是否在同一平面上 tolerance = 1e-6 for point in points[3:]: p = mathutils.Vector(point) distance = abs((p - p1).dot(normal)) if distance > tolerance: return False return True except Exception as e: logger.error(f"共面检查失败: {e}") return False @contextmanager def atomic_mesh_creation(self, name: str): """原子网格创建上下文管理器""" mesh = None obj = None try: # 第1步：创建网格数据 mesh = bpy.data.meshes.new(name) memory_manager.register_mesh(mesh) yield mesh except Exception as e: logger.error(f"原子网格创建失败: {e}") # 自动清理 if obj: self._cleanup_object_safe(obj) if mesh: self._cleanup_mesh_safe(mesh) raise def _cleanup_mesh_safe(self, mesh): """安全清理网格""" try: if mesh and mesh.name in bpy.data.meshes: bpy.data.meshes.remove(mesh, do_unlink=True) memory_manager.creation_stats['cleanup_calls'] += 1 logger.info(f"清理网格: {mesh.name}") except Exception as e: logger.error(f"网格清理失败: {e}") memory_manager.creation_stats['failed_cleanups'] += 1 def _cleanup_object_safe(self, obj): """安全清理对象""" try: if obj and obj.name in bpy.data.objects: # 从所有集合中移除 for collection in obj.users_collection: collection.objects.unlink(obj) # 删除对象 bpy.data.objects.remove(obj, do_unlink=True) memory_manager.creation_stats['cleanup_calls'] += 1 logger.info(f"清理对象: {obj.name}") except Exception as e: logger.error(f"对象清理失败: {e}") memory_manager.creation_stats['failed_cleanups'] += 1 def _create_single_mesh_atomic(self, container: Any, surface: Dict[str, Any], color: str = None, scale: float = None, angle: float = None, series: List = None, uid: str = None, zid: Any = None, pid: Any = None, child: Any = None, options: Dict = None) -> Any: """优化的原子性mesh创建""" with safe_blender_operation(f"创建网格 uid={uid}"): try: # 增加创建计数 self.__class__._mesh_creation_count += 1 creation_count = self.__class__._mesh_creation_count # 定期清理 if creation_count % 100 == 0: logger.info(f"执行定期清理，当前创建数: {creation_count}") memory_manager.cleanup_orphaned_data() gc.collect() # 数据验证 segs = surface.get("segs", []) points = [] for seg in segs: if isinstance(seg, list) and len(seg) >= 1: point_str = seg[0] if isinstance( seg[0], str) else str(seg[0]) try: point = Point3d.parse(point_str) if point: points.append(point) except Exception as e: logger.warning(f"点解析失败: {point_str}") if len(points) < 3: logger.warning(f"有效点数不足: {len(points)}") return None # 面积验证 area = self._calculate_face_area(points) if area < 1e-6: logger.warning(f"面积过小: {area}") return None # 验证几何数据 vertices = [(float(p.x), float(p.y), float(p.z)) for p in points] faces = [list(range(len(vertices)))] if len( vertices) >= 3 else [] if not self.validate_mesh_data(vertices, faces): logger.warning("网格数据验证失败") return None # 原子性创建 timestamp = hash( str(points[0]) + str(uid) + str(child) + str(creation_count)) mesh_name = f"SUWMesh_{abs(timestamp) % 10000}" with self.atomic_mesh_creation(mesh_name) as mesh: # 设置几何数据 mesh.from_pydata(vertices, [], faces) # 创建UV坐标 if len(vertices) >= 3: try: uv_layer = mesh.uv_layers.new(name="UVMap") self._generate_uv_coordinates(mesh, uv_layer) except Exception as e: logger.warning(f"UV坐标创建失败: {e}") # 安全更新网格 if not self._safe_mesh_update(mesh): raise RuntimeError("网格更新失败") # 创建对象 obj_name = f"SUWObj_{abs(timestamp) % 10000}" obj = bpy.data.objects.new(obj_name, mesh) memory_manager.register_object(obj) # 添加到集合 if hasattr(container, 'objects'): container.objects.link(obj) else: bpy.context.scene.collection.objects.link(obj) # 设置属性 self._set_object_attributes_safe( obj, uid, zid, pid, child, area, creation_count) # 应用材质 if color: self._apply_material_safe(obj, color) # 存储对象引用（名称而非对象） self.object_references[obj_name] = { 'uid': uid, 'type': 'mesh', 'creation_time': time.time() } logger.info(f"✅ 网格创建成功: {obj.name}, {len(vertices)}顶点") return obj except Exception as e: logger.error(f"网格创建失败: {e}") return None def _generate_uv_coordinates(self, mesh, uv_layer): """生成UV坐标""" try: for poly in mesh.polygons: for i, loop_index in enumerate(poly.loop_indices): vertex_count = len(poly.vertices) if vertex_count == 4: # 四边形 uv_coords = [(0, 0), (1, 0), (1, 1), (0, 1)] uv_layer.data[loop_index].uv = uv_coords[i % 4] elif vertex_count == 3: # 三角形 uv_coords = [(0, 0), (1, 0), (0.5, 1)] uv_layer.data[loop_index].uv = uv_coords[i % 3] else: # 其他多边形 angle = (i / vertex_count) * 2 * math.pi u = (math.cos(angle) + 1) * 0.5 v = (math.sin(angle) + 1) * 0.5 uv_layer.data[loop_index].uv = (u, v) except Exception as e: logger.error(f"UV坐标生成失败: {e}") def _safe_mesh_update(self, mesh, max_retries=3): """安全更新网格""" for attempt in range(max_retries): try: # 检查网格是否仍然有效 if not mesh or mesh.name not in bpy.data.meshes: return False # 更新网格 mesh.update() # 验证更新结果 if not mesh.vertices: raise RuntimeError("网格更新后无顶点") return True except Exception as e: logger.warning(f"网格更新失败，尝试 {attempt + 1}/{max_retries}: {e}") if attempt < max_retries - 1: time.sleep(0.1) # 短暂等待 else: return False return False def _set_object_attributes_safe(self, obj, uid, zid, pid, child, area, creation_count): """安全设置对象属性""" try: obj["uid"] = str(uid) if uid else "unknown" obj["zid"] = str(zid) if zid else "unknown" obj["pid"] = str(pid) if pid else "unknown" obj["child"] = str(child) if child else "unknown" obj["area"] = float(area) if area else 0.0 obj["creation_index"] = creation_count obj["creation_time"] = time.time() obj["memory_managed"] = True except Exception as e: logger.error(f"设置对象属性失败: {e}") def _apply_material_safe(self, obj, color): """安全应用材质""" try: material = self.get_texture(color) if material: if obj.data.materials: obj.data.materials[0] = material else: obj.data.materials.append(material) logger.info(f"材质应用成功: {color}") else: self._apply_simple_material(obj, color) except Exception as e: logger.error(f"材质应用失败: {e}") try: self._apply_simple_material(obj, color) except: pass def _apply_simple_material(self, obj: Any, color: str): """应用简单材质""" try: # 创建或获取缓存的材质 mat_name = f"SUWMat_{color}" if color else "SUWMat_default" if mat_name in self.material_cache: material_name = self.material_cache[mat_name] if material_name in bpy.data.materials: material = bpy.data.materials[material_name] obj.data.materials.append(material) return # 创建新材质 material = bpy.data.materials.new(mat_name) material.use_nodes = True if material.node_tree and material.node_tree.nodes: principled = material.node_tree.nodes.get("Principled BSDF") if principled: if color and len(color) >= 6: try: # 解析颜色 r = int(color[0:2], 16) / 255.0 g = int(color[2:4], 16) / 255.0 b = int(color[4:6], 16) / 255.0 principled.inputs[0].default_value = (r, g, b, 1.0) except: principled.inputs[0].default_value = ( 0.8, 0.8, 0.8, 1.0) else: principled.inputs[0].default_value = ( 0.8, 0.8, 0.8, 1.0) # 缓存材质 self.material_cache[mat_name] = material.name memory_manager.register_object(material) # 应用材质 obj.data.materials.append(material) logger.info(f"简单材质创建并应用: {mat_name}") except Exception as e: logger.error(f"简单材质应用失败: {e}") # ==================== 优化的删除和清理方法 ==================== def _force_delete_objects_optimized(self, objects_to_delete: List[Any]): """优化的强制删除对象""" try: if not BLENDER_AVAILABLE: return logger.info(f"开始优化删除 {len(objects_to_delete)} 个对象") # 预处理：收集对象名称 object_names = [] for obj in objects_to_delete: if obj and hasattr(obj, 'name'): object_names.append(obj.name) # 批量删除 deleted_count = 0 for obj_name in object_names: try: if obj_name in bpy.data.objects: obj = bpy.data.objects[obj_name] # 从集合中移除 collections_to_unlink = list(obj.users_collection) for collection in collections_to_unlink: try: collection.objects.unlink(obj) except: pass # 删除对象 bpy.data.objects.remove(obj, do_unlink=True) deleted_count += 1 # 从引用中移除 if obj_name in self.object_references: del self.object_references[obj_name] except Exception as e: logger.error(f"删除对象失败 {obj_name}: {e}") # 更新依赖图 if deleted_count > 0: self._update_dependency_graph(full_update=True) logger.info(f"✅ 优化删除完成: {deleted_count} 个对象") except Exception as e: logger.error(f"优化删除失败: {e}") def _update_dependency_graph(self, full_update: bool = False): """更新Blender依赖图""" try: if not BLENDER_AVAILABLE: return if full_update: logger.info("执行全局依赖图更新") bpy.context.view_layer.update() bpy.context.evaluated_depsgraph_get().update() # 刷新视图 try: for area in bpy.context.screen.areas: if area.type in ['VIEW_3D', 'OUTLINER']: area.tag_redraw() except: pass logger.info("全局依赖图更新完成") else: # 快速更新 bpy.context.view_layer.update() except Exception as e: logger.error(f"依赖图更新失败: {e}") # ==================== 保持原有业务逻辑的方法 ==================== def get_zones(self, data: Dict[str, Any]) -> Dict[str, Any]: """获取区域信息""" return self.zones def get_parts(self, data: Dict[str, Any]) -> Dict[str, Any]: """获取零件信息""" return self.parts def get_hardwares(self, data: Dict[str, Any]) -> Dict[str, Any]: """获取硬件信息""" return self.hardwares def get_texture(self, key: str): """获取纹理材质""" if not BLENDER_AVAILABLE: return None try: # 从缓存中获取 if key in self.material_cache: material_name = self.material_cache[key] if material_name in bpy.data.materials: return bpy.data.materials[material_name] # 查找现有材质 for material in bpy.data.materials: if key in material.name: self.material_cache[key] = material.name return material return None except Exception as e: logger.error(f"获取纹理失败: {e}") return None def create_face(self, container: Any, surface: Dict[str, Any], color: str = None, scale: float = None, angle: float = None, series: List = None, uid: str = None, zid: Any = None, pid: Any = None, child: Any = None) -> Any: """创建面 - 使用优化的内存管理""" try: return self._create_single_mesh_atomic( container, surface, color, scale, angle, series, uid, zid, pid, child ) except Exception as e: logger.error(f"创建面失败: {e}") return None def _calculate_face_area(self, points: List[Point3d]) -> float: """计算面积""" try: if len(points) < 3: return 0.0 # 使用三角剖分计算面积 area = 0.0 for i in range(1, len(points) - 1): p1 = points[0] p2 = points[i] p3 = points[i + 1] # 计算三角形面积 v1 = Vector3d(p2.x - p1.x, p2.y - p1.y, p2.z - p1.z) v2 = Vector3d(p3.x - p1.x, p3.y - p1.y, p3.z - p1.z) # 叉积的模长的一半就是三角形面积 cross_x = v1.y * v2.z - v1.z * v2.y cross_y = v1.z * v2.x - v1.x * v2.z cross_z = v1.x * v2.y - v1.y * v2.x triangle_area = 0.5 * \ math.sqrt(cross_x**2 + cross_y**2 + cross_z**2) area += triangle_area return area except Exception as e: logger.error(f"计算面积失败: {e}") return 0.0 # ==================== 系统管理方法 ==================== def get_creation_stats(self) -> Dict[str, Any]: """获取创建统计信息""" try: stats = { "creation_count": self.__class__._mesh_creation_count, "object_references": len(self.object_references), "mesh_cache_size": len(self.mesh_cache), "material_cache_size": len(self.material_cache), "memory_manager_stats": memory_manager.creation_stats.copy(), "blender_available": BLENDER_AVAILABLE } if BLENDER_AVAILABLE: stats["total_objects"] = len(bpy.data.objects) stats["total_meshes"] = len(bpy.data.meshes) stats["total_materials"] = len(bpy.data.materials) return stats except Exception as e: logger.error(f"获取统计信息失败: {e}") return {"error": str(e)} def force_cleanup(self): """强制清理""" try: logger.info("开始强制清理") # 清理孤立数据 cleanup_count = memory_manager.cleanup_orphaned_data() # 清理缓存 self.mesh_cache.clear() # 清理过期的对象引用 current_time = time.time() expired_refs = [] for obj_name, ref_info in self.object_references.items(): if current_time - ref_info.get('creation_time', 0) > 3600: # 1小时 expired_refs.append(obj_name) for obj_name in expired_refs: del self.object_references[obj_name] # 强制垃圾回收 gc.collect() # 更新依赖图 if BLENDER_AVAILABLE: self._update_dependency_graph(full_update=True) logger.info( f"强制清理完成: 清理了 {cleanup_count} 个数据块，{len(expired_refs)} 个过期引用") except Exception as e: logger.error(f"强制清理失败: {e}") def diagnose_system_state(self): """诊断系统状态""" try: logger.info("=== 系统状态诊断 ===") # 内存使用情况 stats = self.get_creation_stats() for key, value in stats.items(): logger.info(f"{key}: {value}") # 检查潜在问题 issues = [] if BLENDER_AVAILABLE: # 检查孤立数据 orphaned_meshes = [m for m in bpy.data.meshes if m.users == 0] if orphaned_meshes: issues.append(f"发现 {len(orphaned_meshes)} 个孤立网格") # 检查空网格 empty_meshes = [m for m in bpy.data.meshes if not m.vertices] if empty_meshes: issues.append(f"发现 {len(empty_meshes)} 个空网格") # 检查总顶点数 total_vertices = sum(len(m.vertices) for m in bpy.data.meshes) if total_vertices > 1000000: # 100万顶点 issues.append(f"顶点数量过多: {total_vertices}") if issues: logger.warning("发现问题:") for issue in issues: logger.warning(f" - {issue}") else: logger.info("✅ 系统状态正常") return issues except Exception as e: logger.error(f"系统诊断失败: {e}") return [f"诊断失败: {e}"] # ==================== 其他必要的方法存根 ==================== def c09(self, data: Dict[str, Any]): """删除操作 - 使用优化的内存管理""" try: uid = data.get("uid") typ = data.get("typ", "uid") oid = data.get("oid") logger.info(f"执行删除操作: uid={uid}, typ={typ}, oid={oid}") # 查找要删除的对象 objects_to_delete = self._find_objects_for_deletion(uid, typ, oid) if objects_to_delete: # 使用优化的删除方法 self._force_delete_objects_optimized(objects_to_delete) # 清理数据 self._safe_data_cleanup(uid, typ, oid) except Exception as e: logger.error(f"删除操作失败: {e}") def _find_objects_for_deletion(self, uid: str, typ: str, oid: Any) -> List[Any]: """查找要删除的对象""" try: if not BLENDER_AVAILABLE: return [] objects_to_delete = [] # 根据对象引用查找 for obj_name, ref_info in self.object_references.items(): if ref_info.get('uid') == uid: if obj_name in bpy.data.objects: objects_to_delete.append(bpy.data.objects[obj_name]) # 根据对象属性查找 for obj in bpy.data.objects: if obj.get("uid") == uid: objects_to_delete.append(obj) logger.info(f"找到 {len(objects_to_delete)} 个要删除的对象") return objects_to_delete except Exception as e: logger.error(f"查找删除对象失败: {e}") return [] def _safe_data_cleanup(self, uid: str, typ: str, oid: Any): """安全的数据清理""" try: cleanup_count = 0 # 清理内部数据结构 data_fields = [self.parts, self.zones, self.hardwares] for data_dict in data_fields: if uid in data_dict: del data_dict[uid] cleanup_count += 1 # 清理对象引用 refs_to_remove = [] for obj_name, ref_info in self.object_references.items(): if ref_info.get('uid') == uid: refs_to_remove.append(obj_name) for obj_name in refs_to_remove: del self.object_references[obj_name] cleanup_count += 1 logger.info(f"数据清理完成: {cleanup_count} 项") except Exception as e: logger.error(f"数据清理失败: {e}") # ==================== 其他业务逻辑方法的存根 ==================== def c00(self, data: Dict[str, Any]): """业务方法存根""" pass def c01(self, data: Dict[str, Any]): """创建区域""" try: uid = data.get("uid") zone_data = data.get("zone", {}) with safe_blender_operation(f"创建区域 uid={uid}"): # 创建区域集合 collection_name = f"Zone_{uid}" if BLENDER_AVAILABLE and collection_name not in bpy.data.collections: collection = bpy.data.collections.new(collection_name) bpy.context.scene.collection.children.link(collection) memory_manager.register_object(collection) # 存储区域数据 self.zones[uid] = { "collection_name": collection_name, "data": zone_data, "parts": {}, "creation_time": time.time() } logger.info(f"创建区域成功: {uid}") except Exception as e: logger.error(f"创建区域失败: {e}") def c02(self, data: Dict[str, Any]): """创建零件""" try: uid = data.get("uid") parts_data = data.get("parts", {}) with safe_blender_operation(f"创建零件 uid={uid}"): # 批量创建零件 created_parts = {} for part_id, part_data in parts_data.items(): try: # 创建零件集合 collection_name = f"Part_{uid}_{part_id}" if BLENDER_AVAILABLE and collection_name not in bpy.data.collections: collection = bpy.data.collections.new( collection_name) # 添加到对应的区域集合 if uid in self.zones: zone_collection_name = self.zones[uid]["collection_name"] if zone_collection_name in bpy.data.collections: parent_collection = bpy.data.collections[zone_collection_name] parent_collection.children.link(collection) else: bpy.context.scene.collection.children.link( collection) memory_manager.register_object(collection) created_parts[part_id] = { "collection_name": collection_name, "data": part_data, "creation_time": time.time() } except Exception as e: logger.error(f"创建零件失败 {part_id}: {e}") # 存储零件数据 self.parts[uid] = created_parts logger.info(f"创建零件成功: {uid}, {len(created_parts)} 个零件") except Exception as e: logger.error(f"创建零件操作失败: {e}") def c03(self, data: Dict[str, Any]): """创建硬件""" try: uid = data.get("uid") hardware_data = data.get("hardwares", {}) with safe_blender_operation(f"创建硬件 uid={uid}"): created_hardware = {} for hw_id, hw_data in hardware_data.items(): try: # 创建硬件集合 collection_name = f"Hardware_{uid}_{hw_id}" if BLENDER_AVAILABLE and collection_name not in bpy.data.collections: collection = bpy.data.collections.new( collection_name) bpy.context.scene.collection.children.link( collection) memory_manager.register_object(collection) created_hardware[hw_id] = { "collection_name": collection_name, "data": hw_data, "creation_time": time.time() } except Exception as e: logger.error(f"创建硬件失败 {hw_id}: {e}") # 存储硬件数据 self.hardwares[uid] = created_hardware logger.info(f"创建硬件成功: {uid}, {len(created_hardware)} 个硬件") except Exception as e: logger.error(f"创建硬件操作失败: {e}") def c04(self, data: Dict[str, Any]): """创建面和边 - 优化版本""" try: uid = data.get("uid") parts_data = data.get("parts", {}) with safe_blender_operation(f"创建面和边 uid={uid}"): # 批量创建优化 total_created = 0 for part_id, part_data in parts_data.items(): try: # 获取零件集合 collection = None if BLENDER_AVAILABLE: collection_name = f"Part_{uid}_{part_id}" if collection_name in bpy.data.collections: collection = bpy.data.collections[collection_name] # 创建面 surfaces = part_data.get("surfaces", []) for surface in surfaces: try: obj = self.create_face( collection or bpy.context.scene.collection, surface, surface.get("color"), surface.get("scale"), surface.get("angle"), uid=uid, pid=part_id ) if obj: total_created += 1 except Exception as e: logger.error(f"创建面失败: {e}") # 创建边 edges = part_data.get("edges", []) for edge in edges: try: self.create_edges( collection or bpy.context.scene.collection, [edge] ) total_created += 1 except Exception as e: logger.error(f"创建边失败: {e}") except Exception as e: logger.error(f"处理零件失败 {part_id}: {e}") logger.info(f"创建面和边完成: {total_created} 个对象") # 定期清理 if total_created > 50: memory_manager.cleanup_orphaned_data() except Exception as e: logger.error(f"创建面和边操作失败: {e}") def create_edges(self, container: Any, edges_data: List) -> List: """创建边""" try: if not BLENDER_AVAILABLE: return [] created_edges = [] for edge_data in edges_data: try: # 解析边的起点和终点 start_point = Point3d.parse( edge_data.get("start", "0,0,0")) end_point = Point3d.parse(edge_data.get("end", "0,0,0")) if start_point and end_point: edge_obj = self._create_line_edge( container, start_point, end_point) if edge_obj: created_edges.append(edge_obj) memory_manager.register_object(edge_obj) except Exception as e: logger.error(f"创建单条边失败: {e}") return created_edges except Exception as e: logger.error(f"创建边失败: {e}") return [] def _create_line_edge(self, container: Any, start: Point3d, end: Point3d) -> Any: """创建线边""" try: if not BLENDER_AVAILABLE: return None # 创建边网格 mesh_name = f"SUWEdge_{hash(str(start) + str(end)) % 10000}" mesh = bpy.data.meshes.new(mesh_name) memory_manager.register_mesh(mesh) # 设置边的顶点和边 vertices = [ (start.x, start.y, start.z), (end.x, end.y, end.z) ] edges = [(0, 1)] mesh.from_pydata(vertices, edges, []) mesh.update() # 创建对象 obj_name = f"SUWEdgeObj_{hash(str(start) + str(end)) % 10000}" obj = bpy.data.objects.new(obj_name, mesh) memory_manager.register_object(obj) # 添加到集合 if hasattr(container, 'objects'): container.objects.link(obj) else: bpy.context.scene.collection.objects.link(obj) return obj except Exception as e: logger.error(f"创建线边失败: {e}") return None # ==================== 选择和可见性管理 ==================== def sel_clear(self): """清除选择""" try: if BLENDER_AVAILABLE: bpy.ops.object.select_all(action='DESELECT') self._selected_uid = None self._selected_obj = None self._selected_zone = None self._selected_part = None logger.info("选择已清除") except Exception as e: logger.error(f"清除选择失败: {e}") def sel_local(self, obj: Any): """选择本地对象""" try: if BLENDER_AVAILABLE and obj: # 清除当前选择 bpy.ops.object.select_all(action='DESELECT') # 选择指定对象 if obj.name in bpy.data.objects: bpy.data.objects[obj.name].select_set(True) bpy.context.view_layer.objects.active = bpy.data.objects[obj.name] # 更新选择状态 self._selected_obj = obj.name # 存储名称而非引用 logger.info(f"选择对象: {obj.name}") except Exception as e: logger.error(f"选择对象失败: {e}") # ==================== 材质和纹理管理 ==================== def add_mat_rgb(self, mat_id: str, alpha: float, r: int, g: int, b: int): """添加RGB材质""" try: if not BLENDER_AVAILABLE: return # 检查材质是否已存在 if mat_id in self.material_cache: material_name = self.material_cache[mat_id] if material_name in bpy.data.materials: return bpy.data.materials[material_name] # 创建新材质 material = bpy.data.materials.new(mat_id) material.use_nodes = True # 设置颜色 if material.node_tree: principled = material.node_tree.nodes.get("Principled BSDF") if principled: color = (r/255.0, g/255.0, b/255.0, alpha) principled.inputs[0].default_value = color # 设置透明度 if alpha < 1.0: material.blend_method = 'BLEND' # Alpha input principled.inputs[21].default_value = alpha # 缓存材质 self.material_cache[mat_id] = material.name memory_manager.register_object(material) logger.info(f"创建RGB材质: {mat_id}") return material except Exception as e: logger.error(f"创建RGB材质失败: {e}") return None # ==================== 系统监控和维护 ==================== def monitor_memory_usage(self): """监控内存使用情况""" try: if not BLENDER_AVAILABLE: return # 检查是否需要清理 if memory_manager.should_cleanup(): logger.info("触发定期内存清理") cleanup_count = memory_manager.cleanup_orphaned_data() if cleanup_count > 0: logger.info(f"清理了 {cleanup_count} 个孤立数据") gc.collect() # 检查内存压力 total_objects = len(bpy.data.objects) total_meshes = len(bpy.data.meshes) if total_objects > 10000 or total_meshes > 5000: logger.warning( f"内存压力警告: {total_objects} 个对象, {total_meshes} 个网格") # 执行强制清理 self.force_cleanup() except Exception as e: logger.error(f"内存监控失败: {e}") def get_memory_report(self) -> Dict[str, Any]: """获取内存报告""" try: report = { "timestamp": time.time(), "creation_stats": self.get_creation_stats(), "system_diagnosis": self.diagnose_system_state(), "memory_manager": { "object_registry_size": len(memory_manager.object_registry), "mesh_registry_size": len(memory_manager.mesh_registry), "last_cleanup": memory_manager.last_cleanup_time, "cleanup_interval": memory_manager.cleanup_interval } } if BLENDER_AVAILABLE: report["blender_data"] = { "objects": len(bpy.data.objects), "meshes": len(bpy.data.meshes), "materials": len(bpy.data.materials), "textures": len(bpy.data.textures), "images": len(bpy.data.images) } return report except Exception as e: logger.error(f"生成内存报告失败: {e}") return {"error": str(e)} # ==================== 属性访问器 ==================== @property def selected_uid(self): return self._selected_uid @property def selected_zone(self): return self._selected_zone @property def selected_part(self): return self._selected_part @property def selected_obj(self): # 返回对象名称而非引用，避免悬空引用 return self._selected_obj @property def server_path(self): return self._server_path @property def default_zone(self): return self._default_zone # ==================== 兼容性方法存根 ==================== def c05(self, data: Dict[str, Any]): """业务方法存根 - 保持兼容性""" logger.info("执行c05操作") def c06(self, data: Dict[str, Any]): """业务方法存根 - 保持兼容性""" logger.info("执行c06操作") def c07(self, data: Dict[str, Any]): """业务方法存根 - 保持兼容性""" logger.info("执行c07操作") def c08(self, data: Dict[str, Any]): """业务方法存根 - 保持兼容性""" logger.info("执行c08操作") def c10(self, data: Dict[str, Any]): """设置门信息""" try: uid = data.get("uid") doors = data.get("drs", []) logger.info(f"设置门信息: uid={uid}, {len(doors)} 个门") # 存储门信息 if uid not in self.parts: self.parts[uid] = {} self.parts[uid]["doors"] = doors except Exception as e: logger.error(f"设置门信息失败: {e}") def show_message(self, data: Dict[str, Any]): """显示消息""" try: message = data.get("message", "") logger.info(f"显示消息: {message}") except Exception as e: logger.error(f"显示消息失败: {e}") def set_config(self, data: Dict[str, Any]): """设置配置""" try: config = data.get("config", {}) logger.info(f"设置配置: {len(config)} 个配置项") # 应用内存管理相关配置 if "memory_cleanup_interval" in config: memory_manager.cleanup_interval = config["memory_cleanup_interval"] if "batch_size" in config: self.batch_size = config["batch_size"] except Exception as e: logger.error(f"设置配置失败: {e}") # ==================== 清理和销毁 ==================== def shutdown(self): """关闭系统""" try: logger.info("开始关闭SUWood系统") # 执行最终清理 self.force_cleanup() # 清理所有缓存 self.mesh_cache.clear() self.material_cache.clear() self.object_references.clear() # 清理数据结构 self.parts.clear() self.zones.clear() self.hardwares.clear() logger.info("✅ SUWood系统关闭完成") except Exception as e: logger.error(f"关闭系统失败: {e}") def __del__(self): """析构函数""" try: self.shutdown() except: pass # ==================== 模块级别的便利函数 ==================== def create_suw_instance(): """创建SUW实例的便利函数""" return SUWImpl.get_instance() def cleanup_blender_memory(): """清理Blender内存的便利函数""" try: if BLENDER_AVAILABLE: cleanup_count = memory_manager.cleanup_orphaned_data() gc.collect() logger.info(f"清理了 {cleanup_count} 个孤立数据") return cleanup_count return 0 except Exception as e: logger.error(f"清理内存失败: {e}") return 0 def get_memory_usage_summary(): """获取内存使用摘要""" try: if BLENDER_AVAILABLE: return { "objects": len(bpy.data.objects), "meshes": len(bpy.data.meshes), "materials": len(bpy.data.materials), "memory_manager_stats": memory_manager.creation_stats.copy() } return {"blender_available": False} except Exception as e: logger.error(f"获取内存使用摘要失败: {e}") return {"error": str(e)} # ==================== 主程序入口 ==================== if __name__ == "__main__": # 测试代码 try: logger.info("开始测试SUWImpl内存管理") # 创建实例 suw = create_suw_instance() suw.startup() # 获取内存报告 report = suw.get_memory_report() logger.info(f"内存报告: {report}") # 执行诊断 issues = suw.diagnose_system_state() if not issues: logger.info("✅ 系统状态正常") # 清理测试 cleanup_count = cleanup_blender_memory() logger.info(f"清理测试完成: {cleanup_count}") except Exception as e: logger.error(f"测试失败: {e}") finally: logger.info("测试完成")