1068 lines
42 KiB
Python
1068 lines
42 KiB
Python
import time
|
||
from datetime import datetime
|
||
from panda3d.core import (
|
||
CollisionTraverser, CollisionHandlerQueue, CollisionNode,
|
||
CollisionRay, CollisionSphere, BitMask32, Point3, Vec3
|
||
)
|
||
from direct.task.TaskManagerGlobal import taskMgr
|
||
|
||
class CollisionManager:
|
||
"""统一的碰撞检测管理器"""
|
||
|
||
def __init__(self, world):
|
||
self.world = world
|
||
self.traverser = CollisionTraverser("main_traverser")
|
||
self.queue = CollisionHandlerQueue()
|
||
|
||
# 碰撞掩码定义
|
||
self.MASKS = {
|
||
# === 基础碰撞掩码定义 ===
|
||
# 每个掩码使用不同的位(bit)来标识，可以进行位运算组合
|
||
|
||
# 'TERRAIN': BitMask32.bit(0), # 地形/地面 - 通常用于地面碰撞检测
|
||
# 'UI_ELEMENT': BitMask32.bit(1), # UI元素 - 界面组件的碰撞检测
|
||
# 'CAMERA': BitMask32.bit(2), # 摄像机 - 相机的碰撞检测
|
||
'MODEL_COLLISION': BitMask32.bit(6), # 模型碰撞 - 通用模型间碰撞检测
|
||
}
|
||
|
||
# 碰撞体形状类型
|
||
self.COLLISION_SHAPES = {
|
||
'SPHERE': 'sphere',
|
||
'BOX': 'box',
|
||
'CAPSULE': 'capsule',
|
||
'PLANE': 'plane',
|
||
'POLYGON': 'polygon',
|
||
'MESH': 'mesh'
|
||
}
|
||
|
||
# 掩码组合定义
|
||
self.MASK_COMBINATIONS = {
|
||
}
|
||
|
||
# 碰撞分组管理
|
||
self.collision_groups = {}
|
||
self.group_enabled = {}
|
||
|
||
# 碰撞回调系统
|
||
self.collision_callbacks = {}
|
||
self.collision_filters = {}
|
||
|
||
# 空间分割系统（八叉树）
|
||
self.spatial_partitioning_enabled = False
|
||
self.octree = None
|
||
self.octree_max_depth = 6
|
||
self.octree_max_objects = 10
|
||
|
||
# 连续碰撞检测
|
||
self.ccd_enabled = False
|
||
self.ccd_threshold = 10.0 # 速度阈值
|
||
|
||
# 性能监控
|
||
self.performance_monitor = CollisionPerformanceMonitor()
|
||
|
||
# 模型间碰撞检测配置
|
||
self.model_collision_enabled = False
|
||
self.collision_detection_frequency = 0.1
|
||
self.collision_distance_threshold = 0
|
||
self.last_collision_check = 0
|
||
self.collision_history = []
|
||
self.max_collision_history = 100
|
||
|
||
# 碰撞状态跟踪 - 避免重复打印
|
||
self.active_collisions = set() # 当前活跃的碰撞对
|
||
|
||
# 模型间碰撞检测任务
|
||
self.collision_task = None
|
||
|
||
print("✅ 碰撞检测管理器初始化完成")
|
||
|
||
def enableModelCollisionDetection(self, enable=True, frequency=0.1, threshold=0.5):
|
||
"""启用/禁用模型间碰撞检测
|
||
|
||
Args:
|
||
enable: 是否启用碰撞检测
|
||
frequency: 检测频率（秒）
|
||
threshold: 碰撞距离阈值
|
||
"""
|
||
self.model_collision_enabled = enable
|
||
self.collision_detection_frequency = frequency
|
||
self.collision_distance_threshold = threshold
|
||
|
||
if enable:
|
||
print(f"✅ 启用模型间碰撞检测 - 频率: {frequency}s, 阈值: {threshold}")
|
||
self._startCollisionDetectionTask()
|
||
else:
|
||
print("❌ 禁用模型间碰撞检测")
|
||
self._stopCollisionDetectionTask()
|
||
|
||
def _startCollisionDetectionTask(self):
|
||
"""启动碰撞检测任务"""
|
||
if self.collision_task:
|
||
taskMgr.remove(self.collision_task)
|
||
|
||
def collisionDetectionTask(task):
|
||
try:
|
||
self.detectModelCollisions()
|
||
return task.again
|
||
except Exception as e:
|
||
print(f"❌ 碰撞检测任务异常: {str(e)}")
|
||
return task.again
|
||
|
||
self.collision_task = taskMgr.doMethodLater(
|
||
self.collision_detection_frequency,
|
||
collisionDetectionTask,
|
||
"modelCollisionDetection"
|
||
)
|
||
|
||
def _stopCollisionDetectionTask(self):
|
||
"""停止碰撞检测任务"""
|
||
if self.collision_task:
|
||
taskMgr.remove(self.collision_task)
|
||
self.collision_task = None
|
||
|
||
def detectModelCollisions(self, specific_models=None, log_results=True):
|
||
"""检测模型间碰撞"""
|
||
if not self.model_collision_enabled and specific_models is None:
|
||
return []
|
||
|
||
start_time = time.perf_counter()
|
||
current_time = datetime.now()
|
||
|
||
models_to_check = specific_models if specific_models else self.world.models
|
||
if len(models_to_check) < 2:
|
||
return []
|
||
|
||
collision_results = []
|
||
checked_pairs = set()
|
||
current_collision_pairs = set()
|
||
|
||
# 遍历所有模型对
|
||
for i, model_a in enumerate(models_to_check):
|
||
for j, model_b in enumerate(models_to_check[i+1:], i+1):
|
||
pair_key = tuple(sorted([id(model_a), id(model_b)]))
|
||
if pair_key in checked_pairs:
|
||
continue
|
||
checked_pairs.add(pair_key)
|
||
|
||
collision_info = self._checkModelPairCollision(model_a, model_b, current_time)
|
||
if collision_info:
|
||
collision_results.append(collision_info)
|
||
current_collision_pairs.add(pair_key)
|
||
|
||
# 只有新碰撞才打印日志
|
||
if log_results and pair_key not in self.active_collisions:
|
||
self._logCollisionInfo(collision_info)
|
||
print(f"🆕 新碰撞检测到！")
|
||
|
||
# 检测碰撞结束的情况
|
||
ended_collisions = self.active_collisions - current_collision_pairs
|
||
for pair_key in ended_collisions:
|
||
print(f"✅ 碰撞结束: 模型对 {pair_key}")
|
||
|
||
# 更新活跃碰撞状态
|
||
self.active_collisions = current_collision_pairs
|
||
|
||
# 记录性能和历史
|
||
detection_time = time.perf_counter() - start_time
|
||
self.performance_monitor.recordModelCollisionDetection(
|
||
detection_time, len(models_to_check), len(collision_results))
|
||
|
||
if collision_results:
|
||
self.collision_history.extend(collision_results)
|
||
if len(self.collision_history) > self.max_collision_history:
|
||
self.collision_history = self.collision_history[-self.max_collision_history:]
|
||
|
||
return collision_results
|
||
|
||
def _checkModelPairCollision(self, model_a, model_b, timestamp):
|
||
"""使用Panda3D内置碰撞系统检查两个模型是否碰撞"""
|
||
try:
|
||
# 检查模型是否有碰撞节点
|
||
collision_node_a = self._findCollisionNode(model_a)
|
||
collision_node_b = self._findCollisionNode(model_b)
|
||
|
||
if not collision_node_a or not collision_node_b:
|
||
return None
|
||
|
||
# 使用Panda3D的碰撞遍历器
|
||
temp_traverser = CollisionTraverser()
|
||
temp_queue = CollisionHandlerQueue()
|
||
|
||
# 设置碰撞检测
|
||
temp_traverser.addCollider(collision_node_a, temp_queue)
|
||
|
||
# 执行碰撞检测
|
||
temp_traverser.traverse(model_b)
|
||
|
||
if temp_queue.getNumEntries() > 0:
|
||
# 有碰撞，获取碰撞信息
|
||
entry = temp_queue.getEntry(0)
|
||
|
||
center_a = model_a.getPos(self.world.render)
|
||
center_b = model_b.getPos(self.world.render)
|
||
distance = (center_b - center_a).length()
|
||
|
||
return {
|
||
'timestamp': timestamp,
|
||
'model_a': {
|
||
'name': model_a.getName(),
|
||
'center': center_a,
|
||
'radius': 0, # 不再需要手动计算半径
|
||
'node': model_a
|
||
},
|
||
'model_b': {
|
||
'name': model_b.getName(),
|
||
'center': center_b,
|
||
'radius': 0,
|
||
'node': model_b
|
||
},
|
||
'collision_point': entry.getSurfacePoint(self.world.render),
|
||
'distance': distance,
|
||
'penetration_depth': 0, # Panda3D会自动处理
|
||
'collision_normal': entry.getSurfaceNormal(self.world.render)
|
||
}
|
||
|
||
return None
|
||
|
||
except Exception as e:
|
||
print(f"❌ 检测模型对碰撞失败 ({model_a.getName()} vs {model_b.getName()}): {str(e)}")
|
||
return None
|
||
|
||
def _findCollisionNode(self, model):
|
||
"""查找模型的碰撞节点"""
|
||
for child in model.getChildren():
|
||
if isinstance(child.node(), CollisionNode):
|
||
return child
|
||
return None
|
||
|
||
def _logCollisionInfo(self, collision_info):
|
||
"""输出碰撞信息日志"""
|
||
timestamp = collision_info['timestamp'].strftime('%H:%M:%S.%f')[:-3]
|
||
model_a = collision_info['model_a']
|
||
model_b = collision_info['model_b']
|
||
|
||
print(f"🔥 [{timestamp}] 检测到碰撞:")
|
||
print(f" 模型A: {model_a['name']} (中心: {model_a['center']}, 半径: {model_a['radius']:.2f})")
|
||
print(f" 模型B: {model_b['name']} (中心: {model_b['center']}, 半径: {model_b['radius']:.2f})")
|
||
print(f" 碰撞点: {collision_info['collision_point']}")
|
||
print(f" 中心距离: {collision_info['distance']:.3f}")
|
||
print(f" 穿透深度: {collision_info['penetration_depth']:.3f}")
|
||
print(f" 碰撞法向: {collision_info['collision_normal']}")
|
||
|
||
def getCollisionHistory(self, limit=None):
|
||
"""获取碰撞历史记录
|
||
|
||
Args:
|
||
limit: 返回记录数量限制
|
||
|
||
Returns:
|
||
list: 碰撞历史记录
|
||
"""
|
||
if limit:
|
||
return self.collision_history[-limit:]
|
||
return self.collision_history.copy()
|
||
|
||
def clearCollisionHistory(self):
|
||
"""清除碰撞历史记录"""
|
||
self.collision_history.clear()
|
||
print("✅ 碰撞历史记录已清除")
|
||
|
||
def getCollisionStatistics(self):
|
||
"""获取碰撞统计信息"""
|
||
if not self.collision_history:
|
||
return {"total_collisions": 0, "unique_pairs": 0, "most_frequent_pair": None}
|
||
|
||
# 统计碰撞对
|
||
collision_pairs = {}
|
||
for collision in self.collision_history:
|
||
pair_key = tuple(sorted([
|
||
collision['model_a']['name'],
|
||
collision['model_b']['name']
|
||
]))
|
||
collision_pairs[pair_key] = collision_pairs.get(pair_key, 0) + 1
|
||
|
||
most_frequent_pair = max(collision_pairs.items(), key=lambda x: x[1])
|
||
|
||
return {
|
||
"total_collisions": len(self.collision_history),
|
||
"unique_pairs": len(collision_pairs),
|
||
"most_frequent_pair": {
|
||
"models": most_frequent_pair[0],
|
||
"count": most_frequent_pair[1]
|
||
},
|
||
"collision_pairs": collision_pairs
|
||
}
|
||
|
||
def createCollisionShape(self, model, shape_type='auto', **kwargs):
|
||
"""为模型创建指定类型的碰撞体
|
||
|
||
Args:
|
||
model: 模型节点
|
||
shape_type: 碰撞体类型 ('auto', 'sphere', 'box', 'capsule', 'plane', 'polygon')
|
||
**kwargs: 形状特定参数
|
||
|
||
Returns:
|
||
CollisionSolid: 创建的碰撞体
|
||
"""
|
||
from panda3d.core import (
|
||
CollisionSphere, CollisionBox, CollisionCapsule,
|
||
CollisionPlane, CollisionPolygon, Plane, Vec3
|
||
)
|
||
|
||
# 获取考虑变换后的实际尺寸和中心
|
||
transformed_info = self._getTransformedModelInfo(model)
|
||
if not transformed_info:
|
||
# 默认小球体
|
||
return CollisionSphere(Point3(0, 0, 0), 1.0)
|
||
|
||
center = transformed_info['center']
|
||
radius = transformed_info['radius']
|
||
actual_size = transformed_info['size']
|
||
scale_factor = transformed_info['scale_factor']
|
||
|
||
# 自动选择最适合的形状
|
||
if shape_type == 'auto':
|
||
shape_type = self._determineOptimalShape(model, transformed_info)
|
||
|
||
if shape_type == 'sphere':
|
||
# 优化球形碰撞体
|
||
sphere_radius = kwargs.get('radius', radius)
|
||
|
||
# 支持位置偏移
|
||
pos_offset = kwargs.get('position_offset', Vec3(0, 0, 0))
|
||
sphere_center = Point3(center.x + pos_offset.x, center.y + pos_offset.y, center.z + pos_offset.z)
|
||
|
||
return CollisionSphere(sphere_center, sphere_radius)
|
||
|
||
elif shape_type == 'box':
|
||
# 优化盒型碰撞体 - 更精确的尺寸和位置控制（考虑缩放）
|
||
# 获取自定义尺寸，如果没有提供则使用变换后的实际尺寸
|
||
width = kwargs.get('width', actual_size.x)
|
||
length = kwargs.get('length', actual_size.y)
|
||
height = kwargs.get('height', actual_size.z)
|
||
|
||
# 支持位置偏移
|
||
pos_offset = kwargs.get('position_offset', Vec3(0, 0, 0))
|
||
box_center = Point3(center.x + pos_offset.x, center.y + pos_offset.y, center.z + pos_offset.z)
|
||
|
||
# 计算盒子的最小和最大点（基于偏移后的中心）
|
||
half_width = width / 2
|
||
half_length = length / 2
|
||
half_height = height / 2
|
||
|
||
min_point = Point3(box_center.x - half_width, box_center.y - half_length, box_center.z - half_height)
|
||
max_point = Point3(box_center.x + half_width, box_center.y + half_length, box_center.z + half_height)
|
||
return CollisionBox(min_point, max_point)
|
||
|
||
elif shape_type == 'capsule':
|
||
# 优化胶囊体碰撞 - 更合理的比例和位置控制（考虑缩放）
|
||
# 使用变换后的实际高度，或自定义高度
|
||
custom_height = kwargs.get('height', actual_size.z)
|
||
|
||
# 更合理的半径计算：基于变换后模型宽度的平均值
|
||
default_radius = min(actual_size.x, actual_size.y) / 2.5 # 稍微小于模型的宽度
|
||
custom_radius = kwargs.get('radius', min(default_radius, custom_height * 0.4))
|
||
|
||
# 支持位置偏移
|
||
pos_offset = kwargs.get('position_offset', Vec3(0, 0, 0))
|
||
capsule_center = Point3(center.x + pos_offset.x, center.y + pos_offset.y, center.z + pos_offset.z)
|
||
|
||
# 计算胶囊体的两个端点（确保半径不会超出高度）
|
||
effective_height = max(custom_height, custom_radius * 2.1) # 确保高度至少是半径的2倍多一点
|
||
point_a = Point3(capsule_center.x, capsule_center.y, capsule_center.z - effective_height/2 + custom_radius)
|
||
point_b = Point3(capsule_center.x, capsule_center.y, capsule_center.z + effective_height/2 - custom_radius)
|
||
return CollisionCapsule(point_a, point_b, custom_radius)
|
||
|
||
elif shape_type == 'plane':
|
||
# 优化平面碰撞 - 支持位置偏移和更灵活的法向量
|
||
normal = kwargs.get('normal', Vec3(0, 0, 1))
|
||
|
||
# 支持位置偏移
|
||
pos_offset = kwargs.get('position_offset', Vec3(0, 0, 0))
|
||
plane_point = kwargs.get('point', Point3(center.x + pos_offset.x, center.y + pos_offset.y, center.z + pos_offset.z))
|
||
|
||
# 标准化法向量
|
||
normal.normalize()
|
||
plane = Plane(normal, plane_point)
|
||
return CollisionPlane(plane)
|
||
|
||
elif shape_type == 'polygon':
|
||
# === 创建多边形碰撞体 ===
|
||
# CollisionPolygon特点：
|
||
# - 单个平面多边形，所有顶点必须共面
|
||
# - 适用于：地板、墙面、简单平台等平面区域
|
||
# - 性能较好，但只能表示平面形状
|
||
#
|
||
# Mesh碰撞体特点：
|
||
# - 由多个三角形组成的复杂3D形状
|
||
# - 适用于：复杂地形、建筑物、不规则物体
|
||
# - 性能较差，但可以表示任意复杂形状
|
||
|
||
vertices = kwargs.get('vertices', [])
|
||
if len(vertices) >= 3:
|
||
# 将顶点转换为Point3对象
|
||
# 要求：所有顶点必须在同一平面上
|
||
collision_poly = CollisionPolygon(*[Point3(*v) for v in vertices])
|
||
return collision_poly
|
||
else:
|
||
#print("⚠️ 多边形至少需要3个顶点，回退到球体")
|
||
return CollisionSphere(center, radius)
|
||
|
||
def _determineOptimalShape(self, model, transformed_info):
|
||
"""根据模型特征自动确定最适合的碰撞体形状"""
|
||
# 获取变换后的模型尺寸比例
|
||
size = transformed_info['size']
|
||
max_dim = max(size.x, size.y, size.z)
|
||
min_dim = min(size.x, size.y, size.z)
|
||
|
||
# 根据模型名称和比例判断
|
||
model_name = model.getName().lower()
|
||
|
||
# 角色类模型使用胶囊体
|
||
if any(keyword in model_name for keyword in ['character', 'human', 'player', 'npc']):
|
||
return 'capsule'
|
||
|
||
# 建筑、箱子类使用包围盒
|
||
if any(keyword in model_name for keyword in ['building', 'box', 'cube', 'wall', 'door']):
|
||
return 'box'
|
||
|
||
# 地面、平台使用平面
|
||
if any(keyword in model_name for keyword in ['ground', 'floor', 'platform', 'terrain']):
|
||
return 'plane'
|
||
|
||
# 根据尺寸比例判断
|
||
aspect_ratio = max_dim / min_dim if min_dim > 0 else 1
|
||
|
||
if aspect_ratio > 3: # 细长物体用胶囊体
|
||
return 'capsule'
|
||
elif aspect_ratio < 1.5: # 接近球形用球体
|
||
return 'sphere'
|
||
else: # 其他用包围盒
|
||
return 'box'
|
||
|
||
def _getTransformedModelInfo(self, model):
|
||
"""获取考虑变换后的模型信息
|
||
|
||
Args:
|
||
model: 模型节点
|
||
|
||
Returns:
|
||
dict: 包含变换后的尺寸、中心、半径等信息
|
||
"""
|
||
try:
|
||
# 获取原始包围盒
|
||
bounds = model.getBounds()
|
||
if bounds.isEmpty():
|
||
return None
|
||
|
||
# 获取模型的变换矩阵
|
||
transform = model.getTransform()
|
||
scale = model.getScale()
|
||
|
||
# 计算缩放因子（取三轴缩放的平均值）
|
||
scale_factor = (abs(scale.x) + abs(scale.y) + abs(scale.z)) / 3.0
|
||
|
||
# 获取原始尺寸
|
||
original_size = bounds.getMax() - bounds.getMin()
|
||
|
||
# 应用缩放到尺寸
|
||
actual_size = Vec3(
|
||
original_size.x * abs(scale.x),
|
||
original_size.y * abs(scale.y),
|
||
original_size.z * abs(scale.z)
|
||
)
|
||
|
||
# 获取变换后的中心点（在世界坐标系中）
|
||
original_center = bounds.getCenter()
|
||
if hasattr(model, 'getPos'):
|
||
# 模型在世界坐标系中的位置
|
||
world_center = model.getPos(model.getParent() if model.getParent() else model)
|
||
center = Point3(world_center.x, world_center.y, world_center.z)
|
||
else:
|
||
# 如果无法获取世界位置，使用原始中心
|
||
center = original_center
|
||
|
||
# 计算变换后的半径（考虑缩放）
|
||
original_radius = bounds.getRadius()
|
||
transformed_radius = original_radius * scale_factor
|
||
|
||
# 调试信息
|
||
print(f"🔍 模型 {model.getName()} 变换信息:")
|
||
print(f" 原始尺寸: {original_size}")
|
||
print(f" 缩放因子: {scale}")
|
||
print(f" 变换后尺寸: {actual_size}")
|
||
print(f" 变换后半径: {transformed_radius:.2f}")
|
||
|
||
return {
|
||
'center': center,
|
||
'radius': transformed_radius,
|
||
'size': actual_size,
|
||
'scale_factor': scale_factor,
|
||
'original_size': original_size,
|
||
'scale': scale,
|
||
'transform': transform
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"⚠️ 获取模型变换信息失败: {e}")
|
||
# 回退到原始包围盒
|
||
bounds = model.getBounds()
|
||
if bounds.isEmpty():
|
||
return None
|
||
|
||
original_size = bounds.getMax() - bounds.getMin()
|
||
return {
|
||
'center': bounds.getCenter(),
|
||
'radius': bounds.getRadius(),
|
||
'size': original_size,
|
||
'scale_factor': 1.0,
|
||
'original_size': original_size,
|
||
'scale': Vec3(1, 1, 1),
|
||
'transform': None
|
||
}
|
||
|
||
def createMouseRay(self, screen_x, screen_y, mask_types=['SELECTABLE']):
|
||
"""创建鼠标射线"""
|
||
# 组合掩码
|
||
combined_mask = BitMask32.allOff()
|
||
for mask_type in mask_types:
|
||
if mask_type in self.MASKS:
|
||
combined_mask |= self.MASKS[mask_type]
|
||
|
||
# 坐标转换
|
||
near_point, far_point = self.world.event_handler.robustCoordinateConversion(
|
||
screen_x, screen_y)
|
||
|
||
if not near_point:
|
||
return None
|
||
|
||
# 创建射线节点
|
||
ray_node = CollisionNode('mouse_ray')
|
||
ray_node.setFromCollideMask(combined_mask)
|
||
ray_node.setIntoCollideMask(BitMask32.allOff())
|
||
|
||
# 创建射线
|
||
direction = far_point - near_point
|
||
direction.normalize()
|
||
ray = CollisionRay(near_point, direction)
|
||
ray_node.addSolid(ray)
|
||
|
||
return ray_node, near_point, far_point
|
||
|
||
def performRaycast(self, ray_node, scene_root=None):
|
||
"""执行射线检测"""
|
||
if scene_root is None:
|
||
scene_root = self.world.render
|
||
|
||
# 创建临时节点路径
|
||
ray_np = self.world.cam.attachNewNode(ray_node)
|
||
|
||
try:
|
||
# 清除之前的结果
|
||
self.queue.clearEntries()
|
||
|
||
# 添加碰撞器
|
||
self.traverser.addCollider(ray_np, self.queue)
|
||
|
||
# 执行遍历
|
||
start_time = time.perf_counter()
|
||
self.traverser.traverse(scene_root)
|
||
detection_time = time.perf_counter() - start_time
|
||
|
||
# 记录性能
|
||
self.performance_monitor.recordDetection(
|
||
detection_time, self.queue.getNumEntries())
|
||
|
||
# 处理结果
|
||
results = []
|
||
if self.queue.getNumEntries() > 0:
|
||
self.queue.sortEntries()
|
||
|
||
for i in range(self.queue.getNumEntries()):
|
||
entry = self.queue.getEntry(i)
|
||
results.append({
|
||
'hit_pos': entry.getSurfacePoint(scene_root),
|
||
'hit_normal': entry.getSurfaceNormal(scene_root),
|
||
'hit_node': entry.getIntoNodePath(),
|
||
'distance': entry.getT()
|
||
})
|
||
|
||
return results
|
||
|
||
finally:
|
||
# 清理资源
|
||
self.traverser.removeCollider(ray_np)
|
||
ray_np.removeNode()
|
||
|
||
def getCollisionMask(self, mask_type):
|
||
"""获取碰撞掩码"""
|
||
return self.MASKS.get(mask_type, BitMask32.allOff())
|
||
|
||
def createCollisionGroup(self, group_name, mask_types, enabled=True):
|
||
"""创建碰撞分组
|
||
|
||
Args:
|
||
group_name: 分组名称
|
||
mask_types: 掩码类型列表
|
||
enabled: 是否启用
|
||
"""
|
||
combined_mask = BitMask32.allOff()
|
||
for mask_type in mask_types:
|
||
if mask_type in self.MASKS:
|
||
combined_mask |= self.MASKS[mask_type]
|
||
|
||
self.collision_groups[group_name] = {
|
||
'mask': combined_mask,
|
||
'types': mask_types,
|
||
'objects': []
|
||
}
|
||
self.group_enabled[group_name] = enabled
|
||
|
||
print(f"✅ 创建碰撞分组: {group_name}, 掩码: {mask_types}")
|
||
|
||
def addToCollisionGroup(self, group_name, model, collision_node):
|
||
"""将模型添加到碰撞分组"""
|
||
if group_name in self.collision_groups:
|
||
self.collision_groups[group_name]['objects'].append({
|
||
'model': model,
|
||
'collision_node': collision_node
|
||
})
|
||
|
||
def enableCollisionGroup(self, group_name, enabled=True):
|
||
"""启用/禁用碰撞分组"""
|
||
if group_name in self.group_enabled:
|
||
self.group_enabled[group_name] = enabled
|
||
|
||
# 更新分组中所有对象的碰撞状态
|
||
if group_name in self.collision_groups:
|
||
for obj in self.collision_groups[group_name]['objects']:
|
||
collision_node = obj['collision_node']
|
||
if enabled:
|
||
collision_node.show()
|
||
else:
|
||
collision_node.hide()
|
||
|
||
print(f"{'✅ 启用' if enabled else '❌ 禁用'}碰撞分组: {group_name}")
|
||
|
||
def registerCollisionCallback(self, mask_a, mask_b, callback_func, filter_func=None):
|
||
"""注册碰撞回调函数
|
||
|
||
Args:
|
||
mask_a: 第一个掩码类型
|
||
mask_b: 第二个掩码类型
|
||
callback_func: 碰撞回调函数 callback(collision_info)
|
||
filter_func: 过滤函数 filter(model_a, model_b) -> bool
|
||
"""
|
||
key = tuple(sorted([mask_a, mask_b]))
|
||
if key not in self.collision_callbacks:
|
||
self.collision_callbacks[key] = []
|
||
|
||
self.collision_callbacks[key].append(callback_func)
|
||
|
||
if filter_func:
|
||
if key not in self.collision_filters:
|
||
self.collision_filters[key] = []
|
||
self.collision_filters[key].append(filter_func)
|
||
|
||
print(f"✅ 注册碰撞回调: {mask_a} <-> {mask_b}")
|
||
|
||
def _executeCollisionCallbacks(self, collision_info):
|
||
"""执行碰撞回调"""
|
||
model_a = collision_info['model_a']['node']
|
||
model_b = collision_info['model_b']['node']
|
||
|
||
# 获取模型的掩码类型
|
||
mask_a = self._getModelMaskType(model_a)
|
||
mask_b = self._getModelMaskType(model_b)
|
||
|
||
if mask_a and mask_b:
|
||
key = tuple(sorted([mask_a, mask_b]))
|
||
|
||
# 检查过滤条件
|
||
if key in self.collision_filters:
|
||
for filter_func in self.collision_filters[key]:
|
||
if not filter_func(model_a, model_b):
|
||
return
|
||
|
||
# 执行回调
|
||
if key in self.collision_callbacks:
|
||
for callback_func in self.collision_callbacks[key]:
|
||
try:
|
||
callback_func(collision_info)
|
||
except Exception as e:
|
||
print(f"❌ 碰撞回调执行失败: {e}")
|
||
|
||
def _getModelMaskType(self, model):
|
||
"""获取模型的掩码类型"""
|
||
# 从模型标签或属性中获取掩码类型
|
||
if model.hasTag('collision_mask_type'):
|
||
return model.getTag('collision_mask_type')
|
||
return None
|
||
|
||
def enableSpatialPartitioning(self, enabled=True, max_depth=6, max_objects=10):
|
||
"""启用空间分割优化
|
||
|
||
Args:
|
||
enabled: 是否启用
|
||
max_depth: 八叉树最大深度
|
||
max_objects: 每个节点最大对象数
|
||
"""
|
||
self.spatial_partitioning_enabled = enabled
|
||
self.octree_max_depth = max_depth
|
||
self.octree_max_objects = max_objects
|
||
|
||
if enabled:
|
||
self._buildOctree()
|
||
print(f"✅ 启用空间分割优化 - 深度: {max_depth}, 对象数: {max_objects}")
|
||
else:
|
||
self.octree = None
|
||
print("❌ 禁用空间分割优化")
|
||
|
||
def _buildOctree(self):
|
||
"""构建八叉树"""
|
||
if not hasattr(self.world, 'models') or not self.world.models:
|
||
return
|
||
|
||
# 计算场景边界
|
||
min_bound = Point3(float('inf'), float('inf'), float('inf'))
|
||
max_bound = Point3(float('-inf'), float('-inf'), float('-inf'))
|
||
|
||
for model in self.world.models:
|
||
bounds = model.getBounds()
|
||
if not bounds.isEmpty():
|
||
model_min = bounds.getMin()
|
||
model_max = bounds.getMax()
|
||
|
||
min_bound.x = min(min_bound.x, model_min.x)
|
||
min_bound.y = min(min_bound.y, model_min.y)
|
||
min_bound.z = min(min_bound.z, model_min.z)
|
||
|
||
max_bound.x = max(max_bound.x, model_max.x)
|
||
max_bound.y = max(max_bound.y, model_max.y)
|
||
max_bound.z = max(max_bound.z, model_max.z)
|
||
|
||
# 创建八叉树根节点
|
||
self.octree = OctreeNode(min_bound, max_bound, 0, self.octree_max_depth, self.octree_max_objects)
|
||
|
||
# 将模型插入八叉树
|
||
for model in self.world.models:
|
||
self.octree.insert(model)
|
||
|
||
def detectModelCollisionsOptimized(self, specific_models=None, log_results=True):
|
||
"""使用空间分割优化的碰撞检测"""
|
||
if not self.spatial_partitioning_enabled or not self.octree:
|
||
return self.detectModelCollisions(specific_models, log_results)
|
||
|
||
start_time = time.perf_counter()
|
||
current_time = datetime.now()
|
||
|
||
models_to_check = specific_models if specific_models else self.world.models
|
||
if len(models_to_check) < 2:
|
||
return []
|
||
|
||
collision_results = []
|
||
current_collision_pairs = set()
|
||
|
||
# 使用八叉树进行优化检测
|
||
for model in models_to_check:
|
||
# 获取可能碰撞的模型列表
|
||
potential_collisions = self.octree.query(model)
|
||
|
||
for other_model in potential_collisions:
|
||
if model == other_model:
|
||
continue
|
||
|
||
pair_key = tuple(sorted([id(model), id(other_model)]))
|
||
if pair_key in current_collision_pairs:
|
||
continue
|
||
|
||
collision_info = self._checkModelPairCollision(model, other_model, current_time)
|
||
if collision_info:
|
||
collision_results.append(collision_info)
|
||
current_collision_pairs.add(pair_key)
|
||
|
||
# 执行回调
|
||
self._executeCollisionCallbacks(collision_info)
|
||
|
||
# 只有新碰撞才打印日志
|
||
if log_results and pair_key not in self.active_collisions:
|
||
self._logCollisionInfo(collision_info)
|
||
print(f"🆕 新碰撞检测到！")
|
||
|
||
# 更新活跃碰撞状态
|
||
ended_collisions = self.active_collisions - current_collision_pairs
|
||
for pair_key in ended_collisions:
|
||
print(f"✅ 碰撞结束: 模型对 {pair_key}")
|
||
|
||
self.active_collisions = current_collision_pairs
|
||
|
||
# 记录性能
|
||
detection_time = time.perf_counter() - start_time
|
||
self.performance_monitor.recordModelCollisionDetection(
|
||
detection_time, len(models_to_check), len(collision_results))
|
||
|
||
return collision_results
|
||
|
||
def cleanup(self):
|
||
"""清理碰撞管理器资源"""
|
||
self._stopCollisionDetectionTask()
|
||
self.collision_history.clear()
|
||
self.active_collisions.clear()
|
||
print("✅ 碰撞管理器已清理")
|
||
|
||
def setupAdvancedCollision(self, model, shape_type='auto', mask_type='SELECTABLE',
|
||
group_name=None, **shape_kwargs):
|
||
"""高级碰撞设置方法
|
||
|
||
Args:
|
||
model: 模型节点
|
||
shape_type: 碰撞体形状类型
|
||
mask_type: 掩码类型
|
||
group_name: 碰撞分组名称
|
||
**shape_kwargs: 形状特定参数
|
||
|
||
Returns:
|
||
collision_node_path: 碰撞节点路径
|
||
"""
|
||
# 创建碰撞节点
|
||
cNode = CollisionNode(f'collision_{model.getName()}')
|
||
|
||
# 设置掩码
|
||
if mask_type in self.MASKS:
|
||
cNode.setIntoCollideMask(self.MASKS[mask_type])
|
||
# 设置模型标签用于回调识别
|
||
model.setTag('collision_mask_type', mask_type)
|
||
|
||
# 创建碰撞体
|
||
collision_solid = self.createCollisionShape(model, shape_type, **shape_kwargs)
|
||
cNode.addSolid(collision_solid)
|
||
|
||
# 附加到模型
|
||
cNodePath = model.attachNewNode(cNode)
|
||
|
||
# 添加到分组
|
||
if group_name:
|
||
if group_name not in self.collision_groups:
|
||
self.createCollisionGroup(group_name, [mask_type])
|
||
self.addToCollisionGroup(group_name, model, cNodePath)
|
||
|
||
# 根据调试设置决定是否显示
|
||
if hasattr(self.world, 'debug_collision') and self.world.debug_collision:
|
||
cNodePath.show()
|
||
else:
|
||
cNodePath.hide()
|
||
|
||
print(f"✅ 为 {model.getName()} 设置高级碰撞 - 形状: {shape_type}, 掩码: {mask_type}")
|
||
return cNodePath
|
||
|
||
def example_usage(self):
|
||
"""碰撞系统使用示例"""
|
||
print("\n=== 碰撞系统使用示例 ===")
|
||
|
||
# 1. 创建碰撞分组
|
||
self.createCollisionGroup('characters', ['CHARACTER', 'DYNAMIC_OBJECT'])
|
||
self.createCollisionGroup('environment', ['STATIC_GEOMETRY', 'TERRAIN'])
|
||
self.createCollisionGroup('pickups', ['PICKUP_ITEM', 'INTERACTIVE'])
|
||
|
||
# 2. 注册碰撞回调
|
||
def character_pickup_callback(collision_info):
|
||
print(f"角色拾取物品: {collision_info['model_a']['name']} -> {collision_info['model_b']['name']}")
|
||
|
||
def character_environment_callback(collision_info):
|
||
print(f"角色碰撞环境: {collision_info['model_a']['name']} 碰到 {collision_info['model_b']['name']}")
|
||
|
||
self.registerCollisionCallback('CHARACTER', 'PICKUP_ITEM', character_pickup_callback)
|
||
self.registerCollisionCallback('CHARACTER', 'STATIC_GEOMETRY', character_environment_callback)
|
||
|
||
# 3. 启用空间分割优化
|
||
self.enableSpatialPartitioning(enabled=True, max_depth=6, max_objects=8)
|
||
|
||
# 4. 为模型设置不同类型的碰撞体
|
||
# model1 = ... # 假设已有模型
|
||
# self.setupAdvancedCollision(model1, 'capsule', 'CHARACTER', 'characters')
|
||
# self.setupAdvancedCollision(model2, 'box', 'PICKUP_ITEM', 'pickups')
|
||
# self.setupAdvancedCollision(model3, 'plane', 'TERRAIN', 'environment')
|
||
|
||
print("✅ 碰撞系统示例配置完成")
|
||
|
||
class CollisionPerformanceMonitor:
|
||
"""碰撞检测性能监控器"""
|
||
|
||
def __init__(self):
|
||
self.detection_times = []
|
||
self.collision_counts = []
|
||
self.model_collision_times = [] # 新增：模型间碰撞检测时间
|
||
self.model_collision_data = [] # 新增：模型间碰撞数据
|
||
self.max_records = 100
|
||
|
||
def recordDetection(self, detection_time, collision_count):
|
||
"""记录射线检测性能"""
|
||
self.detection_times.append(detection_time)
|
||
self.collision_counts.append(collision_count)
|
||
|
||
# 限制记录数量
|
||
if len(self.detection_times) > self.max_records:
|
||
self.detection_times.pop(0)
|
||
self.collision_counts.pop(0)
|
||
|
||
def recordModelCollisionDetection(self, detection_time, model_count, collision_count):
|
||
"""记录模型间碰撞检测性能"""
|
||
self.model_collision_times.append(detection_time)
|
||
self.model_collision_data.append({
|
||
'model_count': model_count,
|
||
'collision_count': collision_count,
|
||
'timestamp': time.time()
|
||
})
|
||
|
||
# 限制记录数量
|
||
if len(self.model_collision_times) > self.max_records:
|
||
self.model_collision_times.pop(0)
|
||
self.model_collision_data.pop(0)
|
||
|
||
def getAverageTime(self):
|
||
"""获取平均射线检测时间"""
|
||
if not self.detection_times:
|
||
return 0
|
||
return sum(self.detection_times) / len(self.detection_times)
|
||
|
||
def getAverageModelCollisionTime(self):
|
||
"""获取平均模型间碰撞检测时间"""
|
||
if not self.model_collision_times:
|
||
return 0
|
||
return sum(self.model_collision_times) / len(self.model_collision_times)
|
||
|
||
def getPerformanceReport(self):
|
||
"""获取完整性能报告"""
|
||
report = []
|
||
|
||
# 射线检测性能
|
||
if self.detection_times:
|
||
avg_time = self.getAverageTime()
|
||
max_time = max(self.detection_times)
|
||
avg_collisions = sum(self.collision_counts) / len(self.collision_counts)
|
||
|
||
report.append("=== 射线检测性能 ===")
|
||
report.append(f"平均检测时间: {avg_time*1000:.2f}ms")
|
||
report.append(f"最大检测时间: {max_time*1000:.2f}ms")
|
||
report.append(f"平均碰撞数量: {avg_collisions:.1f}")
|
||
report.append(f"检测次数: {len(self.detection_times)}")
|
||
|
||
# 模型间碰撞检测性能
|
||
if self.model_collision_times:
|
||
avg_model_time = self.getAverageModelCollisionTime()
|
||
max_model_time = max(self.model_collision_times)
|
||
avg_model_count = sum(d['model_count'] for d in self.model_collision_data) / len(self.model_collision_data)
|
||
total_collisions = sum(d['collision_count'] for d in self.model_collision_data)
|
||
|
||
report.append("\n=== 模型间碰撞检测性能 ===")
|
||
report.append(f"平均检测时间: {avg_model_time*1000:.2f}ms")
|
||
report.append(f"最大检测时间: {max_model_time*1000:.2f}ms")
|
||
report.append(f"平均模型数量: {avg_model_count:.1f}")
|
||
report.append(f"总碰撞次数: {total_collisions}")
|
||
report.append(f"检测次数: {len(self.model_collision_times)}")
|
||
|
||
return "\n".join(report) if report else "没有性能数据"
|
||
|
||
class OctreeNode:
|
||
"""八叉树节点"""
|
||
|
||
def __init__(self, min_bound, max_bound, depth, max_depth, max_objects):
|
||
self.min_bound = min_bound
|
||
self.max_bound = max_bound
|
||
self.depth = depth
|
||
self.max_depth = max_depth
|
||
self.max_objects = max_objects
|
||
|
||
self.objects = []
|
||
self.children = []
|
||
self.is_leaf = True
|
||
|
||
def insert(self, model):
|
||
"""插入模型到八叉树"""
|
||
if not self._contains(model):
|
||
return False
|
||
|
||
if self.is_leaf:
|
||
self.objects.append(model)
|
||
|
||
# 如果超过最大对象数且未达到最大深度，则分割
|
||
if len(self.objects) > self.max_objects and self.depth < self.max_depth:
|
||
self._subdivide()
|
||
|
||
# 重新分配对象到子节点
|
||
remaining_objects = []
|
||
for obj in self.objects:
|
||
inserted = False
|
||
for child in self.children:
|
||
if child.insert(obj):
|
||
inserted = True
|
||
break
|
||
if not inserted:
|
||
remaining_objects.append(obj)
|
||
|
||
self.objects = remaining_objects
|
||
else:
|
||
# 尝试插入到子节点
|
||
inserted = False
|
||
for child in self.children:
|
||
if child.insert(model):
|
||
inserted = True
|
||
break
|
||
|
||
if not inserted:
|
||
self.objects.append(model)
|
||
|
||
return True
|
||
|
||
def query(self, model):
|
||
"""查询可能与指定模型碰撞的模型列表"""
|
||
if not self._contains(model):
|
||
return []
|
||
|
||
result = list(self.objects)
|
||
|
||
if not self.is_leaf:
|
||
for child in self.children:
|
||
result.extend(child.query(model))
|
||
|
||
return result
|
||
|
||
def _contains(self, model):
|
||
"""检查模型是否在此节点范围内"""
|
||
bounds = model.getBounds()
|
||
if bounds.isEmpty():
|
||
return False
|
||
|
||
model_min = bounds.getMin()
|
||
model_max = bounds.getMax()
|
||
|
||
return (model_max.x >= self.min_bound.x and model_min.x <= self.max_bound.x and
|
||
model_max.y >= self.min_bound.y and model_min.y <= self.max_bound.y and
|
||
model_max.z >= self.min_bound.z and model_min.z <= self.max_bound.z)
|
||
|
||
def _subdivide(self):
|
||
"""分割节点为8个子节点"""
|
||
center = Point3(
|
||
(self.min_bound.x + self.max_bound.x) * 0.5,
|
||
(self.min_bound.y + self.max_bound.y) * 0.5,
|
||
(self.min_bound.z + self.max_bound.z) * 0.5
|
||
)
|
||
|
||
# 创建8个子节点
|
||
subdivisions = [
|
||
(self.min_bound, center),
|
||
(Point3(center.x, self.min_bound.y, self.min_bound.z), Point3(self.max_bound.x, center.y, center.z)),
|
||
(Point3(self.min_bound.x, center.y, self.min_bound.z), Point3(center.x, self.max_bound.y, center.z)),
|
||
(Point3(center.x, center.y, self.min_bound.z), Point3(self.max_bound.x, self.max_bound.y, center.z)),
|
||
(Point3(self.min_bound.x, self.min_bound.y, center.z), Point3(center.x, center.y, self.max_bound.z)),
|
||
(Point3(center.x, self.min_bound.y, center.z), Point3(self.max_bound.x, center.y, self.max_bound.z)),
|
||
(Point3(self.min_bound.x, center.y, center.z), Point3(center.x, self.max_bound.y, self.max_bound.z)),
|
||
(center, self.max_bound)
|
||
]
|
||
|
||
for min_b, max_b in subdivisions:
|
||
child = OctreeNode(min_b, max_b, self.depth + 1, self.max_depth, self.max_objects)
|
||
self.children.append(child)
|
||
|
||
self.is_leaf = False
|