""" 高级骨骼动画系统 - 优化和压缩模块 提供动画压缩、LOD、缓存优化等功能 """ import math import json import os from typing import Dict, List, Optional, Tuple, Any, Callable from dataclasses import dataclass, field from enum import Enum from panda3d.core import * from direct.actor.Actor import Actor # 定义常量 OPTIMIZATION_SETTINGS = { 'compression_threshold': 0.001, 'lod_distance_thresholds': [10.0, 25.0, 50.0], 'lod_frame_rates': [1.0, 0.5, 0.25, 0.1], 'cache_max_size': 500, 'cache_cleanup_interval': 30.0 } # 枚举定义 class CompressionMethod(Enum): """压缩方法枚举""" KEYFRAME_REDUCTION = 0 QUANTIZATION = 1 DELTA_COMPRESSION = 2 RUN_LENGTH_ENCODING = 3 class LODStrategy(Enum): """LOD策略枚举""" DISTANCE_BASED = 0 PERFORMANCE_BASED = 1 HYBRID = 2 class CachePolicy(Enum): """缓存策略枚举""" LRU = 0 # 最近最少使用 LFU = 1 # 最不经常使用 FIFO = 2 # 先进先出 @dataclass class CompressedAnimationData: """压缩动画数据类""" name: str frame_rate: float duration: float bone_data: Dict[str, List[Tuple[float, Tuple[float, float, float, float, float, float, float]]]] # 时间, (位置+旋转+缩放) compression_ratio: float = 1.0 original_size: int = 0 compressed_size: int = 0 @dataclass class LODSettings: """LOD设置数据类""" strategy: LODStrategy distance_thresholds: List[float] frame_rates: List[float] enable_auto_lod: bool = True performance_threshold: float = 30.0 # FPS阈值 @dataclass class CacheEntry: """缓存条目数据类""" key: str data: Any timestamp: float access_count: int = 1 size: int = 0 class AnimationCompressor: """ 动画压缩系统 支持多种压缩算法 """ def __init__(self): """初始化动画压缩系统""" self.settings = { 'position_threshold': 0.001, 'rotation_threshold': 0.01, 'scale_threshold': 0.001, 'quantization_bits': 16, 'enable_quantization': True } print("动画压缩系统初始化完成") def compress_animation(self, actor: Actor, anim_name: str, method: CompressionMethod = CompressionMethod.KEYFRAME_REDUCTION) -> Optional[CompressedAnimationData]: """ 压缩指定动画 :param actor: Actor对象 :param anim_name: 动画名称 :param method: 压缩方法 :return: 压缩后的动画数据 """ if anim_name not in actor.getAnimNames(): print(f"错误: 动画 {anim_name} 不存在") return None print(f"开始压缩动画: {anim_name} (方法: {method.name})") # 获取动画控制 anim_control = actor.getAnimControl(anim_name) if not anim_control: print(f"错误: 无法获取动画控制 {anim_name}") return None # 获取动画数据 num_frames = anim_control.getNumFrames() frame_rate = anim_control.getFrameRate() duration = anim_control.getDuration() print(f" 动画信息: {num_frames} 帧, {frame_rate} FPS, {duration:.2f} 秒") # 提取骨骼动画数据 bone_data = self._extract_bone_animation_data(actor, anim_name, num_frames) # 根据方法进行压缩 if method == CompressionMethod.KEYFRAME_REDUCTION: compressed_bone_data = self._compress_keyframe_reduction(bone_data) elif method == CompressionMethod.QUANTIZATION: compressed_bone_data = self._compress_quantization(bone_data) elif method == CompressionMethod.DELTA_COMPRESSION: compressed_bone_data = self._compress_delta(bone_data) elif method == CompressionMethod.RUN_LENGTH_ENCODING: compressed_bone_data = self._compress_rle(bone_data) else: compressed_bone_data = bone_data # 计算压缩比 original_size = self._calculate_data_size(bone_data) compressed_size = self._calculate_data_size(compressed_bone_data) compression_ratio = original_size / compressed_size if compressed_size > 0 else 1.0 # 创建压缩数据 compressed_data = CompressedAnimationData( name=anim_name, frame_rate=frame_rate, duration=duration, bone_data=compressed_bone_data, compression_ratio=compression_ratio, original_size=original_size, compressed_size=compressed_size ) print(f" 压缩完成: 原始大小={original_size}, 压缩大小={compressed_size}, 压缩比={compression_ratio:.2f}:1") return compressed_data def _extract_bone_animation_data(self, actor: Actor, anim_name: str, num_frames: int) -> Dict[str, List]: """ 提取骨骼动画数据 :param actor: Actor对象 :param anim_name: 动画名称 :param num_frames: 帧数 :return: 骨骼动画数据 """ bone_data = {} # 获取所有骨骼 bundle = actor.getPartBundleDict().get('modelRoot') if not bundle: return bone_data # 遍历骨骼 for i in range(bundle.getNumChildren()): child = bundle.getChild(i) if hasattr(child, 'getName'): bone_name = child.getName() bone_data[bone_name] = self._extract_bone_frames(actor, bone_name, anim_name, num_frames) return bone_data def _extract_bone_frames(self, actor: Actor, bone_name: str, anim_name: str, num_frames: int) -> List: """ 提取单个骨骼的帧数据 :param actor: Actor对象 :param bone_name: 骨骼名称 :param anim_name: 动画名称 :param num_frames: 帧数 :return: 帧数据列表 """ frames = [] # 为每一帧提取数据 for frame in range(num_frames): # 设置到指定帧 actor.pose(anim_name, frame) # 获取骨骼变换 joint = actor.exposeJoint(None, "modelRoot", bone_name) if joint: pos = joint.getPos() hpr = joint.getHpr() scale = joint.getScale() # 存储为 (时间, (px, py, pz, rx, ry, rz, sx, sy, sz)) time = frame / actor.getFrameRate(anim_name) frame_data = ( time, (pos.x, pos.y, pos.z, hpr.x, hpr.y, hpr.z, scale.x, scale.y, scale.z) ) frames.append(frame_data) return frames def _compress_keyframe_reduction(self, bone_data: Dict[str, List]) -> Dict[str, List]: """ 关键帧简化压缩 :param bone_data: 骨骼数据 :return: 压缩后的骨骼数据 """ compressed_data = {} for bone_name, frames in bone_data.items(): if len(frames) <= 2: compressed_data[bone_name] = frames continue compressed_frames = [frames[0]] # 总是保留第一帧 last_frame = frames[0] for i in range(1, len(frames) - 1): current_frame = frames[i] next_frame = frames[i + 1] # 检查当前帧是否可以被简化 if not self._is_frame_necessary(last_frame, current_frame, next_frame): continue compressed_frames.append(current_frame) last_frame = current_frame compressed_frames.append(frames[-1]) # 总是保留最后一帧 compressed_data[bone_name] = compressed_frames return compressed_data def _is_frame_necessary(self, prev_frame: Tuple, current_frame: Tuple, next_frame: Tuple) -> bool: """ 判断帧是否必要 :param prev_frame: 前一帧 :param current_frame: 当前帧 :param next_frame: 后一帧 :return: 是否必要 """ prev_time, prev_data = prev_frame curr_time, curr_data = current_frame next_time, next_data = next_frame # 计算线性插值 t = (curr_time - prev_time) / (next_time - prev_time) if (next_time - prev_time) > 0 else 0 interpolated = tuple(prev_data[i] + t * (next_data[i] - prev_data[i]) for i in range(len(prev_data))) # 计算与实际值的差异 total_diff = sum(abs(curr_data[i] - interpolated[i]) for i in range(len(curr_data))) return total_diff > self.settings['position_threshold'] * 3 # 简化阈值检查 def _compress_quantization(self, bone_data: Dict[str, List]) -> Dict[str, List]: """ 量化压缩 :param bone_data: 骨骼数据 :return: 量化后的骨骼数据 """ # 量化实现 - 简化版本 return bone_data # 实际项目中需要实现具体的量化算法 def _compress_delta(self, bone_data: Dict[str, List]) -> Dict[str, List]: """ 差分压缩 :param bone_data: 骨骼数据 :return: 差分压缩后的骨骼数据 """ # 差分压缩实现 - 简化版本 return bone_data # 实际项目中需要实现具体的差分算法 def _compress_rle(self, bone_data: Dict[str, List]) -> Dict[str, List]: """ 游程编码压缩 :param bone_data: 骨骼数据 :return: RLE压缩后的骨骼数据 """ # RLE压缩实现 - 简化版本 return bone_data # 实际项目中需要实现具体的RLE算法 def _calculate_data_size(self, bone_data: Dict[str, List]) -> int: """ 计算数据大小 :param bone_data: 骨骼数据 :return: 数据大小（字节） """ size = 0 for frames in bone_data.values(): size += len(frames) * 100 # 简化估算，每帧约100字节 return size def decompress_animation(self, compressed_data: CompressedAnimationData) -> Dict[str, List]: """ 解压缩动画数据 :param compressed_data: 压缩数据 :return: 解压缩后的骨骼数据 """ # 解压缩实现 - 简化版本 return compressed_data.bone_data def get_compression_stats(self, compressed_data: CompressedAnimationData) -> Dict[str, Any]: """ 获取压缩统计信息 :param compressed_data: 压缩数据 :return: 统计信息 """ return { 'name': compressed_data.name, 'compression_ratio': compressed_data.compression_ratio, 'original_size': compressed_data.original_size, 'compressed_size': compressed_data.compressed_size, 'space_saved': compressed_data.original_size - compressed_data.compressed_size, 'percentage_saved': (1 - compressed_data.compressed_size / compressed_data.original_size) * 100 if compressed_data.original_size > 0 else 0 } class AnimationLODSystem: """ 动画LOD系统 根据距离和性能自动调整动画质量 """ def __init__(self, world): """ 初始化LOD系统 :param world: 世界对象 """ self.world = world self.settings = LODSettings( strategy=LODStrategy.DISTANCE_BASED, distance_thresholds=[10.0, 25.0, 50.0], frame_rates=[1.0, 0.5, 0.25, 0.1], enable_auto_lod=True, performance_threshold=30.0 ) self.actor_lod_levels = {} # {actor_id: lod_level} self.performance_monitor = { 'last_fps': 60.0, 'frame_times': [], 'last_update': 0.0 } self.camera_node = None print("动画LOD系统初始化完成") def set_lod_strategy(self, strategy: LODStrategy) -> None: """ 设置LOD策略 :param strategy: LOD策略 """ self.settings.strategy = strategy print(f"LOD策略设置为: {strategy.name}") def set_distance_thresholds(self, thresholds: List[float]) -> None: """ 设置距离阈值 :param thresholds: 距离阈值列表 """ self.settings.distance_thresholds = sorted(thresholds) print(f"距离阈值设置为: {thresholds}") def set_frame_rates(self, frame_rates: List[float]) -> None: """ 设置帧率级别 :param frame_rates: 帧率列表 """ self.settings.frame_rates = frame_rates print(f"帧率级别设置为: {frame_rates}") def enable_auto_lod(self, enabled: bool = True) -> None: """ 启用/禁用自动LOD :param enabled: 是否启用 """ self.settings.enable_auto_lod = enabled status = "启用" if enabled else "禁用" print(f"自动LOD已{status}") def set_performance_threshold(self, threshold: float) -> None: """ 设置性能阈值 :param threshold: FPS阈值 """ self.settings.performance_threshold = threshold print(f"性能阈值设置为: {threshold} FPS") def set_camera_node(self, camera_node) -> None: """ 设置摄像机节点 :param camera_node: 摄像机节点 """ self.camera_node = camera_node def update_lod_levels(self, actors: Dict[str, Any]) -> None: """ 更新所有Actor的LOD级别 :param actors: Actor字典 {actor_id: actor_info} """ if not self.settings.enable_auto_lod: return current_time = globalClock.getRealTime() # 每隔一定时间更新一次 if current_time - self.performance_monitor['last_update'] < 0.5: return self.performance_monitor['last_update'] = current_time # 获取摄像机位置 camera_pos = Point3(0, 0, 0) if self.camera_node: camera_pos = self.camera_node.getPos() elif hasattr(self.world, 'camera') and self.world.camera: camera_pos = self.world.camera.getPos() # 根据策略更新LOD if self.settings.strategy == LODStrategy.DISTANCE_BASED: self._update_distance_based_lod(actors, camera_pos) elif self.settings.strategy == LODStrategy.PERFORMANCE_BASED: self._update_performance_based_lod(actors) elif self.settings.strategy == LODStrategy.HYBRID: self._update_hybrid_lod(actors, camera_pos) def _update_distance_based_lod(self, actors: Dict[str, Any], camera_pos: Point3) -> None: """ 更新基于距离的LOD :param actors: Actor字典 :param camera_pos: 摄像机位置 """ for actor_id, actor_info in actors.items(): actor = actor_info.actor # 计算距离 distance = (actor.getPos() - camera_pos).length() # 确定LOD级别 lod_level = 0 for i, threshold in enumerate(self.settings.distance_thresholds): if distance > threshold: lod_level = i + 1 # 应用LOD设置 self._apply_lod_settings(actor_info, lod_level) self.actor_lod_levels[actor_id] = lod_level def _update_performance_based_lod(self, actors: Dict[str, Any]) -> None: """ 更新基于性能的LOD :param actors: Actor字典 """ # 根据FPS调整LOD fps = self.performance_monitor['last_fps'] if fps < self.settings.performance_threshold: # 降低所有Actor的LOD级别 for actor_id, actor_info in actors.items(): current_lod = self.actor_lod_levels.get(actor_id, 0) new_lod = min(current_lod + 1, len(self.settings.frame_rates) - 1) self._apply_lod_settings(actor_info, new_lod) self.actor_lod_levels[actor_id] = new_lod def _update_hybrid_lod(self, actors: Dict[str, Any], camera_pos: Point3) -> None: """ 更新混合LOD :param actors: Actor字典 :param camera_pos: 摄像机位置 """ # 结合距离和性能的LOD策略 self._update_distance_based_lod(actors, camera_pos) self._update_performance_based_lod(actors) def _apply_lod_settings(self, actor_info: Any, lod_level: int) -> None: """ 应用LOD设置到Actor :param actor_info: Actor信息 :param lod_level: LOD级别 """ actor = actor_info.actor frame_rate_multiplier = self.settings.frame_rates[lod_level] if lod_level < len(self.settings.frame_rates) else 0.1 # 调整所有动画的播放速度 for anim_name in actor.getAnimNames(): current_rate = actor.getPlayRate(anim_name) new_rate = current_rate * frame_rate_multiplier actor.setPlayRate(new_rate, anim_name) # 更新Actor信息中的LOD级别 actor_info.lod_level = lod_level def update_performance_stats(self, current_fps: float, frame_time: float) -> None: """ 更新性能统计 :param current_fps: 当前FPS :param frame_time: 帧时间 """ self.performance_monitor['last_fps'] = current_fps self.performance_monitor['frame_times'].append(frame_time) # 保持最近30帧的统计数据 if len(self.performance_monitor['frame_times']) > 30: self.performance_monitor['frame_times'] = self.performance_monitor['frame_times'][-30:] def get_lod_stats(self) -> Dict[str, Any]: """ 获取LOD统计信息 :return: 统计信息 """ lod_counts = {} for lod_level in self.actor_lod_levels.values(): if lod_level not in lod_counts: lod_counts[lod_level] = 0 lod_counts[lod_level] += 1 return { 'actor_lod_distribution': lod_counts, 'current_fps': self.performance_monitor['last_fps'], 'average_frame_time': sum(self.performance_monitor['frame_times']) / len(self.performance_monitor['frame_times']) if self.performance_monitor['frame_times'] else 0, 'lod_settings': { 'strategy': self.settings.strategy.name, 'distance_thresholds': self.settings.distance_thresholds, 'frame_rates': self.settings.frame_rates } } class AnimationCacheManager: """ 动画缓存管理器 提供高效的动画数据缓存 """ def __init__(self, max_size: int = 500): """ 初始化缓存管理器 :param max_size: 最大缓存大小 """ self.max_size = max_size self.cache: Dict[str, CacheEntry] = {} self.policy = CachePolicy.LRU self.cleanup_interval = 30.0 self.last_cleanup_time = 0.0 print(f"动画缓存管理器初始化完成 (最大大小: {max_size})") def set_cache_policy(self, policy: CachePolicy) -> None: """ 设置缓存策略 :param policy: 缓存策略 """ self.policy = policy print(f"缓存策略设置为: {policy.name}") def set_max_size(self, max_size: int) -> None: """ 设置最大缓存大小 :param max_size: 最大大小 """ self.max_size = max_size self._cleanup_cache() print(f"最大缓存大小设置为: {max_size}") def put(self, key: str, data: Any, size: int = 0) -> None: """ 放入缓存项 :param key: 键 :param data: 数据 :param size: 数据大小 """ current_time = globalClock.getRealTime() if key in self.cache: # 更新现有条目 entry = self.cache[key] entry.data = data entry.timestamp = current_time entry.access_count += 1 entry.size = size or entry.size else: # 添加新条目 entry = CacheEntry( key=key, data=data, timestamp=current_time, access_count=1, size=size ) self.cache[key] = entry # 检查是否需要清理缓存 self._check_cleanup() def get(self, key: str) -> Optional[Any]: """ 获取缓存项 :param key: 键 :return: 数据或None """ if key in self.cache: entry = self.cache[key] entry.access_count += 1 entry.timestamp = globalClock.getRealTime() return entry.data return None def remove(self, key: str) -> bool: """ 移除缓存项 :param key: 键 :return: 是否成功移除 """ if key in self.cache: del self.cache[key] return True return False def clear(self) -> None: """清空缓存""" self.cache.clear() print("动画缓存已清空") def _check_cleanup(self) -> None: """ 检查是否需要清理缓存 """ current_time = globalClock.getRealTime() if current_time - self.last_cleanup_time > self.cleanup_interval: self._cleanup_cache() self.last_cleanup_time = current_time def _cleanup_cache(self) -> None: """ 清理缓存 """ if len(self.cache) <= self.max_size: return # 根据策略选择要移除的条目 entries_to_remove = len(self.cache) - self.max_size if self.policy == CachePolicy.LRU: # 移除最近最少使用的条目 sorted_entries = sorted(self.cache.values(), key=lambda x: x.timestamp) elif self.policy == CachePolicy.LFU: # 移除最不经常使用的条目 sorted_entries = sorted(self.cache.values(), key=lambda x: x.access_count) else: # FIFO # 移除最早添加的条目 sorted_entries = sorted(self.cache.values(), key=lambda x: x.timestamp) # 移除条目 for i in range(min(entries_to_remove, len(sorted_entries))): entry = sorted_entries[i] del self.cache[entry.key] print(f"缓存清理完成，移除了 {entries_to_remove} 个条目") def get_cache_stats(self) -> Dict[str, Any]: """ 获取缓存统计信息 :return: 统计信息 """ total_size = sum(entry.size for entry in self.cache.values()) total_access = sum(entry.access_count for entry in self.cache.values()) avg_access = total_access / len(self.cache) if self.cache else 0 return { 'size': len(self.cache), 'max_size': self.max_size, 'total_size': total_size, 'hit_rate': 0, # 简化实现，实际项目中需要跟踪命中率 'average_access_count': avg_access, 'policy': self.policy.name } # 使用示例和测试代码 def example_compression_usage(actor: Actor): """ 动画压缩使用示例 """ print("=== 动画压缩使用示例 ===") compressor = AnimationCompressor() # 压缩所有动画 for anim_name in actor.getAnimNames(): print(f"压缩动画: {anim_name}") compressed_data = compressor.compress_animation( actor, anim_name, CompressionMethod.KEYFRAME_REDUCTION ) if compressed_data: stats = compressor.get_compression_stats(compressed_data) print(f" 压缩统计: {stats}") print("动画压缩示例完成") def example_lod_usage(world, actors: Dict[str, Any]): """ LOD系统使用示例 """ print("=== 动画LOD使用示例 ===") lod_system = AnimationLODSystem(world) # 设置LOD参数 lod_system.set_lod_strategy(LODStrategy.HYBRID) lod_system.set_distance_thresholds([15.0, 30.0, 60.0]) lod_system.set_frame_rates([1.0, 0.7, 0.4, 0.1]) # 更新LOD级别 lod_system.update_lod_levels(actors) # 更新性能统计 lod_system.update_performance_stats(45, 0.022) # 45 FPS, 22ms per frame # 获取LOD统计 stats = lod_system.get_lod_stats() print(f" LOD统计: {stats}") print("动画LOD示例完成") def example_cache_usage(): """ 缓存管理使用示例 """ print("=== 动画缓存管理使用示例 ===") cache_manager = AnimationCacheManager(max_size=100) # 设置缓存策略 cache_manager.set_cache_policy(CachePolicy.LRU) # 添加缓存项 dummy_data = {'frames': 100, 'bones': 20, 'data': 'animation_data'} cache_manager.put("actor_1_walk", dummy_data, size=1024) cache_manager.put("actor_1_run", dummy_data, size=2048) cache_manager.put("actor_2_idle", dummy_data, size=512) # 获取缓存项 data = cache_manager.get("actor_1_walk") if data: print(" 成功获取缓存数据") # 查看缓存统计 stats = cache_manager.get_cache_stats() print(f" 缓存统计: {stats}") print("动画缓存管理示例完成") if __name__ == "__main__": print("动画优化和压缩模块加载完成")